Add setup_async.py and test_setup_async.py

androidtv/setup_async.py

@@ -0,0 +1,74 @@
+"""Connect to a device and determine whether it's an Android TV or an Amazon Fire TV.
+
+ADB Debugging must be enabled.
+"""
+
+from .androidtv.androidtv_async import AndroidTVAsync
+from .basetv.basetv_async import BaseTVAsync
+from .constants import DEFAULT_AUTH_TIMEOUT_S
+from .firetv.firetv_async import FireTVAsync
+
+
+__version__ = '0.0.4'
+
+
+async def setup(host, port=5555, adbkey='', adb_server_ip='', adb_server_port=5037, state_detection_rules=None, device_class='auto', auth_timeout_s=DEFAULT_AUTH_TIMEOUT_S):
+    """Connect to a device and determine whether it's an Android TV or an Amazon Fire TV.
+
+    Parameters
+    ----------
+    host : str
+        The address of the device; may be an IP address or a host name
+    port : int
+        The device port to which we are connecting (default is 5555)
+    adbkey : str
+        The path to the ``adbkey`` file for ADB authentication
+    adb_server_ip : str
+        The IP address of the ADB server
+    adb_server_port : int
+        The port for the ADB server
+    state_detection_rules : dict, None
+        A dictionary of rules for determining the state (see :class:`~aio_androidtv.basetv.BaseTV`)
+    device_class : str
+        The type of device: ``'auto'`` (detect whether it is an Android TV or Fire TV device), ``'androidtv'``, or ``'firetv'```
+    auth_timeout_s : float
+        Authentication timeout (in seconds)
+
+    Returns
+    -------
+    aftv : AndroidTVAsync, FireTVAsync
+        The representation of the device
+
+    """
+    if device_class == 'androidtv':
+        atv = AndroidTVAsync(host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules)
+        await atv.adb_connect(auth_timeout_s=auth_timeout_s)
+        atv.device_properties = await atv.get_device_properties()
+        return atv
+
+    if device_class == 'firetv':
+        ftv = FireTVAsync(host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules)
+        await ftv.adb_connect(auth_timeout_s=auth_timeout_s)
+        ftv.device_properties = await ftv.get_device_properties()
+        return ftv
+
+    if device_class != 'auto':
+        raise ValueError("`device_class` must be 'androidtv', 'firetv', or 'auto'.")
+
+    aftv = BaseTVAsync(host, port, adbkey, adb_server_ip, adb_server_port, state_detection_rules)
+
+    # establish the ADB connection
+    await aftv.adb_connect(auth_timeout_s=auth_timeout_s)
+
+    # get device properties
+    aftv.device_properties = await aftv.get_device_properties()
+
+    # Fire TV
+    if aftv.device_properties.get('manufacturer') == 'Amazon':
+        aftv.__class__ = FireTVAsync
+
+    # Android TV
+    else:
+        aftv.__class__ = AndroidTVAsync
+
+    return aftv

tests/test_setup_async.py

@@ -0,0 +1,67 @@
+import sys
+import unittest
+from unittest.mock import patch
+
+
+sys.path.insert(0, '..')
+
+from androidtv.setup_async import setup
+from androidtv.androidtv.androidtv_async import AndroidTVAsync
+from androidtv.firetv.firetv_async import FireTVAsync
+
+from . import async_patchers
+from .async_wrapper import awaiter
+
+
+DEVICE_PROPERTIES_OUTPUT1 = "Amazon\n\n\n\n\n."
+
+DEVICE_PROPERTIES_DICT1 = {'manufacturer': 'Amazon',
+                           'model': '',
+                           'serialno': None,
+                           'sw_version': '',
+                           'wifimac': None,
+                           'ethmac': None}
+
+DEVICE_PROPERTIES_OUTPUT2 = "Not Amazon\n\n\n\n\n."
+
+DEVICE_PROPERTIES_DICT2 = {'manufacturer': 'Not Amazon',
+                           'model': '',
+                           'serialno': None,
+                           'sw_version': '',
+                           'wifimac': None,
+                           'ethmac': None}
+
+
+class TestSetup(unittest.TestCase):
+    PATCH_KEY = 'python'
+
+    @awaiter
+    async def test_setup(self):
+        """Test that the ``setup`` function works correctly.
+        """
+        with self.assertRaises(ValueError):
+            await setup('HOST', 5555, device_class='INVALID')
+
+        with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT1)[self.PATCH_KEY]:
+            ftv = await setup('HOST', 5555)
+            self.assertIsInstance(ftv, FireTVAsync)
+            self.assertDictEqual(ftv.device_properties, DEVICE_PROPERTIES_DICT1)
+
+        with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT2)[self.PATCH_KEY]:
+            atv = await setup('HOST', 5555)
+            self.assertIsInstance(atv, AndroidTVAsync)
+            self.assertDictEqual(atv.device_properties, DEVICE_PROPERTIES_DICT2)
+
+        with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT1)[self.PATCH_KEY]:
+            ftv = await setup('HOST', 5555, device_class='androidtv')
+            self.assertIsInstance(ftv, AndroidTVAsync)
+            self.assertDictEqual(ftv.device_properties, DEVICE_PROPERTIES_DICT1)
+
+        with async_patchers.PATCH_ADB_DEVICE_TCP, async_patchers.patch_connect(True)[self.PATCH_KEY], async_patchers.patch_shell(DEVICE_PROPERTIES_OUTPUT2)[self.PATCH_KEY]:
+            atv = await setup('HOST', 5555, device_class='firetv')
+            self.assertIsInstance(atv, FireTVAsync)
+            self.assertDictEqual(atv.device_properties, DEVICE_PROPERTIES_DICT2)
+
+
+if __name__ == "__main__":
+    unittest.main()