From b9a0fb93ebcddcb5da2d3629daa34bc702180f61 Mon Sep 17 00:00:00 2001 From: Simone Chemelli Date: Sat, 22 May 2021 17:41:18 +0200 Subject: [PATCH] Add samsungtv dhcp and zeroconf discovery (#48022) Co-authored-by: J. Nick Koston --- .coveragerc | 1 + CODEOWNERS | 2 +- .../components/samsungtv/__init__.py | 110 ++- homeassistant/components/samsungtv/bridge.py | 61 +- .../components/samsungtv/config_flow.py | 280 +++++-- homeassistant/components/samsungtv/const.py | 10 + .../components/samsungtv/manifest.json | 18 +- .../components/samsungtv/media_player.py | 76 +- .../components/samsungtv/strings.json | 22 +- .../components/samsungtv/translations/en.json | 18 +- homeassistant/generated/dhcp.py | 4 + homeassistant/generated/zeroconf.py | 6 + tests/components/samsungtv/__init__.py | 14 + tests/components/samsungtv/conftest.py | 112 +++ .../components/samsungtv/test_config_flow.py | 709 ++++++++++++++---- tests/components/samsungtv/test_init.py | 43 +- .../components/samsungtv/test_media_player.py | 103 ++- 17 files changed, 1174 insertions(+), 415 deletions(-) create mode 100644 tests/components/samsungtv/conftest.py diff --git a/.coveragerc b/.coveragerc index 227d932643c..e56773678c2 100644 --- a/.coveragerc +++ b/.coveragerc @@ -873,6 +873,7 @@ omit = homeassistant/components/russound_rnet/media_player.py homeassistant/components/sabnzbd/* homeassistant/components/saj/sensor.py + homeassistant/components/samsungtv/bridge.py homeassistant/components/satel_integra/* homeassistant/components/schluter/* homeassistant/components/scrape/sensor.py diff --git a/CODEOWNERS b/CODEOWNERS index 12689918e55..8049fec94b0 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -412,7 +412,7 @@ homeassistant/components/rpi_power/* @shenxn @swetoast homeassistant/components/ruckus_unleashed/* @gabe565 homeassistant/components/safe_mode/* @home-assistant/core homeassistant/components/saj/* @fredericvl -homeassistant/components/samsungtv/* @escoand +homeassistant/components/samsungtv/* @escoand @chemelli74 homeassistant/components/scene/* @home-assistant/core homeassistant/components/schluter/* @prairieapps homeassistant/components/scrape/* @fabaff diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py index 64646533b2d..31b666793af 100644 --- a/homeassistant/components/samsungtv/__init__.py +++ b/homeassistant/components/samsungtv/__init__.py @@ -5,20 +5,31 @@ import voluptuous as vol from homeassistant import config_entries from homeassistant.components.media_player.const import DOMAIN as MP_DOMAIN -from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT +from homeassistant.const import ( + CONF_HOST, + CONF_METHOD, + CONF_NAME, + CONF_PORT, + CONF_TOKEN, + EVENT_HOMEASSISTANT_STOP, +) +from homeassistant.core import callback import homeassistant.helpers.config_validation as cv -from .const import CONF_ON_ACTION, DEFAULT_NAME, DOMAIN +from .bridge import SamsungTVBridge +from .const import CONF_ON_ACTION, DEFAULT_NAME, DOMAIN, LOGGER 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] + [entry[CONF_HOST] for entry in value] ) return value +PLATFORMS = [MP_DOMAIN] + CONFIG_SCHEMA = vol.Schema( { DOMAIN: vol.All( @@ -43,30 +54,87 @@ CONFIG_SCHEMA = vol.Schema( async def async_setup(hass, config): """Set up the Samsung TV integration.""" - if DOMAIN in config: - hass.data[DOMAIN] = {} - for entry_config in config[DOMAIN]: - ip_address = await hass.async_add_executor_job( - socket.gethostbyname, entry_config[CONF_HOST] - ) - hass.data[DOMAIN][ip_address] = { - CONF_ON_ACTION: entry_config.get(CONF_ON_ACTION) - } - hass.async_create_task( - hass.config_entries.flow.async_init( - DOMAIN, - context={"source": config_entries.SOURCE_IMPORT}, - data=entry_config, - ) - ) + hass.data[DOMAIN] = {} + if DOMAIN not in config: + return True + for entry_config in config[DOMAIN]: + ip_address = await hass.async_add_executor_job( + socket.gethostbyname, entry_config[CONF_HOST] + ) + hass.data[DOMAIN][ip_address] = { + CONF_ON_ACTION: entry_config.get(CONF_ON_ACTION) + } + hass.async_create_task( + hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_IMPORT}, + data=entry_config, + ) + ) return True +@callback +def _async_get_device_bridge(data): + """Get device bridge.""" + return SamsungTVBridge.get_bridge( + data[CONF_METHOD], + data[CONF_HOST], + data[CONF_PORT], + data.get(CONF_TOKEN), + ) + + 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, MP_DOMAIN) + + # Initialize bridge + data = entry.data.copy() + bridge = _async_get_device_bridge(data) + if bridge.port is None and bridge.default_port is not None: + # For backward compat, set default port for websocket tv + data[CONF_PORT] = bridge.default_port + hass.config_entries.async_update_entry(entry, data=data) + bridge = _async_get_device_bridge(data) + + def stop_bridge(event): + """Stop SamsungTV bridge connection.""" + bridge.stop() + + entry.async_on_unload( + hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_bridge) ) + hass.data[DOMAIN][entry.entry_id] = bridge + hass.config_entries.async_setup_platforms(entry, PLATFORMS) + return True + + +async def async_unload_entry(hass, entry): + """Unload a config entry.""" + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + if unload_ok: + hass.data[DOMAIN][entry.entry_id].stop() + return unload_ok + + +async def async_migrate_entry(hass, config_entry): + """Migrate old entry.""" + version = config_entry.version + + LOGGER.debug("Migrating from version %s", version) + + # 1 -> 2: Unique ID format changed, so delete and re-import: + if version == 1: + dev_reg = await hass.helpers.device_registry.async_get_registry() + dev_reg.async_clear_config_entry(config_entry) + + en_reg = await hass.helpers.entity_registry.async_get_registry() + en_reg.async_clear_config_entry(config_entry) + + version = config_entry.version = 2 + hass.config_entries.async_update_entry(config_entry) + LOGGER.debug("Migration to version %s successful", version) + return True diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py index dc8eb862ff7..84b518a4633 100644 --- a/homeassistant/components/samsungtv/bridge.py +++ b/homeassistant/components/samsungtv/bridge.py @@ -1,10 +1,11 @@ """samsungctl and samsungtvws bridge classes.""" from abc import ABC, abstractmethod +import contextlib from samsungctl import Remote from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledResponse from samsungtvws import SamsungTVWS -from samsungtvws.exceptions import ConnectionFailure +from samsungtvws.exceptions import ConnectionFailure, HttpApiError from websocket import WebSocketException from homeassistant.const import ( @@ -25,8 +26,11 @@ from .const import ( RESULT_CANNOT_CONNECT, RESULT_NOT_SUPPORTED, RESULT_SUCCESS, + TIMEOUT_REQUEST, + TIMEOUT_WEBSOCKET, VALUE_CONF_ID, VALUE_CONF_NAME, + WEBSOCKET_PORTS, ) @@ -58,9 +62,14 @@ class SamsungTVBridge(ABC): def try_connect(self): """Try to connect to the TV.""" + @abstractmethod + def device_info(self): + """Try to gather infos of this TV.""" + def is_on(self): """Tells if the TV is on.""" - self.close_remote() + if self._remote: + self.close_remote() try: return self._get_remote() is not None @@ -104,7 +113,7 @@ class SamsungTVBridge(ABC): """Send the key.""" @abstractmethod - def _get_remote(self): + def _get_remote(self, avoid_open: bool = False): """Get Remote object.""" def close_remote(self): @@ -149,7 +158,7 @@ class SamsungTVLegacyBridge(SamsungTVBridge): CONF_METHOD: self.method, CONF_PORT: None, # We need this high timeout because waiting for auth popup is just an open socket - CONF_TIMEOUT: 31, + CONF_TIMEOUT: TIMEOUT_REQUEST, } try: LOGGER.debug("Try config: %s", config) @@ -162,11 +171,15 @@ class SamsungTVLegacyBridge(SamsungTVBridge): except UnhandledResponse: LOGGER.debug("Working but unsupported config: %s", config) return RESULT_NOT_SUPPORTED - except OSError as err: + except (ConnectionClosed, OSError) as err: LOGGER.debug("Failing config: %s, error: %s", config, err) return RESULT_CANNOT_CONNECT - def _get_remote(self): + def device_info(self): + """Try to gather infos of this device.""" + return None + + def _get_remote(self, avoid_open: bool = False): """Create or return a remote control instance.""" if self._remote is None: # We need to create a new instance to reconnect. @@ -184,6 +197,11 @@ class SamsungTVLegacyBridge(SamsungTVBridge): """Send the key using legacy protocol.""" self._get_remote().control(key) + def stop(self): + """Stop Bridge.""" + LOGGER.debug("Stopping SamsungRemote") + self.close_remote() + class SamsungTVWSBridge(SamsungTVBridge): """The Bridge for WebSocket TVs.""" @@ -196,14 +214,14 @@ class SamsungTVWSBridge(SamsungTVBridge): def try_connect(self): """Try to connect to the Websocket TV.""" - for self.port in (8001, 8002): + for self.port in WEBSOCKET_PORTS: config = { CONF_NAME: VALUE_CONF_NAME, CONF_HOST: self.host, CONF_METHOD: self.method, CONF_PORT: self.port, # We need this high timeout because waiting for auth popup is just an open socket - CONF_TIMEOUT: 31, + CONF_TIMEOUT: TIMEOUT_REQUEST, } result = None @@ -234,31 +252,46 @@ class SamsungTVWSBridge(SamsungTVBridge): return RESULT_CANNOT_CONNECT + def device_info(self): + """Try to gather infos of this TV.""" + remote = self._get_remote(avoid_open=True) + if not remote: + return None + with contextlib.suppress(HttpApiError): + return remote.rest_device_info() + def _send_key(self, key): """Send the key using websocket protocol.""" if key == "KEY_POWEROFF": key = "KEY_POWER" self._get_remote().send_key(key) - def _get_remote(self): + def _get_remote(self, avoid_open: bool = False): """Create or return a remote control instance.""" if self._remote is None: # We need to create a new instance to reconnect. try: - LOGGER.debug("Create SamsungTVWS") + LOGGER.debug( + "Create SamsungTVWS for %s (%s)", VALUE_CONF_NAME, self.host + ) self._remote = SamsungTVWS( host=self.host, port=self.port, token=self.token, - timeout=8, + timeout=TIMEOUT_WEBSOCKET, name=VALUE_CONF_NAME, ) - self._remote.open() + if not avoid_open: + self._remote.open() # This is only happening when the auth was switched to DENY # A removed auth will lead to socket timeout because waiting for auth popup is just an open socket except ConnectionFailure: self._notify_callback() - raise - except WebSocketException: + except (WebSocketException, OSError): self._remote = None return self._remote + + def stop(self): + """Stop Bridge.""" + LOGGER.debug("Stopping SamsungTVWS") + self.close_remote() diff --git a/homeassistant/components/samsungtv/config_flow.py b/homeassistant/components/samsungtv/config_flow.py index ed728b19eba..b45f6c5670b 100644 --- a/homeassistant/components/samsungtv/config_flow.py +++ b/homeassistant/components/samsungtv/config_flow.py @@ -4,7 +4,8 @@ from urllib.parse import urlparse import voluptuous as vol -from homeassistant import config_entries +from homeassistant import config_entries, data_entry_flow +from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS from homeassistant.components.ssdp import ( ATTR_SSDP_LOCATION, ATTR_UPNP_MANUFACTURER, @@ -13,59 +14,85 @@ from homeassistant.components.ssdp import ( ) from homeassistant.const import ( CONF_HOST, - CONF_ID, - CONF_IP_ADDRESS, + CONF_MAC, CONF_METHOD, CONF_NAME, CONF_PORT, CONF_TOKEN, ) +from homeassistant.core import callback +from homeassistant.helpers.device_registry import format_mac +from homeassistant.helpers.typing import DiscoveryInfoType from .bridge import SamsungTVBridge from .const import ( + ATTR_PROPERTIES, CONF_MANUFACTURER, CONF_MODEL, + DEFAULT_MANUFACTURER, DOMAIN, + LEGACY_PORT, LOGGER, METHOD_LEGACY, METHOD_WEBSOCKET, RESULT_AUTH_MISSING, RESULT_CANNOT_CONNECT, + RESULT_NOT_SUPPORTED, RESULT_SUCCESS, + RESULT_UNKNOWN_HOST, + WEBSOCKET_PORTS, ) DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME): str}) SUPPORTED_METHODS = [METHOD_LEGACY, METHOD_WEBSOCKET] -def _get_ip(host): - if host is None: - return None - return socket.gethostbyname(host) +def _get_device_info(host): + """Fetch device info by any websocket method.""" + for port in WEBSOCKET_PORTS: + bridge = SamsungTVBridge.get_bridge(METHOD_WEBSOCKET, host, port) + if info := bridge.device_info(): + return info + return None + + +async def async_get_device_info(hass, bridge, host): + """Fetch device info from bridge or websocket.""" + if bridge: + return await hass.async_add_executor_job(bridge.device_info) + + return await hass.async_add_executor_job(_get_device_info, host) + + +def _strip_uuid(udn): + return udn[5:] if udn.startswith("uuid:") else udn class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """Handle a Samsung TV config flow.""" - VERSION = 1 + VERSION = 2 def __init__(self): """Initialize flow.""" + self._reauth_entry = None self._host = None - self._ip = None + self._mac = None + self._udn = None self._manufacturer = None self._model = None self._name = None self._title = None self._id = None self._bridge = None + self._device_info = None - def _get_entry(self): + def _get_entry_from_bridge(self): + """Get device entry.""" data = { CONF_HOST: self._host, - CONF_ID: self._id, - CONF_IP_ADDRESS: self._ip, - CONF_MANUFACTURER: self._manufacturer, + CONF_MAC: self._mac, + CONF_MANUFACTURER: self._manufacturer or DEFAULT_MANUFACTURER, CONF_METHOD: self._bridge.method, CONF_MODEL: self._model, CONF_NAME: self._name, @@ -78,98 +105,205 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): data=data, ) + async def _async_set_device_unique_id(self, raise_on_progress=True): + """Set device unique_id.""" + await self._async_get_and_check_device_info() + await self._async_set_unique_id_from_udn(raise_on_progress) + + async def _async_set_unique_id_from_udn(self, raise_on_progress=True): + """Set the unique id from the udn.""" + await self.async_set_unique_id(self._udn, raise_on_progress=raise_on_progress) + self._async_update_existing_host_entry(self._host) + updates = {CONF_HOST: self._host} + if self._mac: + updates[CONF_MAC] = self._mac + self._abort_if_unique_id_configured(updates=updates) + def _try_connect(self): """Try to connect and check auth.""" for method in SUPPORTED_METHODS: self._bridge = SamsungTVBridge.get_bridge(method, self._host) result = self._bridge.try_connect() + if result == RESULT_SUCCESS: + return if result != RESULT_CANNOT_CONNECT: - return result + raise data_entry_flow.AbortFlow(result) LOGGER.debug("No working config found") - return RESULT_CANNOT_CONNECT + raise data_entry_flow.AbortFlow(RESULT_CANNOT_CONNECT) + + async def _async_get_and_check_device_info(self): + """Try to get the device info.""" + info = await async_get_device_info(self.hass, self._bridge, self._host) + if not info: + raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) + dev_info = info.get("device", {}) + device_type = dev_info.get("type") + if device_type != "Samsung SmartTV": + raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) + self._model = dev_info.get("modelName") + name = dev_info.get("name") + self._name = name.replace("[TV] ", "") if name else device_type + self._title = f"{self._name} ({self._model})" + self._udn = _strip_uuid(dev_info.get("udn", info["id"])) + if dev_info.get("networkType") == "wireless" and dev_info.get("wifiMac"): + self._mac = format_mac(dev_info.get("wifiMac")) + self._device_info = info async def async_step_import(self, user_input=None): """Handle configuration by yaml file.""" - return await self.async_step_user(user_input) + # We need to import even if we cannot validate + # since the TV may be off at startup + await self._async_set_name_host_from_input(user_input) + self._async_abort_entries_match({CONF_HOST: self._host}) + if user_input.get(CONF_PORT) in WEBSOCKET_PORTS: + user_input[CONF_METHOD] = METHOD_WEBSOCKET + else: + user_input[CONF_METHOD] = METHOD_LEGACY + user_input[CONF_PORT] = LEGACY_PORT + user_input[CONF_MANUFACTURER] = DEFAULT_MANUFACTURER + return self.async_create_entry( + title=self._title, + data=user_input, + ) + + async def _async_set_name_host_from_input(self, user_input): + try: + self._host = await self.hass.async_add_executor_job( + socket.gethostbyname, user_input[CONF_HOST] + ) + except socket.gaierror as err: + raise data_entry_flow.AbortFlow(RESULT_UNKNOWN_HOST) from err + self._name = user_input[CONF_NAME] + self._title = self._name 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._name = user_input.get(CONF_NAME) - self._title = self._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() + await self._async_set_name_host_from_input(user_input) + await self.hass.async_add_executor_job(self._try_connect) + self._async_abort_entries_match({CONF_HOST: self._host}) + if self._bridge.method != METHOD_LEGACY: + # Legacy bridge does not provide device info + await self._async_set_device_unique_id(raise_on_progress=False) + return self._get_entry_from_bridge() return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA) - async def async_step_ssdp(self, discovery_info): - """Handle a flow initialized by discovery.""" - host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname - ip_address = await self.hass.async_add_executor_job(_get_ip, host) + @callback + def _async_update_existing_host_entry(self, host): + for entry in self._async_current_entries(include_ignore=False): + if entry.data[CONF_HOST] != host: + continue + entry_kw_args = {} + if self.unique_id and entry.unique_id is None: + entry_kw_args["unique_id"] = self.unique_id + if self._mac and not entry.data.get(CONF_MAC): + data_copy = dict(entry.data) + data_copy[CONF_MAC] = self._mac + entry_kw_args["data"] = data_copy + if entry_kw_args: + self.hass.config_entries.async_update_entry(entry, **entry_kw_args) + return entry + return None + + async def _async_start_discovery_for_host(self, host): + """Start discovery for a host.""" + if entry := self._async_update_existing_host_entry(host): + if entry.unique_id: + # Let the flow continue to fill the missing + # unique id as we may be able to obtain it + # in the next step + raise data_entry_flow.AbortFlow("already_configured") + + self.context[CONF_HOST] = host + for progress in self._async_in_progress(): + if progress.get("context", {}).get(CONF_HOST) == host: + raise data_entry_flow.AbortFlow("already_in_progress") self._host = host - self._ip = self.context[CONF_IP_ADDRESS] = ip_address - self._manufacturer = discovery_info.get(ATTR_UPNP_MANUFACTURER) - self._model = discovery_info.get(ATTR_UPNP_MODEL_NAME) - self._name = f"Samsung {self._model}" - self._id = discovery_info.get(ATTR_UPNP_UDN) - self._title = self._model - # probably access denied - if self._id is None: - return self.async_abort(reason=RESULT_AUTH_MISSING) - if self._id.startswith("uuid:"): - self._id = self._id[5:] - - await self.async_set_unique_id(ip_address) - self._abort_if_unique_id_configured( - { - CONF_ID: self._id, - CONF_MANUFACTURER: self._manufacturer, - CONF_MODEL: self._model, - } + async def async_step_ssdp(self, discovery_info: DiscoveryInfoType): + """Handle a flow initialized by ssdp discovery.""" + self._udn = _strip_uuid(discovery_info[ATTR_UPNP_UDN]) + await self._async_set_unique_id_from_udn() + await self._async_start_discovery_for_host( + urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname ) + self._manufacturer = discovery_info[ATTR_UPNP_MANUFACTURER] + if not self._manufacturer or not self._manufacturer.lower().startswith( + "samsung" + ): + raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) + self._name = self._title = self._model = discovery_info.get( + ATTR_UPNP_MODEL_NAME + ) + self.context["title_placeholders"] = {"device": self._title} + return await self.async_step_confirm() - self.context["title_placeholders"] = {"model": self._model} + async def async_step_dhcp(self, discovery_info: DiscoveryInfoType): + """Handle a flow initialized by dhcp discovery.""" + self._mac = discovery_info[MAC_ADDRESS] + await self._async_start_discovery_for_host(discovery_info[IP_ADDRESS]) + await self._async_set_device_unique_id() + self.context["title_placeholders"] = {"device": self._title} + return await self.async_step_confirm() + + async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType): + """Handle a flow initialized by zeroconf discovery.""" + self._mac = format_mac(discovery_info[ATTR_PROPERTIES]["deviceid"]) + await self._async_start_discovery_for_host(discovery_info[CONF_HOST]) + await self._async_set_device_unique_id() + self.context["title_placeholders"] = {"device": self._title} 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() + await self.hass.async_add_executor_job(self._try_connect) + return self._get_entry_from_bridge() + self._set_confirm_only() return self.async_show_form( - step_id="confirm", description_placeholders={"model": self._model} + step_id="confirm", description_placeholders={"device": self._title} ) - async def async_step_reauth(self, user_input=None): + async def async_step_reauth(self, data): """Handle configuration by re-auth.""" - self._host = user_input[CONF_HOST] - self._id = user_input.get(CONF_ID) - self._ip = user_input[CONF_IP_ADDRESS] - self._manufacturer = user_input.get(CONF_MANUFACTURER) - self._model = user_input.get(CONF_MODEL) - self._name = user_input.get(CONF_NAME) - self._title = self._model or self._name + self._reauth_entry = self.hass.config_entries.async_get_entry( + self.context["entry_id"] + ) + data = self._reauth_entry.data + if data.get(CONF_MODEL) and data.get(CONF_NAME): + self._title = f"{data[CONF_NAME]} ({data[CONF_MODEL]})" + else: + self._title = data.get(CONF_NAME) or data[CONF_HOST] + return await self.async_step_reauth_confirm() - await self.async_set_unique_id(self._ip) - self.context["title_placeholders"] = {"model": self._title} + async def async_step_reauth_confirm(self, user_input=None): + """Confirm reauth.""" + errors = {} + if user_input is not None: + bridge = SamsungTVBridge.get_bridge( + self._reauth_entry.data[CONF_METHOD], self._reauth_entry.data[CONF_HOST] + ) + result = bridge.try_connect() + if result == RESULT_SUCCESS: + new_data = dict(self._reauth_entry.data) + new_data[CONF_TOKEN] = bridge.token + self.hass.config_entries.async_update_entry( + self._reauth_entry, data=new_data + ) + return self.async_abort(reason="reauth_successful") + if result not in (RESULT_AUTH_MISSING, RESULT_CANNOT_CONNECT): + return self.async_abort(reason=result) - return await self.async_step_confirm() + # On websocket we will get RESULT_CANNOT_CONNECT when auth is missing + errors = {"base": RESULT_AUTH_MISSING} + + self.context["title_placeholders"] = {"device": self._title} + return self.async_show_form( + step_id="reauth_confirm", + errors=errors, + description_placeholders={"device": self._title}, + ) diff --git a/homeassistant/components/samsungtv/const.py b/homeassistant/components/samsungtv/const.py index e043c74b347..f2571372b1f 100644 --- a/homeassistant/components/samsungtv/const.py +++ b/homeassistant/components/samsungtv/const.py @@ -4,7 +4,10 @@ import logging LOGGER = logging.getLogger(__package__) DOMAIN = "samsungtv" +ATTR_PROPERTIES = "properties" + DEFAULT_NAME = "Samsung TV" +DEFAULT_MANUFACTURER = "Samsung" VALUE_CONF_NAME = "HomeAssistant" VALUE_CONF_ID = "ha.component.samsung" @@ -18,6 +21,13 @@ RESULT_AUTH_MISSING = "auth_missing" RESULT_SUCCESS = "success" RESULT_CANNOT_CONNECT = "cannot_connect" RESULT_NOT_SUPPORTED = "not_supported" +RESULT_UNKNOWN_HOST = "unknown" METHOD_LEGACY = "legacy" METHOD_WEBSOCKET = "websocket" + +TIMEOUT_REQUEST = 31 +TIMEOUT_WEBSOCKET = 5 + +LEGACY_PORT = 55000 +WEBSOCKET_PORTS = (8002, 8001) diff --git a/homeassistant/components/samsungtv/manifest.json b/homeassistant/components/samsungtv/manifest.json index 81e08ddeaa6..4206aca7213 100644 --- a/homeassistant/components/samsungtv/manifest.json +++ b/homeassistant/components/samsungtv/manifest.json @@ -2,13 +2,27 @@ "domain": "samsungtv", "name": "Samsung Smart TV", "documentation": "https://www.home-assistant.io/integrations/samsungtv", - "requirements": ["samsungctl[websocket]==0.7.1", "samsungtvws==1.6.0"], + "requirements": [ + "samsungctl[websocket]==0.7.1", + "samsungtvws==1.6.0" + ], "ssdp": [ { "st": "urn:samsung.com:device:RemoteControlReceiver:1" } ], - "codeowners": ["@escoand"], + "zeroconf": [ + {"type":"_airplay._tcp.local.","manufacturer":"samsung*"} + ], + "dhcp": [ + { + "hostname": "tizen*" + } + ], + "codeowners": [ + "@escoand", + "@chemelli74" + ], "config_flow": true, "iot_class": "local_polling" } diff --git a/homeassistant/components/samsungtv/media_player.py b/homeassistant/components/samsungtv/media_player.py index a4b61369f99..72e21ed205c 100644 --- a/homeassistant/components/samsungtv/media_player.py +++ b/homeassistant/components/samsungtv/media_player.py @@ -19,22 +19,12 @@ from homeassistant.components.media_player.const import ( SUPPORT_VOLUME_STEP, ) from homeassistant.config_entries import SOURCE_REAUTH -from homeassistant.const import ( - CONF_HOST, - CONF_ID, - CONF_IP_ADDRESS, - CONF_METHOD, - CONF_NAME, - CONF_PORT, - CONF_TOKEN, - STATE_OFF, - STATE_ON, -) +from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, STATE_OFF, STATE_ON import homeassistant.helpers.config_validation as cv +from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC from homeassistant.helpers.script import Script from homeassistant.util import dt as dt_util -from .bridge import SamsungTVBridge from .const import ( CONF_MANUFACTURER, CONF_MODEL, @@ -60,41 +50,19 @@ SUPPORT_SAMSUNGTV = ( ) -async def async_setup_entry(hass, config_entry, async_add_entities): +async def async_setup_entry(hass, entry, async_add_entities): """Set up the Samsung TV from a config entry.""" - ip_address = config_entry.data[CONF_IP_ADDRESS] + bridge = hass.data[DOMAIN][entry.entry_id] + + host = entry.data[CONF_HOST] on_script = None - if ( - DOMAIN in hass.data - and ip_address in hass.data[DOMAIN] - and CONF_ON_ACTION in hass.data[DOMAIN][ip_address] - and hass.data[DOMAIN][ip_address][CONF_ON_ACTION] - ): - turn_on_action = hass.data[DOMAIN][ip_address][CONF_ON_ACTION] + data = hass.data[DOMAIN] + if turn_on_action := data.get(host, {}).get(CONF_ON_ACTION): on_script = Script( - hass, turn_on_action, config_entry.data.get(CONF_NAME, DEFAULT_NAME), DOMAIN + hass, turn_on_action, entry.data.get(CONF_NAME, DEFAULT_NAME), DOMAIN ) - # Initialize bridge - data = config_entry.data.copy() - bridge = SamsungTVBridge.get_bridge( - data[CONF_METHOD], - data[CONF_HOST], - data[CONF_PORT], - data.get(CONF_TOKEN), - ) - if bridge.port is None and bridge.default_port is not None: - # For backward compat, set default port for websocket tv - data[CONF_PORT] = bridge.default_port - hass.config_entries.async_update_entry(config_entry, data=data) - bridge = SamsungTVBridge.get_bridge( - data[CONF_METHOD], - data[CONF_HOST], - data[CONF_PORT], - data.get(CONF_TOKEN), - ) - - async_add_entities([SamsungTVDevice(bridge, config_entry, on_script)]) + async_add_entities([SamsungTVDevice(bridge, entry, on_script)], True) class SamsungTVDevice(MediaPlayerEntity): @@ -103,11 +71,12 @@ class SamsungTVDevice(MediaPlayerEntity): def __init__(self, bridge, config_entry, on_script): """Initialize the Samsung device.""" self._config_entry = config_entry + self._mac = config_entry.data.get(CONF_MAC) self._manufacturer = config_entry.data.get(CONF_MANUFACTURER) self._model = config_entry.data.get(CONF_MODEL) self._name = config_entry.data.get(CONF_NAME) self._on_script = on_script - self._uuid = config_entry.data.get(CONF_ID) + self._uuid = config_entry.unique_id # Assume that the TV is not muted self._muted = False # Assume that the TV is in Play mode @@ -117,21 +86,28 @@ class SamsungTVDevice(MediaPlayerEntity): # sending the next command to avoid turning the TV back ON). self._end_of_power_off = None self._bridge = bridge + self._auth_failed = False self._bridge.register_reauth_callback(self.access_denied) def access_denied(self): """Access denied callback.""" LOGGER.debug("Access denied in getting remote object") + self._auth_failed = True self.hass.add_job( self.hass.config_entries.flow.async_init( DOMAIN, - context={"source": SOURCE_REAUTH}, + context={ + "source": SOURCE_REAUTH, + "entry_id": self._config_entry.entry_id, + }, data=self._config_entry.data, ) ) def update(self): """Update state of device.""" + if self._auth_failed: + return if self._power_off_in_progress(): self._state = STATE_OFF else: @@ -165,15 +141,25 @@ class SamsungTVDevice(MediaPlayerEntity): """Return the state of the device.""" return self._state + @property + def available(self): + """Return the availability of the device.""" + if self._auth_failed: + return False + return self._state == STATE_ON or self._on_script + @property def device_info(self): """Return device specific attributes.""" - return { + info = { "name": self.name, "identifiers": {(DOMAIN, self.unique_id)}, "manufacturer": self._manufacturer, "model": self._model, } + if self._mac: + info["connections"] = {(CONNECTION_NETWORK_MAC, self._mac)} + return info @property def is_volume_muted(self): diff --git a/homeassistant/components/samsungtv/strings.json b/homeassistant/components/samsungtv/strings.json index 3854d040d3e..f92990e6163 100644 --- a/homeassistant/components/samsungtv/strings.json +++ b/homeassistant/components/samsungtv/strings.json @@ -1,6 +1,6 @@ { "config": { - "flow_title": "{model}", + "flow_title": "{device}", "step": { "user": { "description": "Enter your Samsung TV information. If you never connected Home Assistant before you should see a popup on your TV asking for authorization.", @@ -10,16 +10,24 @@ } }, "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 authorization. Manual configurations for this TV will be overwritten." - } + "description": "Do you want to set up {device}? If you never connected Home Assistant before you should see a popup on your TV asking for authorization." + }, + "reauth_confirm": { + "description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds." + } + }, + "error": { + "auth_missing": "[%key:component::samsungtv::config::abort::auth_missing%]" }, "abort": { "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]", "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", - "auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Please check your TV's settings to authorize Home Assistant.", + "auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Check your TV's External Device Manager settings to authorize Home Assistant.", + "id_missing": "This Samsung device doesn't have a SerialNumber.", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", - "not_supported": "This Samsung TV device is currently not supported." + "not_supported": "This Samsung device is currently not supported.", + "unknown": "[%key:common::config_flow::error::unknown%]", + "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]" } } -} +} \ No newline at end of file diff --git a/homeassistant/components/samsungtv/translations/en.json b/homeassistant/components/samsungtv/translations/en.json index 8f05775eb0c..8b48de950ee 100644 --- a/homeassistant/components/samsungtv/translations/en.json +++ b/homeassistant/components/samsungtv/translations/en.json @@ -3,15 +3,23 @@ "abort": { "already_configured": "Device is already configured", "already_in_progress": "Configuration flow is already in progress", - "auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Please check your TV's settings to authorize Home Assistant.", + "auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Check your TV's External Device Manager settings to authorize Home Assistant.", "cannot_connect": "Failed to connect", - "not_supported": "This Samsung TV device is currently not supported." + "id_missing": "This Samsung device doesn't have a SerialNumber.", + "not_supported": "This Samsung device is currently not supported.", + "reauth_successful": "Re-authentication was successful", + "unknown": "Unexpected error" }, - "flow_title": "{model}", + "error": { + "auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Check your TV's External Device Manager settings to authorize Home Assistant." + }, + "flow_title": "{device}", "step": { "confirm": { - "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 authorization. Manual configurations for this TV will be overwritten.", - "title": "Samsung TV" + "description": "Do you want to set up {device}? If you never connected Home Assistant before you should see a popup on your TV asking for authorization." + }, + "reauth_confirm": { + "description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds." }, "user": { "data": { diff --git a/homeassistant/generated/dhcp.py b/homeassistant/generated/dhcp.py index 2a592953123..7ea9d1c1992 100644 --- a/homeassistant/generated/dhcp.py +++ b/homeassistant/generated/dhcp.py @@ -171,6 +171,10 @@ DHCP = [ "hostname": "roomba-*", "macaddress": "80A589*" }, + { + "domain": "samsungtv", + "hostname": "tizen*" + }, { "domain": "screenlogic", "hostname": "pentair: *", diff --git a/homeassistant/generated/zeroconf.py b/homeassistant/generated/zeroconf.py index dce459e2083..014edc4b1f3 100644 --- a/homeassistant/generated/zeroconf.py +++ b/homeassistant/generated/zeroconf.py @@ -11,6 +11,12 @@ ZEROCONF = { "domain": "volumio" } ], + "_airplay._tcp.local.": [ + { + "domain": "samsungtv", + "manufacturer": "samsung*" + } + ], "_api._udp.local.": [ { "domain": "guardian" diff --git a/tests/components/samsungtv/__init__.py b/tests/components/samsungtv/__init__.py index 4ad1622c6ca..84328736822 100644 --- a/tests/components/samsungtv/__init__.py +++ b/tests/components/samsungtv/__init__.py @@ -1 +1,15 @@ """Tests for the samsungtv component.""" +from homeassistant.components.samsungtv.const import DOMAIN as SAMSUNGTV_DOMAIN +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from tests.common import MockConfigEntry + + +async def setup_samsungtv(hass: HomeAssistant, config: dict): + """Set up mock Samsung TV.""" + await async_setup_component(hass, "persistent_notification", {}) + entry = MockConfigEntry(domain=SAMSUNGTV_DOMAIN, data=config) + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() diff --git a/tests/components/samsungtv/conftest.py b/tests/components/samsungtv/conftest.py new file mode 100644 index 00000000000..278c6d7f18a --- /dev/null +++ b/tests/components/samsungtv/conftest.py @@ -0,0 +1,112 @@ +"""Fixtures for Samsung TV.""" +from unittest.mock import Mock, patch + +import pytest + +import homeassistant.util.dt as dt_util + +RESULT_ALREADY_CONFIGURED = "already_configured" +RESULT_ALREADY_IN_PROGRESS = "already_in_progress" + + +@pytest.fixture(name="remote") +def remote_fixture(): + """Patch the samsungctl Remote.""" + with patch( + "homeassistant.components.samsungtv.bridge.Remote" + ) as remote_class, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + remote = Mock() + remote.__enter__ = Mock() + remote.__exit__ = Mock() + remote_class.return_value = remote + yield remote + + +@pytest.fixture(name="remotews") +def remotews_fixture(): + """Patch the samsungtvws SamsungTVWS.""" + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWS" + ) as remotews_class, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + remotews = Mock() + remotews.__enter__ = Mock() + remotews.__exit__ = Mock() + remotews.rest_device_info.return_value = { + "id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4", + "device": { + "modelName": "82GXARRS", + "wifiMac": "aa:bb:cc:dd:ee:ff", + "name": "[TV] Living Room", + "type": "Samsung SmartTV", + "networkType": "wireless", + }, + } + remotews_class.return_value = remotews + remotews_class().__enter__().token = "FAKE_TOKEN" + yield remotews + + +@pytest.fixture(name="remotews_no_device_info") +def remotews_no_device_info_fixture(): + """Patch the samsungtvws SamsungTVWS.""" + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWS" + ) as remotews_class, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + remotews = Mock() + remotews.__enter__ = Mock() + remotews.__exit__ = Mock() + remotews.rest_device_info.return_value = None + remotews_class.return_value = remotews + remotews_class().__enter__().token = "FAKE_TOKEN" + yield remotews + + +@pytest.fixture(name="remotews_soundbar") +def remotews_soundbar_fixture(): + """Patch the samsungtvws SamsungTVWS.""" + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWS" + ) as remotews_class, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + remotews = Mock() + remotews.__enter__ = Mock() + remotews.__exit__ = Mock() + remotews.rest_device_info.return_value = { + "id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4", + "device": { + "modelName": "82GXARRS", + "wifiMac": "aa:bb:cc:dd:ee:ff", + "mac": "aa:bb:cc:dd:ee:ff", + "name": "[TV] Living Room", + "type": "Samsung SoundBar", + }, + } + remotews_class.return_value = remotews + remotews_class().__enter__().token = "FAKE_TOKEN" + yield remotews + + +@pytest.fixture(name="delay") +def delay_fixture(): + """Patch the delay script function.""" + with patch( + "homeassistant.components.samsungtv.media_player.Script.async_run" + ) as delay: + yield delay + + +@pytest.fixture +def mock_now(): + """Fixture for dtutil.now.""" + return dt_util.utcnow() diff --git a/tests/components/samsungtv/test_config_flow.py b/tests/components/samsungtv/test_config_flow.py index fb1b2a2bc67..5ac6caf40a9 100644 --- a/tests/components/samsungtv/test_config_flow.py +++ b/tests/components/samsungtv/test_config_flow.py @@ -1,16 +1,27 @@ """Tests for Samsung TV config flow.""" -from unittest.mock import DEFAULT as DEFAULT_MOCK, Mock, PropertyMock, call, patch +import socket +from unittest.mock import Mock, PropertyMock, call, patch -import pytest from samsungctl.exceptions import AccessDenied, UnhandledResponse from samsungtvws.exceptions import ConnectionFailure -from websocket import WebSocketProtocolException +from websocket import WebSocketException, WebSocketProtocolException from homeassistant import config_entries +from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS from homeassistant.components.samsungtv.const import ( + ATTR_PROPERTIES, CONF_MANUFACTURER, CONF_MODEL, + DEFAULT_MANUFACTURER, DOMAIN, + METHOD_LEGACY, + METHOD_WEBSOCKET, + RESULT_AUTH_MISSING, + RESULT_CANNOT_CONNECT, + RESULT_NOT_SUPPORTED, + RESULT_UNKNOWN_HOST, + TIMEOUT_REQUEST, + TIMEOUT_WEBSOCKET, ) from homeassistant.components.ssdp import ( ATTR_SSDP_LOCATION, @@ -19,22 +30,81 @@ from homeassistant.components.ssdp import ( ATTR_UPNP_MODEL_NAME, ATTR_UPNP_UDN, ) -from homeassistant.const import CONF_HOST, CONF_ID, CONF_METHOD, CONF_NAME, CONF_TOKEN +from homeassistant.const import ( + CONF_HOST, + CONF_ID, + CONF_IP_ADDRESS, + CONF_MAC, + CONF_METHOD, + CONF_NAME, + CONF_PORT, + CONF_TOKEN, +) +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component +from tests.common import MockConfigEntry +from tests.components.samsungtv.conftest import ( + RESULT_ALREADY_CONFIGURED, + RESULT_ALREADY_IN_PROGRESS, +) + +MOCK_IMPORT_DATA = { + CONF_HOST: "fake_host", + CONF_NAME: "fake", + CONF_PORT: 55000, +} +MOCK_IMPORT_WSDATA = { + CONF_HOST: "fake_host", + CONF_NAME: "fake", + CONF_PORT: 8002, +} 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_FRIENDLY_NAME: "[TV] fake_name", + ATTR_UPNP_MANUFACTURER: "Samsung fake_manufacturer", ATTR_UPNP_MODEL_NAME: "fake_model", - ATTR_UPNP_UDN: "uuid:fake_uuid", + ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172de", } 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_MANUFACTURER: "Samsung fake2_manufacturer", ATTR_UPNP_MODEL_NAME: "fake2_model", - ATTR_UPNP_UDN: "fake2_uuid", + ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172df", +} +MOCK_SSDP_DATA_WRONGMODEL = { + ATTR_SSDP_LOCATION: "http://fake2_host:12345/test", + ATTR_UPNP_FRIENDLY_NAME: "fake2_name", + ATTR_UPNP_MANUFACTURER: "fake2_manufacturer", + ATTR_UPNP_MODEL_NAME: "HW-Qfake", + ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172df", +} +MOCK_DHCP_DATA = {IP_ADDRESS: "fake_host", MAC_ADDRESS: "aa:bb:cc:dd:ee:ff"} +MOCK_ZEROCONF_DATA = { + CONF_HOST: "fake_host", + CONF_PORT: 1234, + ATTR_PROPERTIES: { + "deviceid": "aa:bb:cc:dd:ee:ff", + "manufacturer": "fake_manufacturer", + "model": "fake_model", + "serialNumber": "fake_serial", + }, +} +MOCK_OLD_ENTRY = { + CONF_HOST: "fake_host", + CONF_ID: "0d1cef00-00dc-1000-9c80-4844f7b172de_old", + CONF_IP_ADDRESS: "fake_ip_old", + CONF_METHOD: "legacy", + CONF_PORT: None, +} +MOCK_WS_ENTRY = { + CONF_HOST: "fake_host", + CONF_METHOD: METHOD_WEBSOCKET, + CONF_PORT: 8002, + CONF_MODEL: "any", + CONF_NAME: "any", } AUTODETECT_LEGACY = { @@ -44,62 +114,32 @@ AUTODETECT_LEGACY = { "method": "legacy", "port": None, "host": "fake_host", - "timeout": 31, + "timeout": TIMEOUT_REQUEST, } AUTODETECT_WEBSOCKET_PLAIN = { "host": "fake_host", "name": "HomeAssistant", "port": 8001, - "timeout": 31, + "timeout": TIMEOUT_REQUEST, "token": None, } AUTODETECT_WEBSOCKET_SSL = { "host": "fake_host", "name": "HomeAssistant", "port": 8002, - "timeout": 31, + "timeout": TIMEOUT_REQUEST, "token": None, } +DEVICEINFO_WEBSOCKET_SSL = { + "host": "fake_host", + "name": "HomeAssistant", + "port": 8002, + "timeout": TIMEOUT_WEBSOCKET, + "token": "123456789", +} -@pytest.fixture(name="remote") -def remote_fixture(): - """Patch the samsungctl Remote.""" - with patch( - "homeassistant.components.samsungtv.bridge.Remote" - ) as remote_class, patch( - "homeassistant.components.samsungtv.config_flow.socket" - ) as socket_class: - remote = Mock() - remote.__enter__ = Mock() - remote.__exit__ = Mock() - remote_class.return_value = remote - socket = Mock() - socket_class.return_value = socket - socket_class.gethostbyname.return_value = "FAKE_IP_ADDRESS" - yield remote - - -@pytest.fixture(name="remotews") -def remotews_fixture(): - """Patch the samsungtvws SamsungTVWS.""" - with patch( - "homeassistant.components.samsungtv.bridge.SamsungTVWS" - ) as remotews_class, patch( - "homeassistant.components.samsungtv.config_flow.socket" - ) as socket_class: - remotews = Mock() - remotews.__enter__ = Mock() - remotews.__exit__ = Mock() - remotews_class.return_value = remotews - remotews_class().__enter__().token = "FAKE_TOKEN" - socket = Mock() - socket_class.return_value = socket - socket_class.gethostbyname.return_value = "FAKE_IP_ADDRESS" - yield remotews - - -async def test_user_legacy(hass, remote): +async def test_user_legacy(hass: HomeAssistant, remote: Mock): """Test starting a flow by user.""" # show form result = await hass.config_entries.flow.async_init( @@ -118,12 +158,12 @@ async def test_user_legacy(hass, remote): assert result["data"][CONF_HOST] == "fake_host" assert result["data"][CONF_NAME] == "fake_name" assert result["data"][CONF_METHOD] == "legacy" - assert result["data"][CONF_MANUFACTURER] is None + assert result["data"][CONF_MANUFACTURER] == DEFAULT_MANUFACTURER assert result["data"][CONF_MODEL] is None - assert result["data"][CONF_ID] is None + assert result["result"].unique_id is None -async def test_user_websocket(hass, remotews): +async def test_user_websocket(hass: HomeAssistant, remotews: Mock): """Test starting a flow by user.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom") @@ -139,46 +179,46 @@ async def test_user_websocket(hass, remotews): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input=MOCK_USER_DATA ) - # legacy tv entry created + # websocket tv entry created assert result["type"] == "create_entry" - assert result["title"] == "fake_name" + assert result["title"] == "Living Room (82GXARRS)" assert result["data"][CONF_HOST] == "fake_host" - assert result["data"][CONF_NAME] == "fake_name" + assert result["data"][CONF_NAME] == "Living Room" assert result["data"][CONF_METHOD] == "websocket" - assert result["data"][CONF_MANUFACTURER] is None - assert result["data"][CONF_MODEL] is None - assert result["data"][CONF_ID] is None + assert result["data"][CONF_MANUFACTURER] == "Samsung" + assert result["data"][CONF_MODEL] == "82GXARRS" + assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -async def test_user_legacy_missing_auth(hass): +async def test_user_legacy_missing_auth(hass: HomeAssistant, remote: Mock): """Test starting a flow by user with authentication.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=AccessDenied("Boom"), - ), patch("homeassistant.components.samsungtv.config_flow.socket"): + ): # legacy device missing authentication result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "auth_missing" + assert result["reason"] == RESULT_AUTH_MISSING -async def test_user_legacy_not_supported(hass): +async def test_user_legacy_not_supported(hass: HomeAssistant, remote: Mock): """Test starting a flow by user for not supported device.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=UnhandledResponse("Boom"), - ), patch("homeassistant.components.samsungtv.config_flow.socket"): + ): # legacy device not supported result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "not_supported" + assert result["reason"] == RESULT_NOT_SUPPORTED -async def test_user_websocket_not_supported(hass): +async def test_user_websocket_not_supported(hass: HomeAssistant, remotews: Mock): """Test starting a flow by user for not supported device.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -186,18 +226,16 @@ async def test_user_websocket_not_supported(hass): ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWS", side_effect=WebSocketProtocolException("Boom"), - ), patch( - "homeassistant.components.samsungtv.config_flow.socket" ): # websocket device not supported result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "not_supported" + assert result["reason"] == RESULT_NOT_SUPPORTED -async def test_user_not_successful(hass): +async def test_user_not_successful(hass: HomeAssistant, remotews: Mock): """Test starting a flow by user but no connection found.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -205,17 +243,15 @@ async def test_user_not_successful(hass): ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWS", side_effect=OSError("Boom"), - ), patch( - "homeassistant.components.samsungtv.config_flow.socket" ): result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "cannot_connect" + assert result["reason"] == RESULT_CANNOT_CONNECT -async def test_user_not_successful_2(hass): +async def test_user_not_successful_2(hass: HomeAssistant, remotews: Mock): """Test starting a flow by user but no connection found.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -223,34 +259,15 @@ async def test_user_not_successful_2(hass): ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWS", side_effect=ConnectionFailure("Boom"), - ), patch( - "homeassistant.components.samsungtv.config_flow.socket" ): result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "cannot_connect" + assert result["reason"] == RESULT_CANNOT_CONNECT -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": config_entries.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": config_entries.SOURCE_USER}, data=MOCK_USER_DATA - ) - assert result["type"] == "abort" - assert result["reason"] == "already_configured" - - -async def test_ssdp(hass, remote): +async def test_ssdp(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery.""" # confirm to add the entry @@ -267,13 +284,13 @@ async def test_ssdp(hass, remote): assert result["type"] == "create_entry" assert result["title"] == "fake_model" assert result["data"][CONF_HOST] == "fake_host" - assert result["data"][CONF_NAME] == "Samsung fake_model" - assert result["data"][CONF_MANUFACTURER] == "fake_manufacturer" + assert result["data"][CONF_NAME] == "fake_model" + assert result["data"][CONF_MANUFACTURER] == "Samsung fake_manufacturer" assert result["data"][CONF_MODEL] == "fake_model" - assert result["data"][CONF_ID] == "fake_uuid" + assert result["result"].unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de" -async def test_ssdp_noprefix(hass, remote): +async def test_ssdp_noprefix(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery without prefixes.""" # confirm to add the entry @@ -292,18 +309,18 @@ async def test_ssdp_noprefix(hass, remote): assert result["type"] == "create_entry" assert result["title"] == "fake2_model" assert result["data"][CONF_HOST] == "fake2_host" - assert result["data"][CONF_NAME] == "Samsung fake2_model" - assert result["data"][CONF_MANUFACTURER] == "fake2_manufacturer" + assert result["data"][CONF_NAME] == "fake2_model" + assert result["data"][CONF_MANUFACTURER] == "Samsung fake2_manufacturer" assert result["data"][CONF_MODEL] == "fake2_model" - assert result["data"][CONF_ID] == "fake2_uuid" + assert result["result"].unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172df" -async def test_ssdp_legacy_missing_auth(hass): +async def test_ssdp_legacy_missing_auth(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery with authentication.""" with patch( "homeassistant.components.samsungtv.bridge.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( @@ -317,15 +334,15 @@ async def test_ssdp_legacy_missing_auth(hass): result["flow_id"], user_input="whatever" ) assert result["type"] == "abort" - assert result["reason"] == "auth_missing" + assert result["reason"] == RESULT_AUTH_MISSING -async def test_ssdp_legacy_not_supported(hass): +async def test_ssdp_legacy_not_supported(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery for not supported device.""" with patch( "homeassistant.components.samsungtv.bridge.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( @@ -339,10 +356,10 @@ async def test_ssdp_legacy_not_supported(hass): result["flow_id"], user_input="whatever" ) assert result["type"] == "abort" - assert result["reason"] == "not_supported" + assert result["reason"] == RESULT_NOT_SUPPORTED -async def test_ssdp_websocket_not_supported(hass): +async def test_ssdp_websocket_not_supported(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery for not supported device.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -350,8 +367,6 @@ async def test_ssdp_websocket_not_supported(hass): ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWS", side_effect=WebSocketProtocolException("Boom"), - ), patch( - "homeassistant.components.samsungtv.config_flow.socket" ): # confirm to add the entry result = await hass.config_entries.flow.async_init( @@ -365,10 +380,23 @@ async def test_ssdp_websocket_not_supported(hass): result["flow_id"], user_input="whatever" ) assert result["type"] == "abort" - assert result["reason"] == "not_supported" + assert result["reason"] == RESULT_NOT_SUPPORTED -async def test_ssdp_not_successful(hass): +async def test_ssdp_model_not_supported(hass: HomeAssistant, remote: Mock): + """Test starting a flow from discovery.""" + + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_WRONGMODEL, + ) + assert result["type"] == "abort" + assert result["reason"] == RESULT_NOT_SUPPORTED + + +async def test_ssdp_not_successful(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery but no device found.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -376,8 +404,6 @@ async def test_ssdp_not_successful(hass): ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWS", side_effect=OSError("Boom"), - ), patch( - "homeassistant.components.samsungtv.config_flow.socket" ): # confirm to add the entry @@ -392,10 +418,10 @@ async def test_ssdp_not_successful(hass): result["flow_id"], user_input="whatever" ) assert result["type"] == "abort" - assert result["reason"] == "cannot_connect" + assert result["reason"] == RESULT_CANNOT_CONNECT -async def test_ssdp_not_successful_2(hass): +async def test_ssdp_not_successful_2(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery but no device found.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -403,8 +429,6 @@ async def test_ssdp_not_successful_2(hass): ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWS", side_effect=ConnectionFailure("Boom"), - ), patch( - "homeassistant.components.samsungtv.config_flow.socket" ): # confirm to add the entry @@ -419,10 +443,10 @@ async def test_ssdp_not_successful_2(hass): result["flow_id"], user_input="whatever" ) assert result["type"] == "abort" - assert result["reason"] == "cannot_connect" + assert result["reason"] == RESULT_CANNOT_CONNECT -async def test_ssdp_already_in_progress(hass, remote): +async def test_ssdp_already_in_progress(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery twice.""" # confirm to add the entry @@ -437,10 +461,10 @@ async def test_ssdp_already_in_progress(hass, remote): DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA ) assert result["type"] == "abort" - assert result["reason"] == "already_in_progress" + assert result["reason"] == RESULT_ALREADY_IN_PROGRESS -async def test_ssdp_already_configured(hass, remote): +async def test_ssdp_already_configured(hass: HomeAssistant, remote: Mock): """Test starting a flow from discovery when already configured.""" # entry was added @@ -449,60 +473,204 @@ async def test_ssdp_already_configured(hass, remote): ) assert result["type"] == "create_entry" entry = result["result"] - assert entry.data[CONF_MANUFACTURER] is None + assert entry.data[CONF_MANUFACTURER] == DEFAULT_MANUFACTURER assert entry.data[CONF_MODEL] is None - assert entry.data[CONF_ID] is None + assert entry.unique_id is None # failed as already configured result2 = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA ) assert result2["type"] == "abort" - assert result2["reason"] == "already_configured" + assert result2["reason"] == RESULT_ALREADY_CONFIGURED # check updated device info - assert entry.data[CONF_MANUFACTURER] == "fake_manufacturer" - assert entry.data[CONF_MODEL] == "fake_model" - assert entry.data[CONF_ID] == "fake_uuid" + assert entry.unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de" -async def test_autodetect_websocket(hass, remote, remotews): - """Test for send key with autodetection of protocol.""" +async def test_import_legacy(hass: HomeAssistant): + """Test importing from yaml with hostname.""" with patch( - "homeassistant.components.samsungtv.bridge.Remote", - side_effect=OSError("Boom"), - ), patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remotews: - enter = Mock() - type(enter).token = PropertyMock(return_value="123456789") - remote = Mock() - remote.__enter__ = Mock(return_value=enter) - remote.__exit__ = Mock(return_value=False) - remotews.return_value = remote - + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA + DOMAIN, + context={"source": config_entries.SOURCE_IMPORT}, + data=MOCK_IMPORT_DATA, ) - assert result["type"] == "create_entry" - assert result["data"][CONF_METHOD] == "websocket" - assert result["data"][CONF_TOKEN] == "123456789" - assert remotews.call_count == 1 - assert remotews.call_args_list == [call(**AUTODETECT_WEBSOCKET_PLAIN)] + await hass.async_block_till_done() + assert result["type"] == "create_entry" + assert result["title"] == "fake" + assert result["data"][CONF_METHOD] == METHOD_LEGACY + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] == "fake" + assert result["data"][CONF_MANUFACTURER] == "Samsung" + assert result["result"].unique_id is None -async def test_autodetect_websocket_ssl(hass, remote, remotews): +async def test_import_websocket(hass: HomeAssistant): + """Test importing from yaml with hostname.""" + with patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_IMPORT}, + data=MOCK_IMPORT_WSDATA, + ) + await hass.async_block_till_done() + assert result["type"] == "create_entry" + assert result["title"] == "fake" + assert result["data"][CONF_METHOD] == METHOD_WEBSOCKET + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] == "fake" + assert result["data"][CONF_MANUFACTURER] == "Samsung" + assert result["result"].unique_id is None + + +async def test_import_unknown_host(hass: HomeAssistant, remotews: Mock): + """Test importing from yaml with hostname that does not resolve.""" + with patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + side_effect=socket.gaierror, + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_IMPORT}, + data=MOCK_IMPORT_DATA, + ) + await hass.async_block_till_done() + assert result["type"] == "abort" + assert result["reason"] == RESULT_UNKNOWN_HOST + + +async def test_dhcp(hass: HomeAssistant, remotews: Mock): + """Test starting a flow from dhcp.""" + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_DHCP}, + data=MOCK_DHCP_DATA, + ) + await hass.async_block_till_done() + 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"] == "Living Room (82GXARRS)" + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] == "Living Room" + assert result["data"][CONF_MAC] == "aa:bb:cc:dd:ee:ff" + assert result["data"][CONF_MANUFACTURER] == "Samsung" + assert result["data"][CONF_MODEL] == "82GXARRS" + assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +async def test_zeroconf(hass: HomeAssistant, remotews: Mock): + """Test starting a flow from zeroconf.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_ZEROCONF}, + data=MOCK_ZEROCONF_DATA, + ) + await hass.async_block_till_done() + 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"] == "Living Room (82GXARRS)" + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] == "Living Room" + assert result["data"][CONF_MAC] == "aa:bb:cc:dd:ee:ff" + assert result["data"][CONF_MANUFACTURER] == "Samsung" + assert result["data"][CONF_MODEL] == "82GXARRS" + assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +async def test_zeroconf_ignores_soundbar(hass: HomeAssistant, remotews_soundbar: Mock): + """Test starting a flow from zeroconf where the device is actually a soundbar.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_ZEROCONF}, + data=MOCK_ZEROCONF_DATA, + ) + await hass.async_block_till_done() + assert result["type"] == "abort" + assert result["reason"] == "not_supported" + + +async def test_zeroconf_no_device_info( + hass: HomeAssistant, remotews_no_device_info: Mock +): + """Test starting a flow from zeroconf where device_info returns None.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_ZEROCONF}, + data=MOCK_ZEROCONF_DATA, + ) + await hass.async_block_till_done() + assert result["type"] == "abort" + assert result["reason"] == "not_supported" + + +async def test_zeroconf_and_dhcp_same_time(hass: HomeAssistant, remotews: Mock): + """Test starting a flow from zeroconf and dhcp.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_DHCP}, + data=MOCK_DHCP_DATA, + ) + await hass.async_block_till_done() + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + result2 = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_ZEROCONF}, + data=MOCK_ZEROCONF_DATA, + ) + await hass.async_block_till_done() + assert result2["type"] == "abort" + assert result2["reason"] == "already_in_progress" + + +async def test_autodetect_websocket(hass: HomeAssistant, remote: Mock, remotews: Mock): """Test for send key with autodetection of protocol.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"), ), patch( - "homeassistant.components.samsungtv.bridge.SamsungTVWS", - side_effect=[WebSocketProtocolException("Boom"), DEFAULT_MOCK], + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ), patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWS" ) as remotews: enter = Mock() type(enter).token = PropertyMock(return_value="123456789") remote = Mock() remote.__enter__ = Mock(return_value=enter) remote.__exit__ = Mock(return_value=False) + remote.rest_device_info.return_value = { + "id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4", + "device": { + "modelName": "82GXARRS", + "wifiMac": "aa:bb:cc:dd:ee:ff", + "udn": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4", + "mac": "aa:bb:cc:dd:ee:ff", + "name": "[TV] Living Room", + "type": "Samsung SmartTV", + }, + } remotews.return_value = remote result = await hass.config_entries.flow.async_init( @@ -513,54 +681,61 @@ async def test_autodetect_websocket_ssl(hass, remote, remotews): assert result["data"][CONF_TOKEN] == "123456789" assert remotews.call_count == 2 assert remotews.call_args_list == [ - call(**AUTODETECT_WEBSOCKET_PLAIN), call(**AUTODETECT_WEBSOCKET_SSL), + call(**DEVICEINFO_WEBSOCKET_SSL), ] -async def test_autodetect_auth_missing(hass, remote): +async def test_autodetect_auth_missing(hass: HomeAssistant, remote: Mock): """Test for send key with autodetection of protocol.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=[AccessDenied("Boom")], - ) as remote: + ) as remote, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "auth_missing" + assert result["reason"] == RESULT_AUTH_MISSING assert remote.call_count == 1 assert remote.call_args_list == [call(AUTODETECT_LEGACY)] -async def test_autodetect_not_supported(hass, remote): +async def test_autodetect_not_supported(hass: HomeAssistant, remote: Mock): """Test for send key with autodetection of protocol.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=[UnhandledResponse("Boom")], - ) as remote: + ) as remote, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "not_supported" + assert result["reason"] == RESULT_NOT_SUPPORTED assert remote.call_count == 1 assert remote.call_args_list == [call(AUTODETECT_LEGACY)] -async def test_autodetect_legacy(hass, remote): +async def test_autodetect_legacy(hass: HomeAssistant, remote: Mock): """Test for send key with autodetection of protocol.""" with patch("homeassistant.components.samsungtv.bridge.Remote") as remote: result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) + print(result) assert result["type"] == "create_entry" assert result["data"][CONF_METHOD] == "legacy" assert remote.call_count == 1 assert remote.call_args_list == [call(AUTODETECT_LEGACY)] -async def test_autodetect_none(hass, remote, remotews): +async def test_autodetect_none(hass: HomeAssistant, remote: Mock, remotews: Mock): """Test for send key with autodetection of protocol.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -568,18 +743,228 @@ async def test_autodetect_none(hass, remote, remotews): ) as remote, patch( "homeassistant.components.samsungtv.bridge.SamsungTVWS", side_effect=OSError("Boom"), - ) as remotews: + ) as remotews, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) assert result["type"] == "abort" - assert result["reason"] == "cannot_connect" + assert result["reason"] == RESULT_CANNOT_CONNECT assert remote.call_count == 1 assert remote.call_args_list == [ call(AUTODETECT_LEGACY), ] assert remotews.call_count == 2 assert remotews.call_args_list == [ - call(**AUTODETECT_WEBSOCKET_PLAIN), call(**AUTODETECT_WEBSOCKET_SSL), + call(**AUTODETECT_WEBSOCKET_PLAIN), ] + + +async def test_update_old_entry(hass: HomeAssistant, remote: Mock): + """Test update of old entry.""" + with patch("homeassistant.components.samsungtv.bridge.Remote") as remote: + remote().rest_device_info.return_value = { + "device": { + "modelName": "fake_model2", + "name": "[TV] Fake Name", + "udn": "uuid:fake_serial", + } + } + + entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY) + entry.add_to_hass(hass) + + config_entries_domain = hass.config_entries.async_entries(DOMAIN) + assert len(config_entries_domain) == 1 + assert entry is config_entries_domain[0] + assert entry.data[CONF_ID] == "0d1cef00-00dc-1000-9c80-4844f7b172de_old" + assert entry.data[CONF_IP_ADDRESS] == "fake_ip_old" + assert not entry.unique_id + + assert await async_setup_component(hass, DOMAIN, {}) is True + await hass.async_block_till_done() + + # failed as already configured + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "abort" + assert result["reason"] == RESULT_ALREADY_CONFIGURED + + config_entries_domain = hass.config_entries.async_entries(DOMAIN) + assert len(config_entries_domain) == 1 + entry2 = config_entries_domain[0] + + # check updated device info + assert entry2.data.get(CONF_ID) is not None + assert entry2.data.get(CONF_IP_ADDRESS) is not None + assert entry2.unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de" + + +async def test_update_missing_mac_unique_id_added_from_dhcp(hass, remotews: Mock): + """Test missing mac and unique id added.""" + entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY, unique_id=None) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_DHCP}, + data=MOCK_DHCP_DATA, + ) + await hass.async_block_till_done() + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:cc:dd:ee:ff" + assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +async def test_update_missing_mac_unique_id_added_from_zeroconf(hass, remotews: Mock): + """Test missing mac and unique id added.""" + entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY, unique_id=None) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_ZEROCONF}, + data=MOCK_ZEROCONF_DATA, + ) + await hass.async_block_till_done() + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:cc:dd:ee:ff" + assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +async def test_update_missing_mac_added_unique_id_preserved_from_zeroconf( + hass, remotews: Mock +): + """Test missing mac and unique id added.""" + entry = MockConfigEntry( + domain=DOMAIN, + data=MOCK_OLD_ENTRY, + unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de", + ) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_ZEROCONF}, + data=MOCK_ZEROCONF_DATA, + ) + await hass.async_block_till_done() + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:cc:dd:ee:ff" + assert entry.unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de" + + +async def test_form_reauth_legacy(hass, remote: Mock): + """Test reauthenticate legacy.""" + entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"entry_id": entry.entry_id, "source": config_entries.SOURCE_REAUTH}, + data=entry.data, + ) + assert result["type"] == "form" + assert result["errors"] == {} + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {}, + ) + await hass.async_block_till_done() + assert result2["type"] == "abort" + assert result2["reason"] == "reauth_successful" + + +async def test_form_reauth_websocket(hass, remotews: Mock): + """Test reauthenticate websocket.""" + entry = MockConfigEntry(domain=DOMAIN, data=MOCK_WS_ENTRY) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"entry_id": entry.entry_id, "source": config_entries.SOURCE_REAUTH}, + data=entry.data, + ) + assert result["type"] == "form" + assert result["errors"] == {} + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {}, + ) + await hass.async_block_till_done() + assert result2["type"] == "abort" + assert result2["reason"] == "reauth_successful" + + +async def test_form_reauth_websocket_cannot_connect(hass, remotews: Mock): + """Test reauthenticate websocket when we cannot connect on the first attempt.""" + entry = MockConfigEntry(domain=DOMAIN, data=MOCK_WS_ENTRY) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"entry_id": entry.entry_id, "source": config_entries.SOURCE_REAUTH}, + data=entry.data, + ) + assert result["type"] == "form" + assert result["errors"] == {} + + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWS", + side_effect=ConnectionFailure, + ), patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {}, + ) + await hass.async_block_till_done() + + import pprint + + pprint.pprint(result2) + assert result2["type"] == "form" + assert result2["errors"] == {"base": RESULT_AUTH_MISSING} + + result3 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {}, + ) + await hass.async_block_till_done() + + assert result3["type"] == "abort" + assert result3["reason"] == "reauth_successful" + + +async def test_form_reauth_websocket_not_supported(hass, remotews: Mock): + """Test reauthenticate websocket when the device is not supported.""" + entry = MockConfigEntry(domain=DOMAIN, data=MOCK_WS_ENTRY) + entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"entry_id": entry.entry_id, "source": config_entries.SOURCE_REAUTH}, + data=entry.data, + ) + assert result["type"] == "form" + assert result["errors"] == {} + + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWS", + side_effect=WebSocketException, + ), patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {}, + ) + await hass.async_block_till_done() + + assert result2["type"] == "abort" + assert result2["reason"] == "not_supported" diff --git a/tests/components/samsungtv/test_init.py b/tests/components/samsungtv/test_init.py index bb19f120cf6..f728fd4af10 100644 --- a/tests/components/samsungtv/test_init.py +++ b/tests/components/samsungtv/test_init.py @@ -1,21 +1,22 @@ """Tests for the Samsung TV Integration.""" from unittest.mock import Mock, 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, + METHOD_WEBSOCKET, ) from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV from homeassistant.const import ( ATTR_ENTITY_ID, ATTR_SUPPORTED_FEATURES, CONF_HOST, + CONF_METHOD, CONF_NAME, SERVICE_VOLUME_UP, ) +from homeassistant.core import HomeAssistant from homeassistant.setup import async_setup_component ENTITY_ID = f"{DOMAIN}.fake_name" @@ -25,6 +26,7 @@ MOCK_CONFIG = { CONF_HOST: "fake_host", CONF_NAME: "fake_name", CONF_ON_ACTION: [{"delay": "00:00:01"}], + CONF_METHOD: METHOD_WEBSOCKET, } ] } @@ -32,37 +34,22 @@ REMOTE_CALL = { "name": "HomeAssistant", "description": "HomeAssistant", "id": "ha.component.samsung", - "method": "legacy", "host": MOCK_CONFIG[SAMSUNGTV_DOMAIN][0][CONF_HOST], + "method": "legacy", "port": None, "timeout": 1, } -@pytest.fixture(name="remote") -def remote_fixture(): - """Patch the samsungctl Remote.""" - with patch( - "homeassistant.components.samsungtv.bridge.Remote" - ) as remote_class, patch( - "homeassistant.components.samsungtv.config_flow.socket" - ) as socket1, patch( - "homeassistant.components.samsungtv.socket" - ) as socket2: - remote = Mock() - remote.__enter__ = Mock() - remote.__exit__ = Mock() - remote_class.return_value = remote - socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS" - socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS" - yield remote - - -async def test_setup(hass, remote): +async def test_setup(hass: HomeAssistant, remote: Mock): """Test Samsung TV integration is setup.""" - with patch("homeassistant.components.samsungtv.bridge.Remote") as remote: - await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) - await hass.async_block_till_done() + with patch("homeassistant.components.samsungtv.bridge.Remote") as remote, patch( + "homeassistant.components.samsungtv.config_flow.socket.gethostbyname", + return_value="fake_host", + ): + with patch("homeassistant.components.samsungtv.bridge.Remote") as remote: + 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 @@ -80,7 +67,7 @@ async def test_setup(hass, remote): assert remote.call_args == call(REMOTE_CALL) -async def test_setup_duplicate_config(hass, remote, caplog): +async def test_setup_duplicate_config(hass: HomeAssistant, remote: Mock, caplog): """Test duplicate setup of platform.""" DUPLICATE = { SAMSUNGTV_DOMAIN: [ @@ -95,7 +82,7 @@ async def test_setup_duplicate_config(hass, remote, caplog): assert "duplicate host entries found" in caplog.text -async def test_setup_duplicate_entries(hass, remote, caplog): +async def test_setup_duplicate_entries(hass: HomeAssistant, remote: Mock, caplog): """Test duplicate setup of platform.""" await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) await hass.async_block_till_done() diff --git a/tests/components/samsungtv/test_media_player.py b/tests/components/samsungtv/test_media_player.py index 6415f02cdf5..0cf54e32807 100644 --- a/tests/components/samsungtv/test_media_player.py +++ b/tests/components/samsungtv/test_media_player.py @@ -25,6 +25,7 @@ from homeassistant.components.media_player.const import ( from homeassistant.components.samsungtv.const import ( CONF_ON_ACTION, DOMAIN as SAMSUNGTV_DOMAIN, + TIMEOUT_WEBSOCKET, ) from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV from homeassistant.const import ( @@ -37,10 +38,12 @@ from homeassistant.const import ( CONF_METHOD, CONF_NAME, CONF_PORT, + CONF_TIMEOUT, CONF_TOKEN, SERVICE_MEDIA_NEXT_TRACK, SERVICE_MEDIA_PAUSE, SERVICE_MEDIA_PLAY, + SERVICE_MEDIA_PLAY_PAUSE, SERVICE_MEDIA_PREVIOUS_TRACK, SERVICE_TURN_OFF, SERVICE_TURN_ON, @@ -49,6 +52,7 @@ from homeassistant.const import ( SERVICE_VOLUME_UP, STATE_OFF, STATE_ON, + STATE_UNAVAILABLE, ) from homeassistant.setup import async_setup_component import homeassistant.util.dt as dt_util @@ -59,7 +63,7 @@ ENTITY_ID = f"{DOMAIN}.fake" MOCK_CONFIG = { SAMSUNGTV_DOMAIN: [ { - CONF_HOST: "fake", + CONF_HOST: "fake_host", CONF_NAME: "fake", CONF_PORT: 55000, CONF_ON_ACTION: [{"delay": "00:00:01"}], @@ -69,7 +73,7 @@ MOCK_CONFIG = { MOCK_CONFIGWS = { SAMSUNGTV_DOMAIN: [ { - CONF_HOST: "fake", + CONF_HOST: "fake_host", CONF_NAME: "fake", CONF_PORT: 8001, CONF_TOKEN: "123456789", @@ -78,27 +82,20 @@ MOCK_CONFIGWS = { ] } MOCK_CALLS_WS = { - "host": "fake", - "port": 8001, - "token": None, - "timeout": 31, - "name": "HomeAssistant", + CONF_HOST: "fake_host", + CONF_PORT: 8001, + CONF_TOKEN: "123456789", + CONF_TIMEOUT: TIMEOUT_WEBSOCKET, + CONF_NAME: "HomeAssistant", } MOCK_ENTRY_WS = { CONF_IP_ADDRESS: "test", - CONF_HOST: "fake", + CONF_HOST: "fake_host", CONF_METHOD: "websocket", CONF_NAME: "fake", CONF_PORT: 8001, - CONF_TOKEN: "abcde", -} -MOCK_CALLS_ENTRY_WS = { - "host": "fake", - "name": "HomeAssistant", - "port": 8001, - "timeout": 8, - "token": "abcde", + CONF_TOKEN: "123456789", } ENTITY_ID_NOTURNON = f"{DOMAIN}.fake_noturnon" @@ -109,45 +106,6 @@ MOCK_CONFIG_NOTURNON = { } -@pytest.fixture(name="remote") -def remote_fixture(): - """Patch the samsungctl Remote.""" - with patch( - "homeassistant.components.samsungtv.bridge.Remote" - ) as remote_class, patch( - "homeassistant.components.samsungtv.config_flow.socket" - ) as socket1, patch( - "homeassistant.components.samsungtv.socket" - ) as socket2: - remote = Mock() - remote.__enter__ = Mock() - remote.__exit__ = Mock() - remote_class.return_value = remote - socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS" - socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS" - yield remote - - -@pytest.fixture(name="remotews") -def remotews_fixture(): - """Patch the samsungtvws SamsungTVWS.""" - with patch( - "homeassistant.components.samsungtv.bridge.SamsungTVWS" - ) as remote_class, patch( - "homeassistant.components.samsungtv.config_flow.socket" - ) as socket1, patch( - "homeassistant.components.samsungtv.socket" - ) as socket2: - remote = Mock() - remote.__enter__ = Mock() - remote.__exit__ = Mock() - remote_class.return_value = remote - remote_class().__enter__().token = "FAKE_TOKEN" - socket1.gethostbyname.return_value = "FAKE_IP_ADDRESS" - socket2.gethostbyname.return_value = "FAKE_IP_ADDRESS" - yield remote - - @pytest.fixture(name="delay") def delay_fixture(): """Patch the delay script function.""" @@ -226,7 +184,7 @@ async def test_setup_websocket_2(hass, mock_now): state = hass.states.get(entity_id) assert state assert remote.call_count == 1 - assert remote.call_args_list == [call(**MOCK_CALLS_ENTRY_WS)] + assert remote.call_args_list == [call(**MOCK_CALLS_WS)] async def test_update_on(hass, remote, mock_now): @@ -272,12 +230,18 @@ async def test_update_access_denied(hass, remote, mock_now): with patch("homeassistant.util.dt.utcnow", return_value=next_update): async_fire_time_changed(hass, next_update) await hass.async_block_till_done() + next_update = mock_now + timedelta(minutes=10) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() assert [ flow for flow in hass.config_entries.flow.async_progress() if flow["context"]["source"] == "reauth" ] + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_UNAVAILABLE async def test_update_connection_failure(hass, remotews, mock_now): @@ -296,12 +260,18 @@ async def test_update_connection_failure(hass, remotews, mock_now): with patch("homeassistant.util.dt.utcnow", return_value=next_update): async_fire_time_changed(hass, next_update) await hass.async_block_till_done() + next_update = mock_now + timedelta(minutes=10) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() assert [ flow for flow in hass.config_entries.flow.async_progress() if flow["context"]["source"] == "reauth" ] + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_UNAVAILABLE async def test_update_unhandled_response(hass, remote, mock_now): @@ -438,7 +408,8 @@ async def test_state_without_turnon(hass, remote): DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID_NOTURNON}, True ) state = hass.states.get(ENTITY_ID_NOTURNON) - assert state.state == STATE_OFF + # Should be STATE_UNAVAILABLE since there is no way to turn it back on + assert state.state == STATE_UNAVAILABLE async def test_supported_features_with_turnon(hass, remote): @@ -555,6 +526,15 @@ async def test_media_play(hass, remote): assert remote.close.call_count == 1 assert remote.close.call_args_list == [call()] + assert await hass.services.async_call( + DOMAIN, SERVICE_MEDIA_PLAY_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_PLAY"), call("KEY_PAUSE")] + assert remote.close.call_count == 2 + assert remote.close.call_args_list == [call(), call()] + async def test_media_pause(hass, remote): """Test for media_pause.""" @@ -568,6 +548,15 @@ async def test_media_pause(hass, remote): assert remote.close.call_count == 1 assert remote.close.call_args_list == [call()] + assert await hass.services.async_call( + DOMAIN, SERVICE_MEDIA_PLAY_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_PLAY")] + assert remote.close.call_count == 2 + assert remote.close.call_args_list == [call(), call()] + async def test_media_next_track(hass, remote): """Test for media_next_track."""