Add connect tests

tests/const.py

@@ -0,0 +1,12 @@
+"""Constants for tests."""
+
+with open("tests/fixtures/event_ms_channel_connect.json") as file:
+    MS_CHANNEL_CONNECT_SAMPLE = file.read()
+with open("tests/fixtures/event_ed_edentv_update.json") as file:
+    ED_EDENTV_UPDATE_SAMPLE = file.read()
+with open("tests/fixtures/event_ed_apps_launch.json") as file:
+    ED_APPS_LAUNCH_SAMPLE = file.read()
+with open("tests/fixtures/event_ed_installedApp_get.json") as file:
+    ED_INSTALLED_APP_SAMPLE = file.read()
+with open("tests/fixtures/event_ms_error.json") as file:
+    MS_ERROR_SAMPLE = file.read()

tests/fixtures/event_ed_apps_launch.json

@@ -0,0 +1,5 @@
+{
+  "data": 200,
+  "event": "ed.apps.launch",
+  "from": "host"
+}

tests/fixtures/event_ed_edentv_update.json

@@ -0,0 +1,6 @@
+{
+  "data": {
+    "update_type": "ed.edenApp.update"
+  },
+  "event": "ed.edenTV.update"
+}

tests/fixtures/event_ms_channel_connect.json

@@ -0,0 +1,7 @@
+{
+  "data": {
+    "token": 123456789
+  },
+  "event": "ms.channel.connect",
+  "from": "host"
+}

tests/test_art.py

@@ -4,15 +4,19 @@
 from samsungtvws.art import SamsungTVArt
 from samsungtvws.remote import SamsungTVWS
 
+from .const import MS_CHANNEL_CONNECT_SAMPLE
+
 
 def test_create_connection_from_remote() -> None:
     connection = Mock()
     with patch(
         "samsungtvws.connection.websocket.create_connection"
     ) as connection_class:
         connection_class.return_value = connection
-        open_response = '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-        connection.recv.side_effect = [open_response, open_response]
+        connection.recv.side_effect = [
+            MS_CHANNEL_CONNECT_SAMPLE,
+            MS_CHANNEL_CONNECT_SAMPLE,
+        ]
 
         tv_art = SamsungTVWS("127.0.0.1").art()
         tv_art.set_artmode("test")
@@ -31,8 +35,10 @@ def test_create_connection_direct() -> None:
         "samsungtvws.connection.websocket.create_connection"
     ) as connection_class:
         connection_class.return_value = connection
-        open_response = '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-        connection.recv.side_effect = [open_response, open_response]
+        connection.recv.side_effect = [
+            MS_CHANNEL_CONNECT_SAMPLE,
+            MS_CHANNEL_CONNECT_SAMPLE,
+        ]
 
         tv_art = SamsungTVArt("127.0.0.1")
         tv_art.set_artmode("test")
@@ -51,9 +57,10 @@ def test_set_artmode(connection: Mock) -> None:
         "samsungtvws.art.uuid.uuid4",
         return_value="07e72228-7110-4655-aaa6-d81b5188c219",
     ):
-        open_response = '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-
-        connection.recv.side_effect = [open_response, open_response]
+        connection.recv.side_effect = [
+            MS_CHANNEL_CONNECT_SAMPLE,
+            MS_CHANNEL_CONNECT_SAMPLE,
+        ]
         tv_art = SamsungTVArt("127.0.0.1")
         tv_art.set_artmode("test")
 

tests/test_async_remote.py

@@ -5,23 +5,57 @@
 import pytest
 
 from samsungtvws.async_remote import SamsungTVWSAsyncRemote
+from samsungtvws.exceptions import ConnectionFailure
 from samsungtvws.remote import SendRemoteKey
 
