From ef05aa2f39f07068f033900171779f4916a5a3e9 Mon Sep 17 00:00:00 2001 From: escoand Date: Fri, 10 Jan 2020 03:19:10 +0100 Subject: [PATCH] Add Samsung TV config flow (#28306) * add config flow * add tests * add user step error handling * remove unload function * add missing test file * handle authentication correctly * remove old discovery mode * better handling of remote class * optimized abort messages * add already configured test for user flow * Import order * use ip property instead context * Black * small syntax * use snake_case * Revert "use ip property instead context" This reverts commit 91502407eb216f8a0b1b90e3e6fb165b81406f8f. * disable wrong pylint errors * disable wrong no-member * Try to fix review comments * Try to fix review comments * Fix missing self * Fix ip checks * methods to functions * simplify user check * remove user errors * use async_setup for config * fix after rebase * import config to user config flow * patch all samsungctl * fix after rebase * fix notes * remove unused variable * ignore old setup function * fix after merge * pass configuration to import step * isort * fix recursion * remove timeout config * add turn on action (dry without testing) * use upstream checks * cleanup * minor * correctly await async method * ignore unused import * async call send_key * Revert "async call send_key" This reverts commit f37057819fd751a654779da743d0300751e963be. * fix comments * fix timeout test * test turn on action * Update media_player.py * Update test_media_player.py * Update test_media_player.py * use async executor * use newer ssdp data * update manually configured with ssdp data * dont setup component directly * ensure list * check updated device info * Update config_flow.py * Update __init__.py * fix duplicate check * simplified unique check * move method detection to config_flow * move unique test to init * fix after real world test * optimize config_validation * update device_info on ssdp discovery * cleaner update listener * fix lint * fix method signature * add note for manual config to confirm message * fix turn_on_action * pass script * patch delay * remove device info update --- .../components/discovery/__init__.py | 1 - .../components/samsungtv/__init__.py | 59 +++ .../components/samsungtv/config_flow.py | 184 +++++++++ homeassistant/components/samsungtv/const.py | 8 + .../components/samsungtv/manifest.json | 14 +- .../components/samsungtv/media_player.py | 175 ++------ .../components/samsungtv/strings.json | 26 ++ homeassistant/generated/config_flows.py | 1 + homeassistant/generated/ssdp.py | 5 + requirements_all.txt | 1 - requirements_test_all.txt | 1 - .../components/samsungtv/test_config_flow.py | 388 ++++++++++++++++++ tests/components/samsungtv/test_init.py | 97 +++++ .../components/samsungtv/test_media_player.py | 333 ++++----------- 14 files changed, 896 insertions(+), 397 deletions(-) create mode 100644 homeassistant/components/samsungtv/config_flow.py create mode 100644 homeassistant/components/samsungtv/strings.json create mode 100644 tests/components/samsungtv/test_config_flow.py create mode 100644 tests/components/samsungtv/test_init.py diff --git a/homeassistant/components/discovery/__init__.py b/homeassistant/components/discovery/__init__.py index 965782d1228..1e29d066f2d 100644 --- a/homeassistant/components/discovery/__init__.py +++ b/homeassistant/components/discovery/__init__.py @@ -75,7 +75,6 @@ SERVICE_HANDLERS = { "logitech_mediaserver": ("media_player", "squeezebox"), "directv": ("media_player", "directv"), "denonavr": ("media_player", "denonavr"), - "samsung_tv": ("media_player", "samsungtv"), "frontier_silicon": ("media_player", "frontier_silicon"), "openhome": ("media_player", "openhome"), "harmony": ("remote", "harmony"), diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py index 6b4f0e31f02..5647b407bfb 100644 --- a/homeassistant/components/samsungtv/__init__.py +++ b/homeassistant/components/samsungtv/__init__.py @@ -1 +1,60 @@ """The Samsung TV integration.""" +import socket + +import voluptuous as vol + +from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT +import homeassistant.helpers.config_validation as cv + +from .const import CONF_ON_ACTION, DEFAULT_NAME, DOMAIN + + +def ensure_unique_hosts(value): + """Validate that all configs have a unique host.""" + vol.Schema(vol.Unique("duplicate host entries found"))( + [socket.gethostbyname(entry[CONF_HOST]) for entry in value] + ) + return value + + +CONFIG_SCHEMA = vol.Schema( + { + DOMAIN: vol.All( + cv.ensure_list, + [ + vol.Schema( + { + vol.Required(CONF_HOST): cv.string, + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, + vol.Optional(CONF_PORT): cv.port, + vol.Optional(CONF_ON_ACTION): cv.SCRIPT_SCHEMA, + } + ) + ], + ensure_unique_hosts, + ) + }, + extra=vol.ALLOW_EXTRA, +) + + +async def async_setup(hass, config): + """Set up the Samsung TV integration.""" + if DOMAIN in config: + for entry_config in config[DOMAIN]: + hass.async_create_task( + hass.config_entries.flow.async_init( + DOMAIN, context={"source": "import"}, data=entry_config + ) + ) + + return True + + +async def async_setup_entry(hass, entry): + """Set up the Samsung TV platform.""" + hass.async_create_task( + hass.config_entries.async_forward_entry_setup(entry, "media_player") + ) + + return True diff --git a/homeassistant/components/samsungtv/config_flow.py b/homeassistant/components/samsungtv/config_flow.py new file mode 100644 index 00000000000..0bf39cc248b --- /dev/null +++ b/homeassistant/components/samsungtv/config_flow.py @@ -0,0 +1,184 @@ +"""Config flow for Samsung TV.""" +import socket +from urllib.parse import urlparse + +from samsungctl import Remote +from samsungctl.exceptions import AccessDenied, UnhandledResponse +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.components.ssdp import ( + ATTR_SSDP_LOCATION, + ATTR_UPNP_FRIENDLY_NAME, + ATTR_UPNP_MANUFACTURER, + ATTR_UPNP_MODEL_NAME, + ATTR_UPNP_UDN, +) +from homeassistant.const import ( + CONF_HOST, + CONF_ID, + CONF_IP_ADDRESS, + CONF_METHOD, + CONF_NAME, + CONF_PORT, +) + +# pylint:disable=unused-import +from .const import ( + CONF_MANUFACTURER, + CONF_MODEL, + CONF_ON_ACTION, + DOMAIN, + LOGGER, + METHODS, +) + +DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME): str}) + +RESULT_AUTH_MISSING = "auth_missing" +RESULT_SUCCESS = "success" +RESULT_NOT_FOUND = "not_found" +RESULT_NOT_SUPPORTED = "not_supported" + + +def _get_ip(host): + if host is None: + return None + return socket.gethostbyname(host) + + +class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a Samsung TV config flow.""" + + VERSION = 1 + CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL + + # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167 + + def __init__(self): + """Initialize flow.""" + self._host = None + self._ip = None + self._manufacturer = None + self._method = None + self._model = None + self._name = None + self._on_script = None + self._port = None + self._title = None + self._uuid = None + + def _get_entry(self): + return self.async_create_entry( + title=self._title, + data={ + CONF_HOST: self._host, + CONF_ID: self._uuid, + CONF_IP_ADDRESS: self._ip, + CONF_MANUFACTURER: self._manufacturer, + CONF_METHOD: self._method, + CONF_MODEL: self._model, + CONF_NAME: self._name, + CONF_ON_ACTION: self._on_script, + CONF_PORT: self._port, + }, + ) + + def _try_connect(self): + """Try to connect and check auth.""" + for method in METHODS: + config = { + "name": "HomeAssistant", + "description": "HomeAssistant", + "id": "ha.component.samsung", + "host": self._host, + "method": method, + "port": self._port, + "timeout": 1, + } + try: + LOGGER.debug("Try config: %s", config) + with Remote(config.copy()): + LOGGER.debug("Working config: %s", config) + self._method = method + return RESULT_SUCCESS + except AccessDenied: + LOGGER.debug("Working but denied config: %s", config) + return RESULT_AUTH_MISSING + except UnhandledResponse: + LOGGER.debug("Working but unsupported config: %s", config) + return RESULT_NOT_SUPPORTED + except (OSError): + LOGGER.debug("Failing config: %s", config) + + LOGGER.debug("No working config found") + return RESULT_NOT_FOUND + + async def async_step_import(self, user_input=None): + """Handle configuration by yaml file.""" + self._on_script = user_input.get(CONF_ON_ACTION) + self._port = user_input.get(CONF_PORT) + + return await self.async_step_user(user_input) + + async def async_step_user(self, user_input=None): + """Handle a flow initialized by the user.""" + if user_input is not None: + ip_address = await self.hass.async_add_executor_job( + _get_ip, user_input[CONF_HOST] + ) + + await self.async_set_unique_id(ip_address) + self._abort_if_unique_id_configured() + + self._host = user_input.get(CONF_HOST) + self._ip = self.context[CONF_IP_ADDRESS] = ip_address + self._title = user_input.get(CONF_NAME) + + result = await self.hass.async_add_executor_job(self._try_connect) + + if result != RESULT_SUCCESS: + return self.async_abort(reason=result) + return self._get_entry() + + return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA) + + async def async_step_ssdp(self, user_input=None): + """Handle a flow initialized by discovery.""" + host = urlparse(user_input[ATTR_SSDP_LOCATION]).hostname + ip_address = await self.hass.async_add_executor_job(_get_ip, host) + + self._host = host + self._ip = self.context[CONF_IP_ADDRESS] = ip_address + self._manufacturer = user_input[ATTR_UPNP_MANUFACTURER] + self._model = user_input[ATTR_UPNP_MODEL_NAME] + self._name = user_input[ATTR_UPNP_FRIENDLY_NAME] + if self._name.startswith("[TV]"): + self._name = self._name[4:] + self._title = f"{self._name} ({self._model})" + self._uuid = user_input[ATTR_UPNP_UDN] + if self._uuid.startswith("uuid:"): + self._uuid = self._uuid[5:] + + config_entry = await self.async_set_unique_id(ip_address) + if config_entry: + config_entry.data[CONF_ID] = self._uuid + config_entry.data[CONF_MANUFACTURER] = self._manufacturer + config_entry.data[CONF_MODEL] = self._model + self.hass.config_entries.async_update_entry(config_entry) + return self.async_abort(reason="already_configured") + + return await self.async_step_confirm() + + async def async_step_confirm(self, user_input=None): + """Handle user-confirmation of discovered node.""" + if user_input is not None: + result = await self.hass.async_add_executor_job(self._try_connect) + + if result != RESULT_SUCCESS: + return self.async_abort(reason=result) + return self._get_entry() + + return self.async_show_form( + step_id="confirm", description_placeholders={"model": self._model} + ) diff --git a/homeassistant/components/samsungtv/const.py b/homeassistant/components/samsungtv/const.py index 83d74743844..7cf71e406cb 100644 --- a/homeassistant/components/samsungtv/const.py +++ b/homeassistant/components/samsungtv/const.py @@ -3,3 +3,11 @@ import logging LOGGER = logging.getLogger(__package__) DOMAIN = "samsungtv" + +DEFAULT_NAME = "Samsung TV Remote" + +CONF_MANUFACTURER = "manufacturer" +CONF_MODEL = "model" +CONF_ON_ACTION = "turn_on_action" + +METHODS = ("websocket", "legacy") diff --git a/homeassistant/components/samsungtv/manifest.json b/homeassistant/components/samsungtv/manifest.json index d8db31db728..0d0a360fc20 100644 --- a/homeassistant/components/samsungtv/manifest.json +++ b/homeassistant/components/samsungtv/manifest.json @@ -2,7 +2,17 @@ "domain": "samsungtv", "name": "Samsung Smart TV", "documentation": "https://www.home-assistant.io/integrations/samsungtv", - "requirements": ["samsungctl[websocket]==0.7.1", "wakeonlan==1.1.6"], + "requirements": [ + "samsungctl[websocket]==0.7.1" + ], + "ssdp": [ + { + "deviceType": "urn:samsung.com:device:RemoteControlReceiver:1" + } + ], "dependencies": [], - "codeowners": ["@escoand"] + "codeowners": [ + "@escoand" + ], + "config_flow": true } diff --git a/homeassistant/components/samsungtv/media_player.py b/homeassistant/components/samsungtv/media_player.py index fd900fedec1..e7153a7f5d4 100644 --- a/homeassistant/components/samsungtv/media_player.py +++ b/homeassistant/components/samsungtv/media_player.py @@ -1,18 +1,12 @@ """Support for interface with an Samsung TV.""" import asyncio from datetime import timedelta -import socket from samsungctl import Remote as SamsungRemote, exceptions as samsung_exceptions import voluptuous as vol -import wakeonlan from websocket import WebSocketException -from homeassistant.components.media_player import ( - DEVICE_CLASS_TV, - PLATFORM_SCHEMA, - MediaPlayerDevice, -) +from homeassistant.components.media_player import DEVICE_CLASS_TV, MediaPlayerDevice from homeassistant.components.media_player.const import ( MEDIA_TYPE_CHANNEL, SUPPORT_NEXT_TRACK, @@ -27,27 +21,20 @@ from homeassistant.components.media_player.const import ( SUPPORT_VOLUME_STEP, ) from homeassistant.const import ( - CONF_BROADCAST_ADDRESS, CONF_HOST, - CONF_MAC, - CONF_NAME, + CONF_ID, + CONF_METHOD, CONF_PORT, - CONF_TIMEOUT, STATE_OFF, STATE_ON, ) import homeassistant.helpers.config_validation as cv +from homeassistant.helpers.script import Script from homeassistant.util import dt as dt_util -from .const import LOGGER - -DEFAULT_NAME = "Samsung TV Remote" -DEFAULT_TIMEOUT = 1 -DEFAULT_BROADCAST_ADDRESS = "255.255.255.255" +from .const import CONF_MANUFACTURER, CONF_MODEL, CONF_ON_ACTION, DOMAIN, LOGGER KEY_PRESS_TIMEOUT = 1.2 -KNOWN_DEVICES_KEY = "samsungtv_known_devices" -METHODS = ("websocket", "legacy") SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"} SUPPORT_SAMSUNGTV = ( @@ -62,73 +49,33 @@ SUPPORT_SAMSUNGTV = ( | SUPPORT_PLAY_MEDIA ) -PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( - { - vol.Required(CONF_HOST): cv.string, - vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, - vol.Optional(CONF_PORT): cv.port, - vol.Optional(CONF_MAC): cv.string, - vol.Optional( - CONF_BROADCAST_ADDRESS, default=DEFAULT_BROADCAST_ADDRESS - ): cv.string, - vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int, - } -) - -def setup_platform(hass, config, add_entities, discovery_info=None): +async def async_setup_platform( + hass, config, add_entities, discovery_info=None +): # pragma: no cover """Set up the Samsung TV platform.""" - known_devices = hass.data.get(KNOWN_DEVICES_KEY) - if known_devices is None: - known_devices = set() - hass.data[KNOWN_DEVICES_KEY] = known_devices + pass - uuid = None - # Is this a manual configuration? - if config.get(CONF_HOST) is not None: - host = config.get(CONF_HOST) - port = config.get(CONF_PORT) - name = config.get(CONF_NAME) - mac = config.get(CONF_MAC) - broadcast = config.get(CONF_BROADCAST_ADDRESS) - timeout = config.get(CONF_TIMEOUT) - elif discovery_info is not None: - tv_name = discovery_info.get("name") - model = discovery_info.get("model_name") - host = discovery_info.get("host") - name = f"{tv_name} ({model})" - if name.startswith("[TV]"): - name = name[4:] - port = None - timeout = DEFAULT_TIMEOUT - mac = None - broadcast = DEFAULT_BROADCAST_ADDRESS - 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. - ip_addr = socket.gethostbyname(host) - if ip_addr not in known_devices: - known_devices.add(ip_addr) - add_entities([SamsungTVDevice(host, port, name, timeout, mac, broadcast, uuid)]) - LOGGER.info("Samsung TV %s added as '%s'", host, name) - else: - LOGGER.info("Ignoring duplicate Samsung TV %s", host) +async def async_setup_entry(hass, config_entry, async_add_entities): + """Set up the Samsung TV from a config entry.""" + turn_on_action = config_entry.data.get(CONF_ON_ACTION) + on_script = Script(hass, turn_on_action) if turn_on_action else None + async_add_entities([SamsungTVDevice(config_entry, on_script)]) class SamsungTVDevice(MediaPlayerDevice): """Representation of a Samsung TV.""" - def __init__(self, host, port, name, timeout, mac, broadcast, uuid): + def __init__(self, config_entry, on_script): """Initialize the Samsung device.""" - - # Save a reference to the imported classes - self._name = name - self._mac = mac - self._broadcast = broadcast - self._uuid = uuid + self._config_entry = config_entry + self._name = config_entry.title + self._uuid = config_entry.data.get(CONF_ID) + self._manufacturer = config_entry.data.get(CONF_MANUFACTURER) + self._model = config_entry.data.get(CONF_MODEL) + self._on_script = on_script + self._update_listener = None # Assume that the TV is not muted self._muted = False # Assume that the TV is in Play mode @@ -141,57 +88,20 @@ class SamsungTVDevice(MediaPlayerDevice): # Generate a configuration for the Samsung library self._config = { "name": "HomeAssistant", - "description": name, + "description": self._name, "id": "ha.component.samsung", - "method": None, - "port": port, - "host": host, - "timeout": timeout, + "method": config_entry.data[CONF_METHOD], + "port": config_entry.data.get(CONF_PORT), + "host": config_entry.data[CONF_HOST], + "timeout": 1, } - # Select method by port number, mainly for fallback - if self._config["port"] in (8001, 8002): - self._config["method"] = "websocket" - elif self._config["port"] == 55000: - self._config["method"] = "legacy" - def update(self): """Update state of device.""" self.send_key("KEY") 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 = SamsungRemote(self._config.copy()) - self._state = STATE_ON - LOGGER.debug("Found working config: %s", self._config) - break - except ( - samsung_exceptions.UnhandledResponse, - samsung_exceptions.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 = SamsungRemote(self._config.copy()) @@ -219,9 +129,6 @@ class SamsungTVDevice(MediaPlayerDevice): # WebSocketException can occur when timed out self._remote = None self._state = STATE_ON - except AttributeError: - # Auto-detect could not find working config yet - pass except (samsung_exceptions.UnhandledResponse, samsung_exceptions.AccessDenied): # We got a response so it's on. self._state = STATE_ON @@ -256,6 +163,16 @@ class SamsungTVDevice(MediaPlayerDevice): """Return the state of the device.""" return self._state + @property + def device_info(self): + """Return device specific attributes.""" + return { + "name": self.name, + "identifiers": {(DOMAIN, self.unique_id)}, + "manufacturer": self._manufacturer, + "model": self._model, + } + @property def is_volume_muted(self): """Boolean if volume is currently muted.""" @@ -269,7 +186,7 @@ class SamsungTVDevice(MediaPlayerDevice): @property def supported_features(self): """Flag media player features that are supported.""" - if self._mac: + if self._on_script: return SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON return SUPPORT_SAMSUNGTV @@ -344,21 +261,19 @@ class SamsungTVDevice(MediaPlayerDevice): return for digit in media_id: - await self.hass.async_add_job(self.send_key, f"KEY_{digit}") + await self.hass.async_add_executor_job(self.send_key, f"KEY_{digit}") await asyncio.sleep(KEY_PRESS_TIMEOUT, self.hass.loop) - await self.hass.async_add_job(self.send_key, "KEY_ENTER") + await self.hass.async_add_executor_job(self.send_key, "KEY_ENTER") - def turn_on(self): + async def async_turn_on(self): """Turn the media player on.""" - if self._mac: - wakeonlan.send_magic_packet(self._mac, ip_address=self._broadcast) - else: - self.send_key("KEY_POWERON") + if self._on_script: + await self._on_script.async_run() - async def async_select_source(self, source): + def select_source(self, source): """Select input source.""" if source not in SOURCES: LOGGER.error("Unsupported source") return - await self.hass.async_add_job(self.send_key, SOURCES[source]) + self.send_key(SOURCES[source]) diff --git a/homeassistant/components/samsungtv/strings.json b/homeassistant/components/samsungtv/strings.json new file mode 100644 index 00000000000..ee762503e5c --- /dev/null +++ b/homeassistant/components/samsungtv/strings.json @@ -0,0 +1,26 @@ +{ + "config": { + "title": "Samsung TV", + "step": { + "user": { + "title": "Samsung TV", + "description": "Enter your Samsung TV information. If you never connected Home Assistant before you should see a popup on your TV asking for authentication.", + "data": { + "host": "Host or IP address", + "name": "Name" + } + }, + "confirm": { + "title": "Samsung TV", + "description": "Do you want to set up Samsung TV {model}? If you never connected Home Assistant before you should see a popup on your TV asking for authentication. Manual configurations for this TV will be overwritten." + } + }, + "abort": { + "already_in_progress": "Samsung TV configuration is already in progress.", + "already_configured": "This Samsung TV is already configured.", + "auth_missing": "Home Assistant is not authenticated to connect to this Samsung TV.", + "not_found": "No supported Samsung TV devices found on the network.", + "not_supported": "This Samsung TV devices is currently not supported." + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 6f3f0e714f6..c5ea3f1a5d9 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -66,6 +66,7 @@ FLOWS = [ "point", "ps4", "rainmachine", + "samsungtv", "sentry", "simplisafe", "smartthings", diff --git a/homeassistant/generated/ssdp.py b/homeassistant/generated/ssdp.py index adf3a345bbe..01e0726ce54 100644 --- a/homeassistant/generated/ssdp.py +++ b/homeassistant/generated/ssdp.py @@ -27,6 +27,11 @@ SSDP = { "manufacturer": "Royal Philips Electronics" } ], + "samsungtv": [ + { + "deviceType": "urn:samsung.com:device:RemoteControlReceiver:1" + } + ], "sonos": [ { "st": "urn:schemas-upnp-org:device:ZonePlayer:1" diff --git a/requirements_all.txt b/requirements_all.txt index fe14f075618..9fc7a4f27dc 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -2035,7 +2035,6 @@ vtjp==0.1.14 vultr==0.1.2 # homeassistant.components.panasonic_viera -# homeassistant.components.samsungtv # homeassistant.components.wake_on_lan wakeonlan==1.1.6 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 59001b5e1cb..e38912df53f 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -643,7 +643,6 @@ vsure==1.5.4 vultr==0.1.2 # homeassistant.components.panasonic_viera -# homeassistant.components.samsungtv # homeassistant.components.wake_on_lan wakeonlan==1.1.6 diff --git a/tests/components/samsungtv/test_config_flow.py b/tests/components/samsungtv/test_config_flow.py new file mode 100644 index 00000000000..ce6741f0703 --- /dev/null +++ b/tests/components/samsungtv/test_config_flow.py @@ -0,0 +1,388 @@ +"""Tests for Samsung TV config flow.""" +from unittest.mock import call, patch + +from asynctest import mock +import pytest +from samsungctl.exceptions import AccessDenied, UnhandledResponse + +from homeassistant.components.samsungtv.const import ( + CONF_MANUFACTURER, + CONF_MODEL, + DOMAIN, +) +from homeassistant.components.ssdp import ( + ATTR_SSDP_LOCATION, + ATTR_UPNP_FRIENDLY_NAME, + ATTR_UPNP_MANUFACTURER, + ATTR_UPNP_MODEL_NAME, + ATTR_UPNP_UDN, +) +from homeassistant.const import CONF_HOST, CONF_ID, CONF_METHOD, CONF_NAME + +MOCK_USER_DATA = {CONF_HOST: "fake_host", CONF_NAME: "fake_name"} +MOCK_SSDP_DATA = { + ATTR_SSDP_LOCATION: "https://fake_host:12345/test", + ATTR_UPNP_FRIENDLY_NAME: "[TV]fake_name", + ATTR_UPNP_MANUFACTURER: "fake_manufacturer", + ATTR_UPNP_MODEL_NAME: "fake_model", + ATTR_UPNP_UDN: "uuid:fake_uuid", +} +MOCK_SSDP_DATA_NOPREFIX = { + ATTR_SSDP_LOCATION: "http://fake2_host:12345/test", + ATTR_UPNP_FRIENDLY_NAME: "fake2_name", + ATTR_UPNP_MANUFACTURER: "fake2_manufacturer", + ATTR_UPNP_MODEL_NAME: "fake2_model", + ATTR_UPNP_UDN: "fake2_uuid", +} + +AUTODETECT_WEBSOCKET = { + "name": "HomeAssistant", + "description": "HomeAssistant", + "id": "ha.component.samsung", + "method": "websocket", + "port": None, + "host": "fake_host", + "timeout": 1, +} +AUTODETECT_LEGACY = { + "name": "HomeAssistant", + "description": "HomeAssistant", + "id": "ha.component.samsung", + "method": "legacy", + "port": None, + "host": "fake_host", + "timeout": 1, +} + + +@pytest.fixture(name="remote") +def remote_fixture(): + """Patch the samsungctl Remote.""" + with patch("samsungctl.Remote") as remote_class, patch( + "homeassistant.components.samsungtv.config_flow.socket" + ) as socket_class: + remote = mock.Mock() + remote.__enter__ = mock.Mock() + remote.__exit__ = mock.Mock() + remote_class.return_value = remote + socket = mock.Mock() + socket_class.return_value = socket + yield remote + + +async def test_user(hass, remote): + """Test starting a flow by user.""" + + # show form + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"} + ) + assert result["type"] == "form" + assert result["step_id"] == "user" + + # entry was added + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_USER_DATA + ) + assert result["type"] == "create_entry" + assert result["title"] == "fake_name" + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] is None + assert result["data"][CONF_MANUFACTURER] is None + assert result["data"][CONF_MODEL] is None + assert result["data"][CONF_ID] is None + + +async def test_user_missing_auth(hass): + """Test starting a flow by user with authentication.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=AccessDenied("Boom"), + ), patch("homeassistant.components.samsungtv.config_flow.socket"): + + # missing authentication + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "auth_missing" + + +async def test_user_not_supported(hass): + """Test starting a flow by user for not supported device.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=UnhandledResponse("Boom"), + ), patch("homeassistant.components.samsungtv.config_flow.socket"): + + # device not supported + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "not_supported" + + +async def test_user_not_found(hass): + """Test starting a flow by user but no device found.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=OSError("Boom"), + ), patch("homeassistant.components.samsungtv.config_flow.socket"): + + # device not found + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "not_found" + + +async def test_user_already_configured(hass, remote): + """Test starting a flow by user when already configured.""" + + # entry was added + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "create_entry" + + # failed as already configured + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + + +async def test_ssdp(hass, remote): + """Test starting a flow from discovery.""" + + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + # entry was added + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input="whatever" + ) + assert result["type"] == "create_entry" + assert result["title"] == "fake_name (fake_model)" + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] == "fake_name" + assert result["data"][CONF_MANUFACTURER] == "fake_manufacturer" + assert result["data"][CONF_MODEL] == "fake_model" + assert result["data"][CONF_ID] == "fake_uuid" + + +async def test_ssdp_noprefix(hass, remote): + """Test starting a flow from discovery without prefixes.""" + + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA_NOPREFIX + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + # entry was added + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input="whatever" + ) + assert result["type"] == "create_entry" + assert result["title"] == "fake2_name (fake2_model)" + assert result["data"][CONF_HOST] == "fake2_host" + assert result["data"][CONF_NAME] == "fake2_name" + assert result["data"][CONF_MANUFACTURER] == "fake2_manufacturer" + assert result["data"][CONF_MODEL] == "fake2_model" + assert result["data"][CONF_ID] == "fake2_uuid" + + +async def test_ssdp_missing_auth(hass): + """Test starting a flow from discovery with authentication.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=AccessDenied("Boom"), + ), patch("homeassistant.components.samsungtv.config_flow.socket"): + + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + # missing authentication + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input="whatever" + ) + assert result["type"] == "abort" + assert result["reason"] == "auth_missing" + + +async def test_ssdp_not_supported(hass): + """Test starting a flow from discovery for not supported device.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=UnhandledResponse("Boom"), + ), patch("homeassistant.components.samsungtv.config_flow.socket"): + + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + # device not supported + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input="whatever" + ) + assert result["type"] == "abort" + assert result["reason"] == "not_supported" + + +async def test_ssdp_not_found(hass): + """Test starting a flow from discovery but no device found.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=OSError("Boom"), + ), patch("homeassistant.components.samsungtv.config_flow.socket"): + + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + # device not found + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input="whatever" + ) + assert result["type"] == "abort" + assert result["reason"] == "not_found" + + +async def test_ssdp_already_in_progress(hass, remote): + """Test starting a flow from discovery twice.""" + + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + # failed as already in progress + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "already_in_progress" + + +async def test_ssdp_already_configured(hass, remote): + """Test starting a flow from discovery when already configured.""" + + # entry was added + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "create_entry" + assert result["data"][CONF_MANUFACTURER] is None + assert result["data"][CONF_MODEL] is None + assert result["data"][CONF_ID] is None + + # failed as already configured + result2 = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "ssdp"}, data=MOCK_SSDP_DATA + ) + assert result2["type"] == "abort" + assert result2["reason"] == "already_configured" + + # check updated device info + assert result["data"][CONF_MANUFACTURER] == "fake_manufacturer" + assert result["data"][CONF_MODEL] == "fake_model" + assert result["data"][CONF_ID] == "fake_uuid" + + +async def test_autodetect_websocket(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch("homeassistant.components.samsungtv.config_flow.Remote") as remote: + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "create_entry" + assert result["data"][CONF_METHOD] == "websocket" + assert remote.call_count == 1 + assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)] + + +async def test_autodetect_auth_missing(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=[AccessDenied("Boom")], + ) as remote: + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "auth_missing" + assert remote.call_count == 1 + assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)] + + +async def test_autodetect_not_supported(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=[UnhandledResponse("Boom")], + ) as remote: + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "not_supported" + assert remote.call_count == 1 + assert remote.call_args_list == [call(AUTODETECT_WEBSOCKET)] + + +async def test_autodetect_legacy(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=[OSError("Boom"), mock.DEFAULT], + ) as remote: + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "create_entry" + assert result["data"][CONF_METHOD] == "legacy" + assert remote.call_count == 2 + assert remote.call_args_list == [ + call(AUTODETECT_WEBSOCKET), + call(AUTODETECT_LEGACY), + ] + + +async def test_autodetect_none(hass, remote): + """Test for send key with autodetection of protocol.""" + with patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=OSError("Boom"), + ) as remote: + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "user"}, data=MOCK_USER_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == "not_found" + assert remote.call_count == 2 + assert remote.call_args_list == [ + call(AUTODETECT_WEBSOCKET), + call(AUTODETECT_LEGACY), + ] diff --git a/tests/components/samsungtv/test_init.py b/tests/components/samsungtv/test_init.py new file mode 100644 index 00000000000..55ec52b56ae --- /dev/null +++ b/tests/components/samsungtv/test_init.py @@ -0,0 +1,97 @@ +"""Tests for the Samsung TV Integration.""" +from unittest.mock import call, patch + +import pytest + +from homeassistant.components.media_player.const import DOMAIN, SUPPORT_TURN_ON +from homeassistant.components.samsungtv.const import ( + CONF_ON_ACTION, + DOMAIN as SAMSUNGTV_DOMAIN, +) +from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV +from homeassistant.const import ( + ATTR_ENTITY_ID, + ATTR_SUPPORTED_FEATURES, + CONF_HOST, + CONF_NAME, + CONF_PORT, + SERVICE_VOLUME_UP, +) +from homeassistant.setup import async_setup_component + +ENTITY_ID = f"{DOMAIN}.fake_name" +MOCK_CONFIG = { + SAMSUNGTV_DOMAIN: [ + { + CONF_HOST: "fake_host", + CONF_NAME: "fake_name", + CONF_PORT: 1234, + CONF_ON_ACTION: [{"delay": "00:00:01"}], + } + ] +} +REMOTE_CALL = { + "name": "HomeAssistant", + "description": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_NAME], + "id": "ha.component.samsung", + "method": "websocket", + "port": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_PORT], + "host": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_HOST], + "timeout": 1, +} + + +@pytest.fixture(name="remote") +def remote_fixture(): + """Patch the samsungctl Remote.""" + with patch("homeassistant.components.samsungtv.socket"), patch( + "homeassistant.components.samsungtv.config_flow.socket" + ), patch("homeassistant.components.samsungtv.config_flow.Remote"), patch( + "homeassistant.components.samsungtv.media_player.SamsungRemote" + ) as remote: + yield remote + + +async def test_setup(hass, remote): + """Test Samsung TV integration is setup.""" + await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) + await hass.async_block_till_done() + state = hass.states.get(ENTITY_ID) + + # test name and turn_on + assert state + assert state.name == "fake_name" + assert ( + state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON + ) + + # test host and port + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert remote.mock_calls[0] == call(REMOTE_CALL) + + +async def test_setup_duplicate_config(hass, remote, caplog): + """Test duplicate setup of platform.""" + DUPLICATE = { + SAMSUNGTV_DOMAIN: [ + MOCK_CONFIG[SAMSUNGTV_DOMAIN][0], + MOCK_CONFIG[SAMSUNGTV_DOMAIN][0], + ] + } + await async_setup_component(hass, SAMSUNGTV_DOMAIN, DUPLICATE) + await hass.async_block_till_done() + assert hass.states.get(ENTITY_ID) is None + assert len(hass.states.async_all()) == 0 + assert "duplicate host entries found" in caplog.text + + +async def test_setup_duplicate_entries(hass, remote, caplog): + """Test duplicate setup of platform.""" + await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) + await hass.async_block_till_done() + assert hass.states.get(ENTITY_ID) + assert len(hass.states.async_all()) == 1 + await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) + assert len(hass.states.async_all()) == 1 diff --git a/tests/components/samsungtv/test_media_player.py b/tests/components/samsungtv/test_media_player.py index bb40dc28445..3afedda746e 100644 --- a/tests/components/samsungtv/test_media_player.py +++ b/tests/components/samsungtv/test_media_player.py @@ -2,9 +2,9 @@ import asyncio from datetime import timedelta import logging -from unittest.mock import call, patch from asynctest import mock +from asynctest.mock import call, patch import pytest from samsungctl import exceptions from websocket import WebSocketException @@ -22,21 +22,18 @@ from homeassistant.components.media_player.const import ( SERVICE_SELECT_SOURCE, SUPPORT_TURN_ON, ) -from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN -from homeassistant.components.samsungtv.media_player import ( - CONF_TIMEOUT, - SUPPORT_SAMSUNGTV, +from homeassistant.components.samsungtv.const import ( + CONF_ON_ACTION, + DOMAIN as SAMSUNGTV_DOMAIN, ) +from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV from homeassistant.const import ( ATTR_DEVICE_CLASS, ATTR_ENTITY_ID, ATTR_FRIENDLY_NAME, ATTR_SUPPORTED_FEATURES, - CONF_BROADCAST_ADDRESS, CONF_HOST, - CONF_MAC, CONF_NAME, - CONF_PLATFORM, CONF_PORT, SERVICE_MEDIA_NEXT_TRACK, SERVICE_MEDIA_PAUSE, @@ -49,9 +46,7 @@ from homeassistant.const import ( SERVICE_VOLUME_UP, STATE_OFF, STATE_ON, - STATE_UNKNOWN, ) -from homeassistant.helpers.discovery import async_load_platform from homeassistant.setup import async_setup_component import homeassistant.util.dt as dt_util @@ -59,107 +54,46 @@ from tests.common import async_fire_time_changed 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: "38:f9:d3:82:b4:f1", - } + SAMSUNGTV_DOMAIN: [ + { + CONF_HOST: "fake", + CONF_NAME: "fake", + CONF_PORT: 8001, + CONF_ON_ACTION: [{"delay": "00:00:01"}], + } + ] } -ENTITY_ID_BROADCAST = f"{DOMAIN}.fake_broadcast" -MOCK_CONFIG_BROADCAST = { - DOMAIN: { - CONF_PLATFORM: SAMSUNGTV_DOMAIN, - CONF_HOST: "fake_broadcast", - CONF_NAME: "fake_broadcast", - CONF_PORT: 8001, - CONF_TIMEOUT: 10, - CONF_MAC: "38:f9:d3:82:b4:f1", - CONF_BROADCAST_ADDRESS: "192.168.5.255", - } -} - -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, +ENTITY_ID_NOTURNON = f"{DOMAIN}.fake_noturnon" +MOCK_CONFIG_NOTURNON = { + SAMSUNGTV_DOMAIN: [ + {CONF_HOST: "fake_noturnon", CONF_NAME: "fake_noturnon", CONF_PORT: 55000} + ] } @pytest.fixture(name="remote") def remote_fixture(): """Patch the samsungctl Remote.""" - with patch( + with patch("homeassistant.components.samsungtv.config_flow.socket"), patch( + "homeassistant.components.samsungtv.config_flow.Remote" + ), patch( "homeassistant.components.samsungtv.media_player.SamsungRemote" ) as remote_class, patch( - "homeassistant.components.samsungtv.media_player.socket" - ) as socket_class: + "homeassistant.components.samsungtv.socket" + ): remote = mock.Mock() remote_class.return_value = remote - socket = mock.Mock() - socket_class.return_value = socket yield remote -@pytest.fixture(name="wakeonlan") -def wakeonlan_fixture(): - """Patch the wakeonlan Remote.""" +@pytest.fixture(name="delay") +def delay_fixture(): + """Patch the delay script function.""" with patch( - "homeassistant.components.samsungtv.media_player.wakeonlan" - ) as wakeonlan_module: - yield wakeonlan_module + "homeassistant.components.samsungtv.media_player.Script.async_run" + ) as delay: + yield delay @pytest.fixture @@ -170,61 +104,20 @@ def mock_now(): async def setup_samsungtv(hass, config): """Set up mock Samsung TV.""" - await async_setup_component(hass, "media_player", config) + await async_setup_component(hass, SAMSUNGTV_DOMAIN, config) await hass.async_block_till_done() -async def test_setup_with_mac(hass, remote): +async def test_setup_with_turnon(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): +async def test_setup_without_turnon(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" + await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON) + assert hass.states.get(ENTITY_ID_NOTURNON) async def test_update_on(hass, remote, mock_now): @@ -254,7 +147,7 @@ async def test_update_off(hass, remote, mock_now): assert state.state == STATE_OFF -async def test_send_key(hass, remote, wakeonlan): +async def test_send_key(hass, remote): """Test for send key.""" await setup_samsungtv(hass, MOCK_CONFIG) assert await hass.services.async_call( @@ -267,85 +160,6 @@ async def test_send_key(hass, remote, wakeonlan): assert state.state == STATE_ON -async def test_send_key_autodetect_websocket(hass, remote): - """Test for send key with autodetection of protocol.""" - with patch( - "homeassistant.components.samsungtv.media_player.SamsungRemote" - ) 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.""" - caplog.set_level(logging.DEBUG) - with patch( - "homeassistant.components.samsungtv.media_player.SamsungRemote", - 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( - "homeassistant.components.samsungtv.media_player.SamsungRemote", - 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( - "homeassistant.components.samsungtv.media_player.SamsungRemote", - 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) @@ -417,7 +231,7 @@ async def test_name(hass, remote): assert state.attributes[ATTR_FRIENDLY_NAME] == "fake" -async def test_state_with_mac(hass, remote, wakeonlan): +async def test_state_with_turnon(hass, remote, delay): """Test for state property.""" await setup_samsungtv(hass, MOCK_CONFIG) assert await hass.services.async_call( @@ -425,6 +239,8 @@ async def test_state_with_mac(hass, remote, wakeonlan): ) state = hass.states.get(ENTITY_ID) assert state.state == STATE_ON + assert delay.call_count == 1 + assert await hass.services.async_call( DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True ) @@ -432,22 +248,22 @@ async def test_state_with_mac(hass, remote, wakeonlan): assert state.state == STATE_OFF -async def test_state_without_mac(hass, remote): +async def test_state_without_turnon(hass, remote): """Test for state property.""" - await setup_samsungtv(hass, MOCK_CONFIG_NOMAC) + await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON) assert await hass.services.async_call( - DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True ) - state = hass.states.get(ENTITY_ID_NOMAC) + state = hass.states.get(ENTITY_ID_NOTURNON) assert state.state == STATE_ON assert await hass.services.async_call( - DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOMAC}, True + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True ) - state = hass.states.get(ENTITY_ID_NOMAC) + state = hass.states.get(ENTITY_ID_NOTURNON) assert state.state == STATE_OFF -async def test_supported_features_with_mac(hass, remote): +async def test_supported_features_with_turnon(hass, remote): """Test for supported_features property.""" await setup_samsungtv(hass, MOCK_CONFIG) state = hass.states.get(ENTITY_ID) @@ -456,10 +272,10 @@ async def test_supported_features_with_mac(hass, remote): ) -async def test_supported_features_without_mac(hass, remote): +async def test_supported_features_without_turnon(hass, remote): """Test for supported_features property.""" - await setup_samsungtv(hass, MOCK_CONFIG_NOMAC) - state = hass.states.get(ENTITY_ID_NOMAC) + await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON) + state = hass.states.get(ENTITY_ID_NOTURNON) assert state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORT_SAMSUNGTV @@ -481,15 +297,25 @@ async def test_turn_off_websocket(hass, remote): assert remote.control.call_args_list == [call("KEY_POWER")] -async def test_turn_off_legacy(hass, remote): +async def test_turn_off_legacy(hass): """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")] + with patch("homeassistant.components.samsungtv.config_flow.socket"), patch( + "homeassistant.components.samsungtv.config_flow.Remote", + side_effect=[OSError("Boom"), mock.DEFAULT], + ), patch( + "homeassistant.components.samsungtv.media_player.SamsungRemote" + ) as remote_class, patch( + "homeassistant.components.samsungtv.socket" + ): + remote = mock.Mock() + remote_class.return_value = remote + await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON) + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, 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): @@ -583,37 +409,20 @@ async def test_media_previous_track(hass, remote): assert remote.control.call_args_list == [call("KEY_CHDOWN"), call("KEY")] -async def test_turn_on_with_mac(hass, remote, wakeonlan): +async def test_turn_on_with_turnon(hass, remote, delay): """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("38:f9:d3:82:b4:f1", ip_address="255.255.255.255") - ] + assert delay.call_count == 1 -async def test_turn_on_with_mac_and_broadcast(hass, remote, wakeonlan): +async def test_turn_on_without_turnon(hass, remote): """Test turn on.""" - await setup_samsungtv(hass, MOCK_CONFIG_BROADCAST) + await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON) assert await hass.services.async_call( - DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID_BROADCAST}, True - ) - # key and update called - assert wakeonlan.send_magic_packet.call_count == 1 - assert wakeonlan.send_magic_packet.call_args_list == [ - call("38:f9:d3:82:b4:f1", ip_address="192.168.5.255") - ] - - -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 + DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True ) # nothing called as not supported feature assert remote.control.call_count == 0