From 85473d2c989fe45cad0a2b3989604e965ab697ac Mon Sep 17 00:00:00 2001 From: Jeff Irion Date: Mon, 2 Sep 2019 13:08:01 -0700 Subject: [PATCH] Bump androidtv to 0.0.26 and update tests (#26340) * Move the patchers to a separate file * Got a pytest test working (mostly) * Checkpoint * Switch to pytest for all tests * Bump androidtv to 0.0.26 and update tests * More robust patching * Remove unused constants * Combine two lines * Add 2 additional checks * Check that state objects are not None; add more description to tests * Use f strings --- .../components/androidtv/manifest.json | 2 +- requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- tests/components/androidtv/patchers.py | 129 ++++++ .../components/androidtv/test_media_player.py | 407 +++++++++--------- 5 files changed, 344 insertions(+), 198 deletions(-) create mode 100644 tests/components/androidtv/patchers.py diff --git a/homeassistant/components/androidtv/manifest.json b/homeassistant/components/androidtv/manifest.json index 91ea4019c05..2c8fc98e248 100644 --- a/homeassistant/components/androidtv/manifest.json +++ b/homeassistant/components/androidtv/manifest.json @@ -3,7 +3,7 @@ "name": "Androidtv", "documentation": "https://www.home-assistant.io/components/androidtv", "requirements": [ - "androidtv==0.0.25" + "androidtv==0.0.26" ], "dependencies": [], "codeowners": ["@JeffLIrion"] diff --git a/requirements_all.txt b/requirements_all.txt index a6d8b0f414f..cda3ed77bf0 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -194,7 +194,7 @@ ambiclimate==0.2.1 amcrest==1.5.3 # homeassistant.components.androidtv -androidtv==0.0.25 +androidtv==0.0.26 # homeassistant.components.anel_pwrctrl anel_pwrctrl-homeassistant==0.0.1.dev2 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index c76473cd328..37e45278e33 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -79,7 +79,7 @@ aiowwlln==1.0.0 ambiclimate==0.2.1 # homeassistant.components.androidtv -androidtv==0.0.25 +androidtv==0.0.26 # homeassistant.components.apns apns2==0.3.0 diff --git a/tests/components/androidtv/patchers.py b/tests/components/androidtv/patchers.py new file mode 100644 index 00000000000..86d1c1c15bd --- /dev/null +++ b/tests/components/androidtv/patchers.py @@ -0,0 +1,129 @@ +"""Define patches used for androidtv tests.""" + +from socket import error as socket_error +from unittest.mock import patch + + +class AdbCommandsFake: + """A fake of the `adb.adb_commands.AdbCommands` class.""" + + def ConnectDevice(self, *args, **kwargs): # pylint: disable=invalid-name + """Try to connect to a device.""" + raise NotImplementedError + + def Shell(self, cmd): # pylint: disable=invalid-name + """Send an ADB shell command.""" + raise NotImplementedError + + +class AdbCommandsFakeSuccess(AdbCommandsFake): + """A fake of the `adb.adb_commands.AdbCommands` class when the connection attempt succeeds.""" + + def ConnectDevice(self, *args, **kwargs): # pylint: disable=invalid-name + """Successfully connect to a device.""" + return self + + +class AdbCommandsFakeFail(AdbCommandsFake): + """A fake of the `adb.adb_commands.AdbCommands` class when the connection attempt fails.""" + + def ConnectDevice( + self, *args, **kwargs + ): # pylint: disable=invalid-name, no-self-use + """Fail to connect to a device.""" + raise socket_error + + +class ClientFakeSuccess: + """A fake of the `adb_messenger.client.Client` class when the connection and shell commands succeed.""" + + def __init__(self, host="127.0.0.1", port=5037): + """Initialize a `ClientFakeSuccess` instance.""" + self._devices = [] + + def devices(self): + """Get a list of the connected devices.""" + return self._devices + + def device(self, serial): + """Mock the `Client.device` method when the device is connected via ADB.""" + device = DeviceFake(serial) + self._devices.append(device) + return device + + +class ClientFakeFail: + """A fake of the `adb_messenger.client.Client` class when the connection and shell commands fail.""" + + def __init__(self, host="127.0.0.1", port=5037): + """Initialize a `ClientFakeFail` instance.""" + self._devices = [] + + def devices(self): + """Get a list of the connected devices.""" + return self._devices + + def device(self, serial): + """Mock the `Client.device` method when the device is not connected via ADB.""" + self._devices = [] + + +class DeviceFake: + """A fake of the `adb_messenger.device.Device` class.""" + + def __init__(self, host): + """Initialize a `DeviceFake` instance.""" + self.host = host + + def get_serial_no(self): + """Get the serial number for the device (IP:PORT).""" + return self.host + + def shell(self, cmd): + """Send an ADB shell command.""" + raise NotImplementedError + + +def patch_connect(success): + """Mock the `adb.adb_commands.AdbCommands` and `adb_messenger.client.Client` classes.""" + + if success: + return { + "python": patch( + "androidtv.adb_manager.AdbCommands", AdbCommandsFakeSuccess + ), + "server": patch("androidtv.adb_manager.Client", ClientFakeSuccess), + } + return { + "python": patch("androidtv.adb_manager.AdbCommands", AdbCommandsFakeFail), + "server": patch("androidtv.adb_manager.Client", ClientFakeFail), + } + + +def patch_shell(response=None, error=False): + """Mock the `AdbCommandsFake.Shell` and `DeviceFake.shell` methods.""" + + def shell_success(self, cmd): + """Mock the `AdbCommandsFake.Shell` and `DeviceFake.shell` methods when they are successful.""" + self.shell_cmd = cmd + return response + + def shell_fail_python(self, cmd): + """Mock the `AdbCommandsFake.Shell` method when it fails.""" + self.shell_cmd = cmd + raise AttributeError + + def shell_fail_server(self, cmd): + """Mock the `DeviceFake.shell` method when it fails.""" + self.shell_cmd = cmd + raise ConnectionResetError + + if not error: + return { + "python": patch(f"{__name__}.AdbCommandsFake.Shell", shell_success), + "server": patch(f"{__name__}.DeviceFake.shell", shell_success), + } + return { + "python": patch(f"{__name__}.AdbCommandsFake.Shell", shell_fail_python), + "server": patch(f"{__name__}.DeviceFake.shell", shell_fail_server), + } diff --git a/tests/components/androidtv/test_media_player.py b/tests/components/androidtv/test_media_player.py index e787fddd3bc..39b392c97ee 100644 --- a/tests/components/androidtv/test_media_player.py +++ b/tests/components/androidtv/test_media_player.py @@ -1,232 +1,249 @@ """The tests for the androidtv platform.""" import logging -from socket import error as socket_error -import unittest -from unittest.mock import patch +from homeassistant.setup import async_setup_component from homeassistant.components.androidtv.media_player import ( - AndroidTVDevice, - FireTVDevice, - setup, + ANDROIDTV_DOMAIN, + CONF_ADB_SERVER_IP, +) +from homeassistant.components.media_player.const import DOMAIN +from homeassistant.const import ( + CONF_DEVICE_CLASS, + CONF_HOST, + CONF_NAME, + CONF_PLATFORM, + STATE_IDLE, + STATE_OFF, + STATE_UNAVAILABLE, ) - -def connect_device_success(self, *args, **kwargs): - """Return `self`, which will result in the ADB connection being interpreted as available.""" - return self +from . import patchers -def connect_device_fail(self, *args, **kwargs): - """Raise a socket error.""" - raise socket_error +# Android TV device with Python ADB implementation +CONFIG_ANDROIDTV_PYTHON_ADB = { + DOMAIN: { + CONF_PLATFORM: ANDROIDTV_DOMAIN, + CONF_HOST: "127.0.0.1", + CONF_NAME: "Android TV", + } +} + +# Android TV device with ADB server +CONFIG_ANDROIDTV_ADB_SERVER = { + DOMAIN: { + CONF_PLATFORM: ANDROIDTV_DOMAIN, + CONF_HOST: "127.0.0.1", + CONF_NAME: "Android TV", + CONF_ADB_SERVER_IP: "127.0.0.1", + } +} + +# Fire TV device with Python ADB implementation +CONFIG_FIRETV_PYTHON_ADB = { + DOMAIN: { + CONF_PLATFORM: ANDROIDTV_DOMAIN, + CONF_HOST: "127.0.0.1", + CONF_NAME: "Fire TV", + CONF_DEVICE_CLASS: "firetv", + } +} + +# Fire TV device with ADB server +CONFIG_FIRETV_ADB_SERVER = { + DOMAIN: { + CONF_PLATFORM: ANDROIDTV_DOMAIN, + CONF_HOST: "127.0.0.1", + CONF_NAME: "Fire TV", + CONF_DEVICE_CLASS: "firetv", + CONF_ADB_SERVER_IP: "127.0.0.1", + } +} -def adb_shell_python_adb_error(self, cmd): - """Raise an error that is among those caught for the Python ADB implementation.""" - raise AttributeError +async def _test_reconnect(hass, caplog, config): + """Test that the error and reconnection attempts are logged correctly. + "Handles device/service unavailable. Log a warning once when + unavailable, log once when reconnected." -def adb_shell_adb_server_error(self, cmd): - """Raise an error that is among those caught for the ADB server implementation.""" - raise ConnectionResetError + https://developers.home-assistant.io/docs/en/integration_quality_scale_index.html + """ + if CONF_ADB_SERVER_IP not in config[DOMAIN]: + patch_key = "python" + else: + patch_key = "server" + if config[DOMAIN].get(CONF_DEVICE_CLASS) != "firetv": + entity_id = "media_player.android_tv" + else: + entity_id = "media_player.fire_tv" -class AdbAvailable: - """A class that indicates the ADB connection is available.""" + with patchers.patch_connect(True)[patch_key], patchers.patch_shell("")[patch_key]: + assert await async_setup_component(hass, DOMAIN, config) + await hass.helpers.entity_component.async_update_entity(entity_id) + state = hass.states.get(entity_id) + assert state is not None + assert state.state == STATE_OFF - def shell(self, cmd): - """Send an ADB shell command (ADB server implementation).""" - return "" + caplog.clear() + caplog.set_level(logging.WARNING) + with patchers.patch_connect(False)[patch_key], patchers.patch_shell(error=True)[ + patch_key + ]: + for _ in range(5): + await hass.helpers.entity_component.async_update_entity(entity_id) + state = hass.states.get(entity_id) + assert state is not None + assert state.state == STATE_UNAVAILABLE -class AdbUnavailable: - """A class with ADB shell methods that raise errors.""" + assert len(caplog.record_tuples) == 2 + assert caplog.record_tuples[0][1] == logging.ERROR + assert caplog.record_tuples[1][1] == logging.WARNING - def __bool__(self): - """Return `False` to indicate that the ADB connection is unavailable.""" - return False + caplog.set_level(logging.DEBUG) + with patchers.patch_connect(True)[patch_key], patchers.patch_shell("1")[patch_key]: + # Update 1 will reconnect + await hass.helpers.entity_component.async_update_entity(entity_id) - def shell(self, cmd): - """Raise an error that pertains to the Python ADB implementation.""" - raise ConnectionResetError + # If using an ADB server, the state will get updated; otherwise, the + # state will be the last known state + state = hass.states.get(entity_id) + if patch_key == "server": + assert state.state == STATE_IDLE + else: + assert state.state == STATE_OFF + # Update 2 will update the state, regardless of which ADB connection + # method is used + await hass.helpers.entity_component.async_update_entity(entity_id) + state = hass.states.get(entity_id) + assert state is not None + assert state.state == STATE_IDLE -PATCH_PYTHON_ADB_CONNECT_SUCCESS = patch( - "adb.adb_commands.AdbCommands.ConnectDevice", connect_device_success -) -PATCH_PYTHON_ADB_COMMAND_SUCCESS = patch( - "adb.adb_commands.AdbCommands.Shell", return_value="" -) -PATCH_PYTHON_ADB_CONNECT_FAIL = patch( - "adb.adb_commands.AdbCommands.ConnectDevice", connect_device_fail -) -PATCH_PYTHON_ADB_COMMAND_FAIL = patch( - "adb.adb_commands.AdbCommands.Shell", adb_shell_python_adb_error -) -PATCH_PYTHON_ADB_COMMAND_NONE = patch( - "adb.adb_commands.AdbCommands.Shell", return_value=None -) - -PATCH_ADB_SERVER_CONNECT_SUCCESS = patch( - "adb_messenger.client.Client.device", return_value=AdbAvailable() -) -PATCH_ADB_SERVER_AVAILABLE = patch( - "androidtv.basetv.BaseTV.available", return_value=True -) -PATCH_ADB_SERVER_CONNECT_FAIL = patch( - "adb_messenger.client.Client.device", return_value=AdbUnavailable() -) -PATCH_ADB_SERVER_COMMAND_FAIL = patch( - "{}.AdbAvailable.shell".format(__name__), adb_shell_adb_server_error -) -PATCH_ADB_SERVER_COMMAND_NONE = patch( - "{}.AdbAvailable.shell".format(__name__), return_value=None -) - - -class TestAndroidTVPythonImplementation(unittest.TestCase): - """Test the androidtv media player for an Android TV device.""" - - def setUp(self): - """Set up an `AndroidTVDevice` media player.""" - with PATCH_PYTHON_ADB_CONNECT_SUCCESS, PATCH_PYTHON_ADB_COMMAND_SUCCESS: - aftv = setup("IP:PORT", device_class="androidtv") - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, None, None) - - def test_reconnect(self): - """Test that the error and reconnection attempts are logged correctly. - - "Handles device/service unavailable. Log a warning once when - unavailable, log once when reconnected." - - https://developers.home-assistant.io/docs/en/integration_quality_scale_index.html - """ - with self.assertLogs(level=logging.WARNING) as logs: - with PATCH_PYTHON_ADB_CONNECT_FAIL, PATCH_PYTHON_ADB_COMMAND_FAIL: - for _ in range(5): - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) - - assert len(logs.output) == 2 - assert logs.output[0].startswith("ERROR") - assert logs.output[1].startswith("WARNING") - - with self.assertLogs(level=logging.DEBUG) as logs: - with PATCH_PYTHON_ADB_CONNECT_SUCCESS, PATCH_PYTHON_ADB_COMMAND_SUCCESS: - # Update 1 will reconnect - self.aftv.update() - self.assertTrue(self.aftv.available) - - # Update 2 will update the state - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) - + if patch_key == "python": assert ( - "ADB connection to {} successfully established".format(self.aftv.aftv.host) - in logs.output[0] + "ADB connection to 127.0.0.1:5555 successfully established" + in caplog.record_tuples[2] + ) + else: + assert ( + "ADB connection to 127.0.0.1:5555 via ADB server 127.0.0.1:5037 successfully established" + in caplog.record_tuples[2] ) - def test_adb_shell_returns_none(self): - """Test the case that the ADB shell command returns `None`. - - The state should be `None` and the device should be unavailable. - """ - with PATCH_PYTHON_ADB_COMMAND_NONE: - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) - - with PATCH_PYTHON_ADB_CONNECT_SUCCESS, PATCH_PYTHON_ADB_COMMAND_SUCCESS: - # Update 1 will reconnect - self.aftv.update() - self.assertTrue(self.aftv.available) - - # Update 2 will update the state - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) + return True -class TestAndroidTVServerImplementation(unittest.TestCase): - """Test the androidtv media player for an Android TV device.""" +async def _test_adb_shell_returns_none(hass, config): + """Test the case that the ADB shell command returns `None`. - def setUp(self): - """Set up an `AndroidTVDevice` media player.""" - with PATCH_ADB_SERVER_CONNECT_SUCCESS, PATCH_ADB_SERVER_AVAILABLE: - aftv = setup( - "IP:PORT", adb_server_ip="ADB_SERVER_IP", device_class="androidtv" - ) - self.aftv = AndroidTVDevice(aftv, "Fake Android TV", {}, None, None) + The state should be `None` and the device should be unavailable. + """ + if CONF_ADB_SERVER_IP not in config[DOMAIN]: + patch_key = "python" + else: + patch_key = "server" - def test_reconnect(self): - """Test that the error and reconnection attempts are logged correctly. + if config[DOMAIN].get(CONF_DEVICE_CLASS) != "firetv": + entity_id = "media_player.android_tv" + else: + entity_id = "media_player.fire_tv" - "Handles device/service unavailable. Log a warning once when - unavailable, log once when reconnected." + with patchers.patch_connect(True)[patch_key], patchers.patch_shell("")[patch_key]: + assert await async_setup_component(hass, DOMAIN, config) + await hass.helpers.entity_component.async_update_entity(entity_id) + state = hass.states.get(entity_id) + assert state is not None + assert state.state != STATE_UNAVAILABLE - https://developers.home-assistant.io/docs/en/integration_quality_scale_index.html - """ - with self.assertLogs(level=logging.WARNING) as logs: - with PATCH_ADB_SERVER_CONNECT_FAIL, PATCH_ADB_SERVER_COMMAND_FAIL: - for _ in range(5): - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) + with patchers.patch_shell(None)[patch_key], patchers.patch_shell(error=True)[ + patch_key + ]: + await hass.helpers.entity_component.async_update_entity(entity_id) + state = hass.states.get(entity_id) + assert state is not None + assert state.state == STATE_UNAVAILABLE - assert len(logs.output) == 2 - assert logs.output[0].startswith("ERROR") - assert logs.output[1].startswith("WARNING") - - with self.assertLogs(level=logging.DEBUG) as logs: - with PATCH_ADB_SERVER_CONNECT_SUCCESS: - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) - - assert ( - "ADB connection to {} via ADB server {}:{} successfully established".format( - self.aftv.aftv.host, - self.aftv.aftv.adb_server_ip, - self.aftv.aftv.adb_server_port, - ) - in logs.output[0] - ) - - def test_adb_shell_returns_none(self): - """Test the case that the ADB shell command returns `None`. - - The state should be `None` and the device should be unavailable. - """ - with PATCH_ADB_SERVER_COMMAND_NONE: - self.aftv.update() - self.assertFalse(self.aftv.available) - self.assertIsNone(self.aftv.state) - - with PATCH_ADB_SERVER_CONNECT_SUCCESS: - self.aftv.update() - self.assertTrue(self.aftv.available) - self.assertIsNotNone(self.aftv.state) + return True -class TestFireTVPythonImplementation(TestAndroidTVPythonImplementation): - """Test the androidtv media player for a Fire TV device.""" +async def test_reconnect_androidtv_python_adb(hass, caplog): + """Test that the error and reconnection attempts are logged correctly. - def setUp(self): - """Set up a `FireTVDevice` media player.""" - with PATCH_PYTHON_ADB_CONNECT_SUCCESS, PATCH_PYTHON_ADB_COMMAND_SUCCESS: - aftv = setup("IP:PORT", device_class="firetv") - self.aftv = FireTVDevice(aftv, "Fake Fire TV", {}, True, None, None) + * Device type: Android TV + * ADB connection method: Python ADB implementation + + """ + assert await _test_reconnect(hass, caplog, CONFIG_ANDROIDTV_PYTHON_ADB) -class TestFireTVServerImplementation(TestAndroidTVServerImplementation): - """Test the androidtv media player for a Fire TV device.""" +async def test_adb_shell_returns_none_androidtv_python_adb(hass): + """Test the case that the ADB shell command returns `None`. - def setUp(self): - """Set up a `FireTVDevice` media player.""" - with PATCH_ADB_SERVER_CONNECT_SUCCESS, PATCH_ADB_SERVER_AVAILABLE: - aftv = setup( - "IP:PORT", adb_server_ip="ADB_SERVER_IP", device_class="firetv" - ) - self.aftv = FireTVDevice(aftv, "Fake Fire TV", {}, True, None, None) + * Device type: Android TV + * ADB connection method: Python ADB implementation + + """ + assert await _test_adb_shell_returns_none(hass, CONFIG_ANDROIDTV_PYTHON_ADB) + + +async def test_reconnect_firetv_python_adb(hass, caplog): + """Test that the error and reconnection attempts are logged correctly. + + * Device type: Fire TV + * ADB connection method: Python ADB implementation + + """ + assert await _test_reconnect(hass, caplog, CONFIG_FIRETV_PYTHON_ADB) + + +async def test_adb_shell_returns_none_firetv_python_adb(hass): + """Test the case that the ADB shell command returns `None`. + + * Device type: Fire TV + * ADB connection method: Python ADB implementation + + """ + assert await _test_adb_shell_returns_none(hass, CONFIG_FIRETV_PYTHON_ADB) + + +async def test_reconnect_androidtv_adb_server(hass, caplog): + """Test that the error and reconnection attempts are logged correctly. + + * Device type: Android TV + * ADB connection method: ADB server + + """ + assert await _test_reconnect(hass, caplog, CONFIG_ANDROIDTV_ADB_SERVER) + + +async def test_adb_shell_returns_none_androidtv_adb_server(hass): + """Test the case that the ADB shell command returns `None`. + + * Device type: Android TV + * ADB connection method: ADB server + + """ + assert await _test_adb_shell_returns_none(hass, CONFIG_ANDROIDTV_ADB_SERVER) + + +async def test_reconnect_firetv_adb_server(hass, caplog): + """Test that the error and reconnection attempts are logged correctly. + + * Device type: Fire TV + * ADB connection method: ADB server + + """ + assert await _test_reconnect(hass, caplog, CONFIG_FIRETV_ADB_SERVER) + + +async def test_adb_shell_returns_none_firetv_adb_server(hass): + """Test the case that the ADB shell command returns `None`. + + * Device type: Fire TV + * ADB connection method: ADB server + + """ + assert await _test_adb_shell_returns_none(hass, CONFIG_FIRETV_ADB_SERVER)