+from .const import (
+    ED_APPS_LAUNCH_SAMPLE,
+    ED_EDENTV_UPDATE_SAMPLE,
+    ED_INSTALLED_APP_SAMPLE,
+    MS_CHANNEL_CONNECT_SAMPLE,
+    MS_ERROR_SAMPLE,
+)
+
+
+def create_future_with_result(result) -> asyncio.Future:
+    future = asyncio.Future()
+    future.set_result(result)
+    return future
+
+
+ED_APPS_LAUNCH_FUTURE = create_future_with_result(ED_APPS_LAUNCH_SAMPLE)
+ED_EDENTV_UPDATE_FUTURE = create_future_with_result(ED_EDENTV_UPDATE_SAMPLE)
+ED_INSTALLED_APP_FUTURE = create_future_with_result(ED_INSTALLED_APP_SAMPLE)
+MS_CHANNEL_CONNECT_FUTURE = create_future_with_result(MS_CHANNEL_CONNECT_SAMPLE)
+MS_ERROR_EVENT_FUTURE = create_future_with_result(MS_ERROR_SAMPLE)
+NONE_FUTURE = create_future_with_result(None)
+
 
 @pytest.mark.asyncio
-async def test_send_key(async_connection: Mock) -> None:
+async def test_connect(async_connection: Mock) -> None:
     """Ensure simple data can be parsed."""
