From 9661efc31275280540940a5700703778c9d5f3ae Mon Sep 17 00:00:00 2001 From: escoand Date: Fri, 25 Oct 2019 14:32:12 +0200 Subject: [PATCH] Add Samsung TV automatic protocol detection (#27492) * added automatic protocol detection * fix logger tests * fix async tests * add missin const.py * fix log formatting * wait for first update call * migrate first tests * migrated all test functions * started to use state machine * updated all tests to use async_setup_component * slove hints * update tests * get state at correct position * remove impossible tests * fix autodetect tests * use caplog fixture * add test for duplicate * catch concrete exceptions * don't mock samsungctl exceptions * add test for discovery * get state when possible * add test for autodetect without connection --- CODEOWNERS | 1 + .../components/samsungtv/__init__.py | 2 +- homeassistant/components/samsungtv/const.py | 5 + .../components/samsungtv/manifest.json | 4 +- .../components/samsungtv/media_player.py | 76 +- .../components/samsungtv/test_media_player.py | 934 ++++++++++++------ 6 files changed, 689 insertions(+), 333 deletions(-) create mode 100644 homeassistant/components/samsungtv/const.py diff --git a/CODEOWNERS b/CODEOWNERS index 809101a5271..40e37ec4697 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -247,6 +247,7 @@ homeassistant/components/rfxtrx/* @danielhiversen homeassistant/components/rmvtransport/* @cgtobi homeassistant/components/roomba/* @pschmitt homeassistant/components/saj/* @fredericvl +homeassistant/components/samsungtv/* @escoand homeassistant/components/scene/* @home-assistant/core homeassistant/components/scrape/* @fabaff homeassistant/components/script/* @home-assistant/core diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py index e43ea1ba984..6b4f0e31f02 100644 --- a/homeassistant/components/samsungtv/__init__.py +++ b/homeassistant/components/samsungtv/__init__.py @@ -1 +1 @@ -"""The samsungtv component.""" +"""The Samsung TV integration.""" diff --git a/homeassistant/components/samsungtv/const.py b/homeassistant/components/samsungtv/const.py new file mode 100644 index 00000000000..83d74743844 --- /dev/null +++ b/homeassistant/components/samsungtv/const.py @@ -0,0 +1,5 @@ +"""Constants for the Samsung TV integration.""" +import logging + +LOGGER = logging.getLogger(__package__) +DOMAIN = "samsungtv" diff --git a/homeassistant/components/samsungtv/manifest.json b/homeassistant/components/samsungtv/manifest.json index a080fac112a..405d757cbef 100644 --- a/homeassistant/components/samsungtv/manifest.json +++ b/homeassistant/components/samsungtv/manifest.json @@ -1,11 +1,11 @@ { "domain": "samsungtv", - "name": "Samsungtv", + "name": "Samsung TV", "documentation": "https://www.home-assistant.io/integrations/samsungtv", "requirements": [ "samsungctl[websocket]==0.7.1", "wakeonlan==1.1.6" ], "dependencies": [], - "codeowners": [] + "codeowners": ["@escoand"] } diff --git a/homeassistant/components/samsungtv/media_player.py b/homeassistant/components/samsungtv/media_player.py index fd1da31497e..94e9131ed32 100644 --- a/homeassistant/components/samsungtv/media_player.py +++ b/homeassistant/components/samsungtv/media_player.py @@ -1,7 +1,6 @@ """Support for interface with an Samsung TV.""" import asyncio from datetime import timedelta -import logging import socket import voluptuous as vol @@ -36,14 +35,14 @@ from homeassistant.const import ( import homeassistant.helpers.config_validation as cv from homeassistant.util import dt as dt_util -_LOGGER = logging.getLogger(__name__) +from .const import LOGGER DEFAULT_NAME = "Samsung TV Remote" -DEFAULT_PORT = 55000 DEFAULT_TIMEOUT = 1 KEY_PRESS_TIMEOUT = 1.2 KNOWN_DEVICES_KEY = "samsungtv_known_devices" +METHODS = ("websocket", "legacy") SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"} SUPPORT_SAMSUNGTV = ( @@ -62,7 +61,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( { vol.Required(CONF_HOST): cv.string, vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, - vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port, + vol.Optional(CONF_PORT): cv.port, vol.Optional(CONF_MAC): cv.string, vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int, } @@ -89,15 +88,14 @@ def setup_platform(hass, config, add_entities, discovery_info=None): model = discovery_info.get("model_name") host = discovery_info.get("host") name = f"{tv_name} ({model})" - port = DEFAULT_PORT + if name.startswith("[TV]"): + name = name[4:] + port = None timeout = DEFAULT_TIMEOUT mac = None - udn = discovery_info.get("udn") - if udn and udn.startswith("uuid:"): - uuid = udn[len("uuid:") :] - else: - _LOGGER.warning("Cannot determine device") - return + uuid = discovery_info.get("udn") + if uuid and uuid.startswith("uuid:"): + uuid = uuid[len("uuid:") :] # Only add a device once, so discovered devices do not override manual # config. @@ -105,9 +103,9 @@ def setup_platform(hass, config, add_entities, discovery_info=None): if ip_addr not in known_devices: known_devices.add(ip_addr) add_entities([SamsungTVDevice(host, port, name, timeout, mac, uuid)]) - _LOGGER.info("Samsung TV %s:%d added as '%s'", host, port, name) + LOGGER.info("Samsung TV %s added as '%s'", host, name) else: - _LOGGER.info("Ignoring duplicate Samsung TV %s:%d", host, port) + LOGGER.info("Ignoring duplicate Samsung TV %s", host) class SamsungTVDevice(MediaPlayerDevice): @@ -140,14 +138,16 @@ class SamsungTVDevice(MediaPlayerDevice): "name": "HomeAssistant", "description": name, "id": "ha.component.samsung", + "method": None, "port": port, "host": host, "timeout": timeout, } + # Select method by port number, mainly for fallback if self._config["port"] in (8001, 8002): self._config["method"] = "websocket" - else: + elif self._config["port"] == 55000: self._config["method"] = "legacy" def update(self): @@ -156,16 +156,47 @@ class SamsungTVDevice(MediaPlayerDevice): def get_remote(self): """Create or return a remote control instance.""" + + # Try to find correct method automatically + if self._config["method"] not in METHODS: + for method in METHODS: + try: + self._config["method"] = method + LOGGER.debug("Try config: %s", self._config) + self._remote = self._remote_class(self._config.copy()) + self._state = STATE_ON + LOGGER.debug("Found working config: %s", self._config) + break + except ( + self._exceptions_class.UnhandledResponse, + self._exceptions_class.AccessDenied, + ): + # We got a response so it's working. + self._state = STATE_ON + LOGGER.debug( + "Found working config without connection: %s", self._config + ) + break + except OSError as err: + LOGGER.debug("Failing config: %s error was: %s", self._config, err) + self._config["method"] = None + + # Unable to find working connection + if self._config["method"] is None: + self._remote = None + self._state = None + return None + if self._remote is None: # We need to create a new instance to reconnect. - self._remote = self._remote_class(self._config) + self._remote = self._remote_class(self._config.copy()) return self._remote def send_key(self, key): """Send a key to the tv and handles exceptions.""" if self._power_off_in_progress() and key not in ("KEY_POWER", "KEY_POWEROFF"): - _LOGGER.info("TV is powering off, not sending command: %s", key) + LOGGER.info("TV is powering off, not sending command: %s", key) return try: # recreate connection if connection was dead @@ -178,6 +209,9 @@ class SamsungTVDevice(MediaPlayerDevice): # BrokenPipe can occur when the commands is sent to fast self._remote = None self._state = STATE_ON + except AttributeError: + # Auto-detect could not find working config yet + pass except ( self._exceptions_class.UnhandledResponse, self._exceptions_class.AccessDenied, @@ -185,7 +219,7 @@ class SamsungTVDevice(MediaPlayerDevice): # We got a response so it's on. self._state = STATE_ON self._remote = None - _LOGGER.debug("Failed sending command %s", key, exc_info=True) + LOGGER.debug("Failed sending command %s", key, exc_info=True) return except OSError: self._state = STATE_OFF @@ -249,7 +283,7 @@ class SamsungTVDevice(MediaPlayerDevice): self.get_remote().close() self._remote = None except OSError: - _LOGGER.debug("Could not establish connection.") + LOGGER.debug("Could not establish connection.") def volume_up(self): """Volume up the media player.""" @@ -291,14 +325,14 @@ class SamsungTVDevice(MediaPlayerDevice): async def async_play_media(self, media_type, media_id, **kwargs): """Support changing a channel.""" if media_type != MEDIA_TYPE_CHANNEL: - _LOGGER.error("Unsupported media type") + LOGGER.error("Unsupported media type") return # media_id should only be a channel number try: cv.positive_int(media_id) except vol.Invalid: - _LOGGER.error("Media ID must be positive integer") + LOGGER.error("Media ID must be positive integer") return for digit in media_id: @@ -316,7 +350,7 @@ class SamsungTVDevice(MediaPlayerDevice): async def async_select_source(self, source): """Select input source.""" if source not in SOURCES: - _LOGGER.error("Unsupported source") + LOGGER.error("Unsupported source") return await self.hass.async_add_job(self.send_key, SOURCES[source]) diff --git a/tests/components/samsungtv/test_media_player.py b/tests/components/samsungtv/test_media_player.py index 1428ba3b39b..c178710e3f9 100644 --- a/tests/components/samsungtv/test_media_player.py +++ b/tests/components/samsungtv/test_media_player.py @@ -1,309 +1,573 @@ """Tests for samsungtv Components.""" import asyncio -import unittest -from unittest.mock import call, patch, MagicMock - from asynctest import mock - +from datetime import timedelta import pytest +from samsungctl import exceptions +from tests.common import MockDependency, async_fire_time_changed +from unittest.mock import call, patch -import tests.common from homeassistant.components.media_player import DEVICE_CLASS_TV from homeassistant.components.media_player.const import ( + ATTR_INPUT_SOURCE, + ATTR_MEDIA_CONTENT_ID, + ATTR_MEDIA_CONTENT_TYPE, + ATTR_MEDIA_VOLUME_MUTED, + DOMAIN, + SERVICE_PLAY_MEDIA, + SERVICE_SELECT_SOURCE, SUPPORT_TURN_ON, MEDIA_TYPE_CHANNEL, MEDIA_TYPE_URL, ) +from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN from homeassistant.components.samsungtv.media_player import ( - setup_platform, CONF_TIMEOUT, - SamsungTVDevice, SUPPORT_SAMSUNGTV, ) from homeassistant.const import ( + ATTR_DEVICE_CLASS, + ATTR_ENTITY_ID, + ATTR_FRIENDLY_NAME, + ATTR_SUPPORTED_FEATURES, CONF_HOST, - CONF_NAME, - CONF_PORT, - STATE_ON, CONF_MAC, + CONF_NAME, + CONF_PLATFORM, + CONF_PORT, + SERVICE_MEDIA_NEXT_TRACK, + SERVICE_MEDIA_PAUSE, + SERVICE_MEDIA_PLAY, + SERVICE_MEDIA_PREVIOUS_TRACK, + SERVICE_TURN_OFF, + SERVICE_TURN_ON, + SERVICE_VOLUME_DOWN, + SERVICE_VOLUME_MUTE, + SERVICE_VOLUME_UP, STATE_OFF, + STATE_ON, + STATE_UNKNOWN, ) -from tests.common import MockDependency -from homeassistant.util import dt as dt_util -from datetime import timedelta +from homeassistant.helpers.discovery import async_load_platform +from homeassistant.setup import async_setup_component +import homeassistant.util.dt as dt_util -WORKING_CONFIG = { - CONF_HOST: "fake", - CONF_NAME: "fake", - CONF_PORT: 8001, - CONF_TIMEOUT: 10, - CONF_MAC: "fake", - "uuid": None, + +ENTITY_ID = f"{DOMAIN}.fake" +MOCK_CONFIG = { + DOMAIN: { + CONF_PLATFORM: SAMSUNGTV_DOMAIN, + CONF_HOST: "fake", + CONF_NAME: "fake", + CONF_PORT: 8001, + CONF_TIMEOUT: 10, + CONF_MAC: "fake", + } } -DISCOVERY_INFO = {"name": "fake", "model_name": "fake", "host": "fake"} +ENTITY_ID_NOMAC = f"{DOMAIN}.fake_nomac" +MOCK_CONFIG_NOMAC = { + DOMAIN: { + CONF_PLATFORM: SAMSUNGTV_DOMAIN, + CONF_HOST: "fake_nomac", + CONF_NAME: "fake_nomac", + CONF_PORT: 55000, + CONF_TIMEOUT: 10, + } +} + +ENTITY_ID_AUTO = f"{DOMAIN}.fake_auto" +MOCK_CONFIG_AUTO = { + DOMAIN: { + CONF_PLATFORM: SAMSUNGTV_DOMAIN, + CONF_HOST: "fake_auto", + CONF_NAME: "fake_auto", + } +} + +ENTITY_ID_DISCOVERY = f"{DOMAIN}.fake_discovery_fake_model" +MOCK_CONFIG_DISCOVERY = { + "name": "fake_discovery", + "model_name": "fake_model", + "host": "fake_host", + "udn": "fake_uuid", +} + +ENTITY_ID_DISCOVERY_PREFIX = f"{DOMAIN}.fake_discovery_prefix_fake_model_prefix" +MOCK_CONFIG_DISCOVERY_PREFIX = { + "name": "[TV]fake_discovery_prefix", + "model_name": "fake_model_prefix", + "host": "fake_host_prefix", + "udn": "uuid:fake_uuid_prefix", +} + +AUTODETECT_WEBSOCKET = { + "name": "HomeAssistant", + "description": "fake_auto", + "id": "ha.component.samsung", + "method": "websocket", + "port": None, + "host": "fake_auto", + "timeout": 1, +} +AUTODETECT_LEGACY = { + "name": "HomeAssistant", + "description": "fake_auto", + "id": "ha.component.samsung", + "method": "legacy", + "port": None, + "host": "fake_auto", + "timeout": 1, +} -class AccessDenied(Exception): - """Dummy Exception.""" +@pytest.fixture(name="remote") +def remote_fixture(): + """Patch the samsungctl Remote.""" + with patch("samsungctl.Remote") as remote_class, patch( + "homeassistant.components.samsungtv.media_player.socket" + ) as socket_class: + remote = mock.Mock() + remote_class.return_value = remote + socket = mock.Mock() + socket_class.return_value = socket + yield remote -class ConnectionClosed(Exception): - """Dummy Exception.""" - - -class UnhandledResponse(Exception): - """Dummy Exception.""" - - -class TestSamsungTv(unittest.TestCase): - """Testing Samsungtv component.""" - - @MockDependency("samsungctl") - @MockDependency("wakeonlan") - def setUp(self, samsung_mock, wol_mock): - """Set up test environment.""" - self.hass = tests.common.get_test_home_assistant() - self.hass.start() - self.hass.block_till_done() - self.device = SamsungTVDevice(**WORKING_CONFIG) - self.device._exceptions_class = mock.Mock() - self.device._exceptions_class.UnhandledResponse = UnhandledResponse - self.device._exceptions_class.AccessDenied = AccessDenied - self.device._exceptions_class.ConnectionClosed = ConnectionClosed - - def tearDown(self): - """Tear down test data.""" - self.hass.stop() - - @MockDependency("samsungctl") - @MockDependency("wakeonlan") - def test_setup(self, samsung_mock, wol_mock): - """Testing setup of platform.""" - with mock.patch("homeassistant.components.samsungtv.media_player.socket"): - add_entities = mock.Mock() - setup_platform(self.hass, WORKING_CONFIG, add_entities) - - @MockDependency("samsungctl") - @MockDependency("wakeonlan") - def test_setup_discovery(self, samsung_mock, wol_mock): - """Testing setup of platform with discovery.""" - with mock.patch("homeassistant.components.samsungtv.media_player.socket"): - add_entities = mock.Mock() - setup_platform(self.hass, {}, add_entities, discovery_info=DISCOVERY_INFO) - - @MockDependency("samsungctl") - @MockDependency("wakeonlan") - @mock.patch("homeassistant.components.samsungtv.media_player._LOGGER.warning") - def test_setup_none(self, samsung_mock, wol_mock, mocked_warn): - """Testing setup of platform with no data.""" - with mock.patch("homeassistant.components.samsungtv.media_player.socket"): - add_entities = mock.Mock() - setup_platform(self.hass, {}, add_entities, discovery_info=None) - mocked_warn.assert_called_once_with("Cannot determine device") - add_entities.assert_not_called() - - def test_update_on(self): - """Testing update tv on.""" - self.device.update() - self.assertEqual(STATE_ON, self.device._state) - - def test_update_off(self): - """Testing update tv off.""" - _remote = mock.Mock() - _remote.control = mock.Mock(side_effect=OSError("Boom")) - self.device.get_remote = mock.Mock(return_value=_remote) - self.device.update() - assert STATE_OFF == self.device._state - - def test_send_key(self): - """Test for send key.""" - self.device.send_key("KEY_POWER") - self.assertEqual(STATE_ON, self.device._state) - - def test_send_key_broken_pipe(self): - """Testing broken pipe Exception.""" - _remote = mock.Mock() - _remote.control = mock.Mock(side_effect=BrokenPipeError("Boom")) - self.device.get_remote = mock.Mock(return_value=_remote) - self.device.send_key("HELLO") - self.assertIsNone(self.device._remote) - self.assertEqual(STATE_ON, self.device._state) - - def test_send_key_connection_closed_retry_succeed(self): - """Test retry on connection closed.""" - _remote = mock.Mock() - _remote.control = mock.Mock( - side_effect=[ - self.device._exceptions_class.ConnectionClosed("Boom"), - mock.DEFAULT, - ] - ) - self.device.get_remote = mock.Mock(return_value=_remote) - command = "HELLO" - self.device.send_key(command) - self.assertEqual(STATE_ON, self.device._state) - # verify that _remote.control() get called twice because of retry logic - expected = [mock.call(command), mock.call(command)] - assert expected == _remote.control.call_args_list - - def test_send_key_unhandled_response(self): - """Testing unhandled response exception.""" - _remote = mock.Mock() - _remote.control = mock.Mock( - side_effect=self.device._exceptions_class.UnhandledResponse("Boom") - ) - self.device.get_remote = mock.Mock(return_value=_remote) - self.device.send_key("HELLO") - self.assertIsNone(self.device._remote) - self.assertEqual(STATE_ON, self.device._state) - - def test_send_key_os_error(self): - """Testing broken pipe Exception.""" - _remote = mock.Mock() - _remote.control = mock.Mock(side_effect=OSError("Boom")) - self.device.get_remote = mock.Mock(return_value=_remote) - self.device.send_key("HELLO") - assert self.device._remote is None - assert STATE_OFF == self.device._state - - def test_power_off_in_progress(self): - """Test for power_off_in_progress.""" - assert not self.device._power_off_in_progress() - self.device._end_of_power_off = dt_util.utcnow() + timedelta(seconds=15) - assert self.device._power_off_in_progress() - - def test_name(self): - """Test for name property.""" - assert "fake" == self.device.name - - def test_state(self): - """Test for state property.""" - self.device._state = STATE_ON - self.assertEqual(STATE_ON, self.device.state) - self.device._state = STATE_OFF - assert STATE_OFF == self.device.state - - def test_is_volume_muted(self): - """Test for is_volume_muted property.""" - self.device._muted = False - assert not self.device.is_volume_muted - self.device._muted = True - assert self.device.is_volume_muted - - def test_supported_features(self): - """Test for supported_features property.""" - self.device._mac = None - assert SUPPORT_SAMSUNGTV == self.device.supported_features - self.device._mac = "fake" - assert SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON == self.device.supported_features - - def test_device_class(self): - """Test for device_class property.""" - assert DEVICE_CLASS_TV == self.device.device_class - - def test_turn_off(self): - """Test for turn_off.""" - self.device.send_key = mock.Mock() - _remote = mock.Mock() - _remote.close = mock.Mock() - self.get_remote = mock.Mock(return_value=_remote) - self.device._end_of_power_off = None - self.device.turn_off() - assert self.device._end_of_power_off is not None - self.device.send_key.assert_called_once_with("KEY_POWER") - self.device.send_key = mock.Mock() - self.device._config["method"] = "legacy" - self.device.turn_off() - self.device.send_key.assert_called_once_with("KEY_POWEROFF") - - @mock.patch("homeassistant.components.samsungtv.media_player._LOGGER.debug") - def test_turn_off_os_error(self, mocked_debug): - """Test for turn_off with OSError.""" - _remote = mock.Mock() - _remote.close = mock.Mock(side_effect=OSError("BOOM")) - self.device.get_remote = mock.Mock(return_value=_remote) - self.device.turn_off() - mocked_debug.assert_called_once_with("Could not establish connection.") - - def test_volume_up(self): - """Test for volume_up.""" - self.device.send_key = mock.Mock() - self.device.volume_up() - self.device.send_key.assert_called_once_with("KEY_VOLUP") - - def test_volume_down(self): - """Test for volume_down.""" - self.device.send_key = mock.Mock() - self.device.volume_down() - self.device.send_key.assert_called_once_with("KEY_VOLDOWN") - - def test_mute_volume(self): - """Test for mute_volume.""" - self.device.send_key = mock.Mock() - self.device.mute_volume(True) - self.device.send_key.assert_called_once_with("KEY_MUTE") - - def test_media_play_pause(self): - """Test for media_next_track.""" - self.device.send_key = mock.Mock() - self.device._playing = False - self.device.media_play_pause() - self.device.send_key.assert_called_once_with("KEY_PLAY") - assert self.device._playing - self.device.send_key = mock.Mock() - self.device.media_play_pause() - self.device.send_key.assert_called_once_with("KEY_PAUSE") - assert not self.device._playing - - def test_media_play(self): - """Test for media_play.""" - self.device.send_key = mock.Mock() - self.device._playing = False - self.device.media_play() - self.device.send_key.assert_called_once_with("KEY_PLAY") - assert self.device._playing - - def test_media_pause(self): - """Test for media_pause.""" - self.device.send_key = mock.Mock() - self.device._playing = True - self.device.media_pause() - self.device.send_key.assert_called_once_with("KEY_PAUSE") - assert not self.device._playing - - def test_media_next_track(self): - """Test for media_next_track.""" - self.device.send_key = mock.Mock() - self.device.media_next_track() - self.device.send_key.assert_called_once_with("KEY_FF") - - def test_media_previous_track(self): - """Test for media_previous_track.""" - self.device.send_key = mock.Mock() - self.device.media_previous_track() - self.device.send_key.assert_called_once_with("KEY_REWIND") - - def test_turn_on(self): - """Test turn on.""" - self.device.send_key = mock.Mock() - self.device._mac = None - self.device.turn_on() - self.device.send_key.assert_called_once_with("KEY_POWERON") - self.device._wol.send_magic_packet = mock.Mock() - self.device._mac = "fake" - self.device.turn_on() - self.device._wol.send_magic_packet.assert_called_once_with("fake") +@pytest.fixture(name="wakeonlan") +def wakeonlan_fixture(): + """Patch the wakeonlan Remote.""" + with MockDependency("wakeonlan") as wakeonlan: + yield wakeonlan @pytest.fixture -def samsung_mock(): - """Mock samsungctl.""" - with patch.dict("sys.modules", {"samsungctl": MagicMock()}): - yield +def mock_now(): + """Fixture for dtutil.now.""" + return dt_util.utcnow() -async def test_play_media(hass, samsung_mock): +async def setup_samsungtv(hass, config): + """Set up mock Samsung TV.""" + await async_setup_component(hass, "media_player", config) + await hass.async_block_till_done() + + +async def test_setup_with_mac(hass, remote): + """Test setup of platform.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert hass.states.get(ENTITY_ID) + + +async def test_setup_duplicate(hass, remote, caplog): + """Test duplicate setup of platform.""" + DUPLICATE = {DOMAIN: [MOCK_CONFIG[DOMAIN], MOCK_CONFIG[DOMAIN]]} + await setup_samsungtv(hass, DUPLICATE) + assert "Ignoring duplicate Samsung TV fake" in caplog.text + + +async def test_setup_without_mac(hass, remote): + """Test setup of platform.""" + await setup_samsungtv(hass, MOCK_CONFIG_NOMAC) + assert hass.states.get(ENTITY_ID_NOMAC) + + +async def test_setup_discovery(hass, remote): + """Test setup of platform with discovery.""" + hass.async_create_task( + async_load_platform( + hass, DOMAIN, SAMSUNGTV_DOMAIN, MOCK_CONFIG_DISCOVERY, {DOMAIN: {}} + ) + ) + await hass.async_block_till_done() + state = hass.states.get(ENTITY_ID_DISCOVERY) + assert state + assert state.name == "fake_discovery (fake_model)" + entity_registry = await hass.helpers.entity_registry.async_get_registry() + entry = entity_registry.async_get(ENTITY_ID_DISCOVERY) + assert entry + assert entry.unique_id == "fake_uuid" + + +async def test_setup_discovery_prefix(hass, remote): + """Test setup of platform with discovery.""" + hass.async_create_task( + async_load_platform( + hass, DOMAIN, SAMSUNGTV_DOMAIN, MOCK_CONFIG_DISCOVERY_PREFIX, {DOMAIN: {}} + ) + ) + await hass.async_block_till_done() + state = hass.states.get(ENTITY_ID_DISCOVERY_PREFIX) + assert state + assert state.name == "fake_discovery_prefix (fake_model_prefix)" + entity_registry = await hass.helpers.entity_registry.async_get_registry() + entry = entity_registry.async_get(ENTITY_ID_DISCOVERY_PREFIX) + assert entry + assert entry.unique_id == "fake_uuid_prefix" + + +async def test_update_on(hass, remote, mock_now): + """Testing update tv on.""" + await setup_samsungtv(hass, MOCK_CONFIG) + + next_update = mock_now + timedelta(minutes=5) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + +async def test_update_off(hass, remote, mock_now): + """Testing update tv off.""" + await setup_samsungtv(hass, MOCK_CONFIG) + remote.control = mock.Mock(side_effect=OSError("Boom")) + + next_update = mock_now + timedelta(minutes=5) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_OFF + + +async def test_send_key(hass, remote, wakeonlan): + """Test for send key.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + state = hass.states.get(ENTITY_ID) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_VOLUP"), call("KEY")] + assert state.state == STATE_ON + + +async def test_send_key_autodetect_websocket(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch("samsungctl.Remote") as remote, patch( + "homeassistant.components.samsungtv.media_player.socket" + ): + await setup_samsungtv(hass, MOCK_CONFIG_AUTO) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True + ) + state = hass.states.get(ENTITY_ID_AUTO) + assert remote.call_count == 1 + assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)] + assert state.state == STATE_ON + + +async def test_send_key_autodetect_websocket_exception(hass, caplog): + """Test for send key with autodetection of protocol.""" + with patch( + "samsungctl.Remote", side_effect=[exceptions.AccessDenied("Boom"), mock.DEFAULT] + ) as remote, patch("homeassistant.components.samsungtv.media_player.socket"): + await setup_samsungtv(hass, MOCK_CONFIG_AUTO) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True + ) + state = hass.states.get(ENTITY_ID_AUTO) + # called 2 times because of the exception and the send key + assert remote.call_count == 2 + assert remote.call_args_list == [ + call(AUTODETECT_WEBSOCKET), + call(AUTODETECT_WEBSOCKET), + ] + assert state.state == STATE_ON + assert "Found working config without connection: " in caplog.text + assert "Failing config: " not in caplog.text + + +async def test_send_key_autodetect_legacy(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch( + "samsungctl.Remote", side_effect=[OSError("Boom"), mock.DEFAULT] + ) as remote, patch("homeassistant.components.samsungtv.media_player.socket"): + await setup_samsungtv(hass, MOCK_CONFIG_AUTO) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True + ) + state = hass.states.get(ENTITY_ID_AUTO) + assert remote.call_count == 2 + assert remote.call_args_list == [ + call(AUTODETECT_WEBSOCKET), + call(AUTODETECT_LEGACY), + ] + assert state.state == STATE_ON + + +async def test_send_key_autodetect_none(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch("samsungctl.Remote", side_effect=OSError("Boom")) as remote, patch( + "homeassistant.components.samsungtv.media_player.socket" + ): + await setup_samsungtv(hass, MOCK_CONFIG_AUTO) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_AUTO}, True + ) + state = hass.states.get(ENTITY_ID_AUTO) + # 4 calls because of retry + assert remote.call_count == 4 + assert remote.call_args_list == [ + call(AUTODETECT_WEBSOCKET), + call(AUTODETECT_LEGACY), + call(AUTODETECT_WEBSOCKET), + call(AUTODETECT_LEGACY), + ] + assert state.state == STATE_UNKNOWN + + +async def test_send_key_broken_pipe(hass, remote): + """Testing broken pipe Exception.""" + await setup_samsungtv(hass, MOCK_CONFIG) + remote.control = mock.Mock(side_effect=BrokenPipeError("Boom")) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + +async def test_send_key_connection_closed_retry_succeed(hass, remote): + """Test retry on connection closed.""" + await setup_samsungtv(hass, MOCK_CONFIG) + remote.control = mock.Mock( + side_effect=[exceptions.ConnectionClosed("Boom"), mock.DEFAULT, mock.DEFAULT] + ) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + state = hass.states.get(ENTITY_ID) + # key because of retry two times and update called + assert remote.control.call_count == 3 + assert remote.control.call_args_list == [ + call("KEY_VOLUP"), + call("KEY_VOLUP"), + call("KEY"), + ] + assert state.state == STATE_ON + + +async def test_send_key_unhandled_response(hass, remote): + """Testing unhandled response exception.""" + await setup_samsungtv(hass, MOCK_CONFIG) + remote.control = mock.Mock(side_effect=exceptions.UnhandledResponse("Boom")) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + +async def test_send_key_os_error(hass, remote): + """Testing broken pipe Exception.""" + await setup_samsungtv(hass, MOCK_CONFIG) + remote.control = mock.Mock(side_effect=OSError("Boom")) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_OFF + + +async def test_name(hass, remote): + """Test for name property.""" + await setup_samsungtv(hass, MOCK_CONFIG) + state = hass.states.get(ENTITY_ID) + assert state.attributes[ATTR_FRIENDLY_NAME] == "fake" + + +async def test_state_with_mac(hass, remote, wakeonlan): + """Test for state property.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_OFF + + +async def test_state_without_mac(hass, remote): + """Test for state property.""" + await setup_samsungtv(hass, MOCK_CONFIG_NOMAC) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True + ) + state = hass.states.get(ENTITY_ID_NOMAC) + assert state.state == STATE_ON + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True + ) + state = hass.states.get(ENTITY_ID_NOMAC) + assert state.state == STATE_OFF + + +async def test_supported_features_with_mac(hass, remote): + """Test for supported_features property.""" + await setup_samsungtv(hass, MOCK_CONFIG) + state = hass.states.get(ENTITY_ID) + assert ( + state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON + ) + + +async def test_supported_features_without_mac(hass, remote): + """Test for supported_features property.""" + await setup_samsungtv(hass, MOCK_CONFIG_NOMAC) + state = hass.states.get(ENTITY_ID_NOMAC) + assert state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV + + +async def test_device_class(hass, remote): + """Test for device_class property.""" + await setup_samsungtv(hass, MOCK_CONFIG) + state = hass.states.get(ENTITY_ID) + assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS_TV + + +async def test_turn_off_websocket(hass, remote): + """Test for turn_off.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key called + assert remote.control.call_count == 1 + assert remote.control.call_args_list == [call("KEY_POWER")] + + +async def test_turn_off_legacy(hass, remote): + """Test for turn_off.""" + await setup_samsungtv(hass, MOCK_CONFIG_NOMAC) + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True + ) + # key called + assert remote.control.call_count == 1 + assert remote.control.call_args_list == [call("KEY_POWEROFF")] + + +async def test_turn_off_os_error(hass, remote, caplog): + """Test for turn_off with OSError.""" + await setup_samsungtv(hass, MOCK_CONFIG) + remote.close = mock.Mock(side_effect=OSError("BOOM")) + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert "Could not establish connection." in caplog.text + + +async def test_volume_up(hass, remote): + """Test for volume_up.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_VOLUP"), call("KEY")] + + +async def test_volume_down(hass, remote): + """Test for volume_down.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_DOWN, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_VOLDOWN"), call("KEY")] + + +async def test_mute_volume(hass, remote): + """Test for mute_volume.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, + SERVICE_VOLUME_MUTE, + {ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: True}, + True, + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_MUTE"), call("KEY")] + + +async def test_media_play(hass, remote): + """Test for media_play.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_MEDIA_PLAY, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_PLAY"), call("KEY")] + + +async def test_media_pause(hass, remote): + """Test for media_pause.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_MEDIA_PAUSE, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_PAUSE"), call("KEY")] + + +async def test_media_next_track(hass, remote): + """Test for media_next_track.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_MEDIA_NEXT_TRACK, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_FF"), call("KEY")] + + +async def test_media_previous_track(hass, remote): + """Test for media_previous_track.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_MEDIA_PREVIOUS_TRACK, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_REWIND"), call("KEY")] + + +async def test_turn_on_with_mac(hass, remote, wakeonlan): + """Test turn on.""" + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + # key and update called + assert wakeonlan.send_magic_packet.call_count == 1 + assert wakeonlan.send_magic_packet.call_args_list == [call("fake")] + + +async def test_turn_on_without_mac(hass, remote): + """Test turn on.""" + await setup_samsungtv(hass, MOCK_CONFIG_NOMAC) + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True + ) + # nothing called as not supported feature + assert remote.control.call_count == 0 + + +async def test_play_media(hass, remote): """Test for play_media.""" asyncio_sleep = asyncio.sleep sleeps = [] @@ -312,57 +576,109 @@ async def test_play_media(hass, samsung_mock): sleeps.append(duration) await asyncio_sleep(0, loop=loop) + await setup_samsungtv(hass, MOCK_CONFIG) with patch("asyncio.sleep", new=sleep): - device = SamsungTVDevice(**WORKING_CONFIG) - device.hass = hass - - device.send_key = mock.Mock() - await device.async_play_media(MEDIA_TYPE_CHANNEL, "576") - - exp = [call("KEY_5"), call("KEY_7"), call("KEY_6"), call("KEY_ENTER")] - assert device.send_key.call_args_list == exp + assert await hass.services.async_call( + DOMAIN, + SERVICE_PLAY_MEDIA, + { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_CHANNEL, + ATTR_MEDIA_CONTENT_ID: "576", + }, + True, + ) + # keys and update called + assert remote.control.call_count == 5 + assert remote.control.call_args_list == [ + call("KEY_5"), + call("KEY_7"), + call("KEY_6"), + call("KEY_ENTER"), + call("KEY"), + ] assert len(sleeps) == 3 -async def test_play_media_invalid_type(hass, samsung_mock): +async def test_play_media_invalid_type(hass, remote): """Test for play_media with invalid media type.""" url = "https://example.com" - device = SamsungTVDevice(**WORKING_CONFIG) - device.send_key = mock.Mock() - await device.async_play_media(MEDIA_TYPE_URL, url) - assert device.send_key.call_count == 0 + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, + SERVICE_PLAY_MEDIA, + { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_URL, + ATTR_MEDIA_CONTENT_ID: url, + }, + True, + ) + # only update called + assert remote.control.call_count == 1 + assert remote.control.call_args_list == [call("KEY")] -async def test_play_media_channel_as_string(hass, samsung_mock): +async def test_play_media_channel_as_string(hass, remote): """Test for play_media with invalid channel as string.""" url = "https://example.com" - device = SamsungTVDevice(**WORKING_CONFIG) - device.send_key = mock.Mock() - await device.async_play_media(MEDIA_TYPE_CHANNEL, url) - assert device.send_key.call_count == 0 + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, + SERVICE_PLAY_MEDIA, + { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_CHANNEL, + ATTR_MEDIA_CONTENT_ID: url, + }, + True, + ) + # only update called + assert remote.control.call_count == 1 + assert remote.control.call_args_list == [call("KEY")] -async def test_play_media_channel_as_non_positive(hass, samsung_mock): +async def test_play_media_channel_as_non_positive(hass, remote): """Test for play_media with invalid channel as non positive integer.""" - device = SamsungTVDevice(**WORKING_CONFIG) - device.send_key = mock.Mock() - await device.async_play_media(MEDIA_TYPE_CHANNEL, "-4") - assert device.send_key.call_count == 0 + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, + SERVICE_PLAY_MEDIA, + { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_CHANNEL, + ATTR_MEDIA_CONTENT_ID: "-4", + }, + True, + ) + # only update called + assert remote.control.call_count == 1 + assert remote.control.call_args_list == [call("KEY")] -async def test_select_source(hass, samsung_mock): +async def test_select_source(hass, remote): """Test for select_source.""" - device = SamsungTVDevice(**WORKING_CONFIG) - device.hass = hass - device.send_key = mock.Mock() - await device.async_select_source("HDMI") - exp = [call("KEY_HDMI")] - assert device.send_key.call_args_list == exp + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, + SERVICE_SELECT_SOURCE, + {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "HDMI"}, + True, + ) + # key and update called + assert remote.control.call_count == 2 + assert remote.control.call_args_list == [call("KEY_HDMI"), call("KEY")] -async def test_select_source_invalid_source(hass, samsung_mock): +async def test_select_source_invalid_source(hass, remote): """Test for select_source with invalid source.""" - device = SamsungTVDevice(**WORKING_CONFIG) - device.send_key = mock.Mock() - await device.async_select_source("INVALID") - assert device.send_key.call_count == 0 + await setup_samsungtv(hass, MOCK_CONFIG) + assert await hass.services.async_call( + DOMAIN, + SERVICE_SELECT_SOURCE, + {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "INVALID"}, + True, + ) + # only update called + assert remote.control.call_count == 1 + assert remote.control.call_args_list == [call("KEY")]