-    open_response = (
-        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-    )
-    open_response_future = asyncio.Future()
-    open_response_future.set_result(open_response)
+    async_connection.recv = Mock(side_effect=[MS_CHANNEL_CONNECT_FUTURE])
+    async_connection.send = Mock(return_value=NONE_FUTURE)
+    tv = SamsungTVWSAsyncRemote("127.0.0.1")
+    await tv.start_listening()
+    assert tv.token == 123456789
+
+
+@pytest.mark.asyncio
+async def test_connection_failure(async_connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    async_connection.recv = Mock(side_effect=[MS_ERROR_EVENT_FUTURE])
+    async_connection.send = Mock(return_value=NONE_FUTURE)
+    tv = SamsungTVWSAsyncRemote("127.0.0.1")
+    with pytest.raises(ConnectionFailure):
+        await tv.start_listening()
 
-    send_command_future = asyncio.Future()
-    send_command_future.set_result(None)
 
-    async_connection.recv = Mock(side_effect=[open_response_future])
-    async_connection.send = Mock(return_value=send_command_future)
+@pytest.mark.asyncio
+async def test_send_key(async_connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    async_connection.recv = Mock(side_effect=[MS_CHANNEL_CONNECT_FUTURE])
+    async_connection.send = Mock(return_value=NONE_FUTURE)
     tv = SamsungTVWSAsyncRemote("127.0.0.1")
     await tv.send_command(SendRemoteKey.click("KEY_POWER"))
     async_connection.send.assert_called_once_with(
@@ -35,26 +69,73 @@ async def test_send_key(async_connection: Mock) -> None:
 
 
 @pytest.mark.asyncio
-async def test_send_hold_key(async_connection: Mock) -> None:
-    """Ensure simple data can be parsed."""
-    open_response = (
-        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
+async def test_app_list(async_connection: Mock) -> None:
+    """Ensure valid app_list data can be parsed."""
+    async_connection.recv = Mock(
+        side_effect=[
+            MS_CHANNEL_CONNECT_FUTURE,
+            ED_INSTALLED_APP_FUTURE,
+        ]
     )
-    open_response_future = asyncio.Future()
-    open_response_future.set_result(open_response)
+    tv = SamsungTVWSAsyncRemote("127.0.0.1")
+    await tv.start_listening()
+    assert await tv.app_list() == [
+        {
+            "appId": "111299001912",
+            "app_type": 2,
+            "icon": "/opt/share/webappservice/apps_icon/FirstScreen/111299001912/250x250.png",
+            "is_lock": 0,
+            "name": "YouTube",
+        },
+        {
+            "appId": "3201608010191",
+            "app_type": 2,
+            "icon": "/opt/share/webappservice/apps_icon/FirstScreen/3201608010191/250x250.png",
+            "is_lock": 0,
+            "name": "Deezer",
+        },
+    ]
 
-    send_command_future = asyncio.Future()
-    send_command_future.set_result(None)
 
-    async_connection.recv = Mock(side_effect=[open_response_future])
-    async_connection.send = Mock(return_value=send_command_future)
+@pytest.mark.asyncio
+async def test_app_list_bad_order(async_connection: Mock) -> None:
+    """Ensure valid app_list data can be parsed, even if we get events in the wrong order."""
+    async_connection.recv = Mock(
+        side_effect=[
+            MS_CHANNEL_CONNECT_FUTURE,
+            ED_APPS_LAUNCH_FUTURE,
+            ED_INSTALLED_APP_FUTURE,
+        ]
+    )
+    tv = SamsungTVWSAsyncRemote("127.0.0.1")
+    await tv.start_listening()
+    assert await tv.app_list() == [
+        {
+            "appId": "111299001912",
+            "app_type": 2,
+            "icon": "/opt/share/webappservice/apps_icon/FirstScreen/111299001912/250x250.png",
+            "is_lock": 0,
+            "name": "YouTube",
+        },
+        {
+            "appId": "3201608010191",
+            "app_type": 2,
+            "icon": "/opt/share/webappservice/apps_icon/FirstScreen/3201608010191/250x250.png",
+            "is_lock": 0,
+            "name": "Deezer",
+        },
+    ]
+
 
-    sleep_future = asyncio.Future()
-    sleep_future.set_result(None)
+@pytest.mark.asyncio
+async def test_send_hold_key(async_connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    async_connection.recv = Mock(side_effect=[MS_CHANNEL_CONNECT_FUTURE])
+    async_connection.send = Mock(return_value=NONE_FUTURE)
 
     tv = SamsungTVWSAsyncRemote("127.0.0.1")
     with patch(
-        "samsungtvws.async_connection.asyncio.sleep", return_value=sleep_future
+        "samsungtvws.async_connection.asyncio.sleep", return_value=NONE_FUTURE
     ) as patch_sleep:
         await tv.send_command(SendRemoteKey.hold_key("KEY_POWER", 3))
 

tests/test_event.py

@@ -3,10 +3,11 @@
 from samsungtvws import event
 from samsungtvws.exceptions import MessageError
 
+from .const import ED_INSTALLED_APP_SAMPLE, MS_ERROR_SAMPLE
+
 
 def test_parse_installed_app() -> None:
-    with open("tests/fixtures/event_ed_installedApp_get.json") as file:
-        json_response = json.load(file)
+    json_response = json.loads(ED_INSTALLED_APP_SAMPLE)
     assert event.parse_installed_app(json_response) == [
         {
             "appId": "111299001912",
@@ -26,8 +27,7 @@ def test_parse_installed_app() -> None:
 
 
 def test_parse_ms_error() -> None:
-    with open("tests/fixtures/event_ms_error.json") as file:
-        json_response = json.load(file)
+    json_response = json.loads(MS_ERROR_SAMPLE)
     error = event.parse_ms_error(json_response)
     assert isinstance(error, MessageError)
     assert str(error) == "unrecognized method value : ms.application.stop"

tests/test_helper.py

@@ -1,10 +1,10 @@
 """Tests for helper module."""
 from samsungtvws.helper import process_api_response
 
+from .const import ED_APPS_LAUNCH_SAMPLE
+
 
 def test_data_simple() -> None:
     """Ensure simple data can be parsed."""
-    data = '{"data": 200, "event": "ed.apps.launch", "from": "host"}'
-
-    parsed_response = process_api_response(data)
+    parsed_response = process_api_response(ED_APPS_LAUNCH_SAMPLE)
     assert parsed_response == {"data": 200, "event": "ed.apps.launch", "from": "host"}

tests/test_remote.py

@@ -1,16 +1,38 @@
 """Tests for remote module."""
 from unittest.mock import Mock, call, patch
 
+import pytest
+
+from samsungtvws.exceptions import ConnectionFailure
 from samsungtvws.remote import SamsungTVWS
 
+from .const import (
+    ED_APPS_LAUNCH_SAMPLE,
+    ED_INSTALLED_APP_SAMPLE,
+    MS_CHANNEL_CONNECT_SAMPLE,
+    MS_ERROR_SAMPLE,
+)
 
-def test_send_key(connection: Mock) -> None:
+
+def test_connect(connection: Mock) -> None:
     """Ensure simple data can be parsed."""
-    open_response = (
-        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-    )
+    connection.recv = Mock(side_effect=[MS_CHANNEL_CONNECT_SAMPLE])
+    tv = SamsungTVWS("127.0.0.1")
+    tv.open()
+    assert tv.token == 123456789
 
-    connection.recv.side_effect = [open_response]
+
+def test_connection_failure(connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    connection.recv = Mock(side_effect=[MS_ERROR_SAMPLE])
+    tv = SamsungTVWS("127.0.0.1")
+    with pytest.raises(ConnectionFailure):
+        tv.open()
+
+
+def test_send_key(connection: Mock) -> None:
+    """Ensure simple data can be parsed."""
+    connection.recv.side_effect = [MS_CHANNEL_CONNECT_SAMPLE]
     tv = SamsungTVWS("127.0.0.1")
     tv.send_key("KEY_POWER")
     connection.send.assert_called_once_with(
@@ -25,12 +47,10 @@ def test_send_key(connection: Mock) -> None:
 
 def test_app_list(connection: Mock) -> None:
     """Ensure valid app_list data can be parsed."""
-    open_response = (
-        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-    )
-    app_list_response = '{"data":{"data":[{"appId":"111299001912","app_type":2,"icon":"/opt/share/webappservice/apps_icon/FirstScreen/111299001912/250x250.png","is_lock":0,"name":"YouTube"},{"appId":"3201608010191","app_type":2,"icon":"/opt/share/webappservice/apps_icon/FirstScreen/3201608010191/250x250.png","is_lock":0,"name":"Deezer"}]},"event":"ed.installedApp.get","from":"host"}'
-
-    connection.recv.side_effect = [open_response, app_list_response]
+    connection.recv.side_effect = [
+        MS_CHANNEL_CONNECT_SAMPLE,
+        ED_INSTALLED_APP_SAMPLE,
+    ]
     tv = SamsungTVWS("127.0.0.1")
     assert tv.app_list() == [
         {
@@ -52,12 +72,10 @@ def test_app_list(connection: Mock) -> None:
 
 def test_app_list_invalid(connection: Mock) -> None:
     """Ensure simple data can be parsed."""
-    open_response = (
-        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-    )
-    app_list_response = '{"data": 200, "event": "ed.apps.launch", "from": "host"}'
-
-    connection.recv.side_effect = [open_response, app_list_response]
+    connection.recv.side_effect = [
+        MS_CHANNEL_CONNECT_SAMPLE,
+        ED_APPS_LAUNCH_SAMPLE,
+    ]
     tv = SamsungTVWS("127.0.0.1")
     assert tv.app_list() is None
     connection.send.assert_called_once_with(
@@ -67,10 +85,7 @@ def test_app_list_invalid(connection: Mock) -> None:
 
 def test_send_hold_key(connection: Mock) -> None:
     """Ensure simple data can be parsed."""
-    open_response = (
-        '{"data": {"token": 123456789}, "event": "ms.channel.connect", "from": "host"}'
-    )
-    connection.recv.side_effect = [open_response]
+    connection.recv.side_effect = [MS_CHANNEL_CONNECT_SAMPLE]
 
     tv = SamsungTVWS("127.0.0.1")
     with patch("samsungtvws.connection.time.sleep") as patch_sleep: