From 87378016c14f681a97df4e212832d29f14efd460 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Tue, 22 Mar 2022 11:11:41 +0100 Subject: [PATCH] Add basic support for SamsungTV encrypted models (#68500) Co-authored-by: epenet --- .../components/samsungtv/__init__.py | 28 ++- homeassistant/components/samsungtv/bridge.py | 212 +++++++++++++++++- .../components/samsungtv/config_flow.py | 58 ++++- homeassistant/components/samsungtv/const.py | 4 + .../components/samsungtv/diagnostics.py | 4 +- .../components/samsungtv/media_player.py | 19 +- .../components/samsungtv/strings.json | 8 +- .../components/samsungtv/translations/en.json | 8 +- tests/components/samsungtv/__init__.py | 22 ++ tests/components/samsungtv/conftest.py | 27 +++ tests/components/samsungtv/const.py | 54 +++++ .../components/samsungtv/test_config_flow.py | 126 ++++++++++- .../components/samsungtv/test_diagnostics.py | 102 +++++++-- tests/components/samsungtv/test_init.py | 3 + .../components/samsungtv/test_media_player.py | 149 +++++++++++- 15 files changed, 766 insertions(+), 58 deletions(-) diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py index 4ce2b2350d4..470d7e80584 100644 --- a/homeassistant/components/samsungtv/__init__.py +++ b/homeassistant/components/samsungtv/__init__.py @@ -17,11 +17,12 @@ from homeassistant.const import ( CONF_METHOD, CONF_NAME, CONF_PORT, + CONF_TOKEN, EVENT_HOMEASSISTANT_STOP, Platform, ) from homeassistant.core import Event, HomeAssistant, callback -from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady import homeassistant.helpers.config_validation as cv from homeassistant.helpers.typing import ConfigType @@ -29,10 +30,12 @@ from .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info from .const import ( CONF_MODEL, CONF_ON_ACTION, + CONF_SESSION_ID, DEFAULT_NAME, DOMAIN, LEGACY_PORT, LOGGER, + METHOD_ENCRYPTED_WEBSOCKET, METHOD_LEGACY, ) @@ -110,6 +113,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up the Samsung TV platform.""" # Initialize bridge + if entry.data.get(CONF_METHOD) == METHOD_ENCRYPTED_WEBSOCKET: + if not entry.data.get(CONF_TOKEN) or not entry.data.get(CONF_SESSION_ID): + raise ConfigEntryAuthFailed( + "Token and session id are required in encrypted mode" + ) bridge = await _async_create_bridge_with_updated_data(hass, entry) # Ensure updates get saved against the config_entry @@ -120,6 +128,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: bridge.register_update_config_entry_callback(_update_config_entry) + # Allow bridge to force the reload of the config_entry + @callback + def _reload_config_entry() -> None: + """Update config entry with the new token.""" + hass.async_create_task(hass.config_entries.async_reload(entry.entry_id)) + + bridge.register_reload_callback(_reload_config_entry) + async def stop_bridge(event: Event) -> None: """Stop SamsungTV bridge connection.""" LOGGER.debug("Stopping SamsungTVBridge %s", bridge.host) @@ -134,6 +150,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return True +def _model_requires_encryption(model: str | None) -> bool: + """H and J models need pairing with PIN.""" + return model is not None and len(model) > 4 and model[4] in ("H", "J") + + async def _async_create_bridge_with_updated_data( hass: HomeAssistant, entry: ConfigEntry ) -> SamsungTVBridge: @@ -194,12 +215,13 @@ async def _async_create_bridge_with_updated_data( LOGGER.info("Updated model to %s for %s", model, host) updated_data[CONF_MODEL] = model - if model and len(model) > 4 and model[4] in ("H", "J"): + if _model_requires_encryption(model) and method != METHOD_ENCRYPTED_WEBSOCKET: LOGGER.info( "Detected model %s for %s. Some televisions from H and J series use " - "an encrypted protocol that may not be supported in this integration", + "an encrypted protocol but you are using %s which may not be supported", model, host, + method, ) if updated_data: diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py index e9036d1aa59..b6f9aebd86b 100644 --- a/homeassistant/components/samsungtv/bridge.py +++ b/homeassistant/components/samsungtv/bridge.py @@ -4,7 +4,7 @@ from __future__ import annotations from abc import ABC, abstractmethod import asyncio from asyncio.exceptions import TimeoutError as AsyncioTimeoutError -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterable, Mapping import contextlib from typing import Any, cast @@ -13,6 +13,7 @@ from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledRespo from samsungtvws.async_remote import SamsungTVWSAsyncRemote from samsungtvws.async_rest import SamsungTVAsyncRest from samsungtvws.command import SamsungTVCommand +from samsungtvws.encrypted.remote import SamsungTVEncryptedWSAsyncRemote from samsungtvws.event import MS_ERROR_EVENT from samsungtvws.exceptions import ConnectionFailure, HttpApiError from samsungtvws.remote import ChannelEmitCommand, SendRemoteKey @@ -28,13 +29,17 @@ from homeassistant.const import ( CONF_TOKEN, ) from homeassistant.core import CALLBACK_TYPE, HomeAssistant +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.device_registry import format_mac from .const import ( CONF_DESCRIPTION, + CONF_SESSION_ID, + ENCRYPTED_WEBSOCKET_PORT, LEGACY_PORT, LOGGER, + METHOD_ENCRYPTED_WEBSOCKET, METHOD_LEGACY, METHOD_WEBSOCKET, RESULT_AUTH_MISSING, @@ -64,19 +69,31 @@ async def async_get_device_info( host: str, ) -> tuple[int | None, str | None, dict[str, Any] | None]: """Fetch the port, method, and device info.""" + # Bridge is defined if bridge and bridge.port: return bridge.port, bridge.method, await bridge.async_device_info() + # Try websocket ports for port in WEBSOCKET_PORTS: bridge = SamsungTVBridge.get_bridge(hass, METHOD_WEBSOCKET, host, port) if info := await bridge.async_device_info(): return port, METHOD_WEBSOCKET, info + # Try encrypted websocket port + bridge = SamsungTVBridge.get_bridge( + hass, METHOD_ENCRYPTED_WEBSOCKET, host, ENCRYPTED_WEBSOCKET_PORT + ) + result = await bridge.async_try_connect() + if result == RESULT_SUCCESS: + return port, METHOD_ENCRYPTED_WEBSOCKET, await bridge.async_device_info() + + # Try legacy port bridge = SamsungTVBridge.get_bridge(hass, METHOD_LEGACY, host, LEGACY_PORT) result = await bridge.async_try_connect() if result in (RESULT_SUCCESS, RESULT_AUTH_MISSING): return LEGACY_PORT, METHOD_LEGACY, await bridge.async_device_info() + # Failed to get info return None, None, None @@ -94,6 +111,8 @@ class SamsungTVBridge(ABC): """Get Bridge instance.""" if method == METHOD_LEGACY or port == LEGACY_PORT: return SamsungTVLegacyBridge(hass, method, host, port) + if method == METHOD_ENCRYPTED_WEBSOCKET or port == ENCRYPTED_WEBSOCKET_PORT: + return SamsungTVEncryptedBridge(hass, method, host, port, entry_data) return SamsungTVWSBridge(hass, method, host, port, entry_data) def __init__( @@ -105,13 +124,19 @@ class SamsungTVBridge(ABC): self.method = method self.host = host self.token: str | None = None + self.session_id: str | None = None self._reauth_callback: CALLBACK_TYPE | None = None + self._reload_callback: CALLBACK_TYPE | None = None self._update_config_entry: Callable[[Mapping[str, Any]], None] | None = None def register_reauth_callback(self, func: CALLBACK_TYPE) -> None: """Register a callback function.""" self._reauth_callback = func + def register_reload_callback(self, func: CALLBACK_TYPE) -> None: + """Register a callback function.""" + self._reload_callback = func + def register_update_config_entry_callback( self, func: Callable[[Mapping[str, Any]], None] ) -> None: @@ -140,7 +165,7 @@ class SamsungTVBridge(ABC): @abstractmethod async def async_power_off(self) -> None: - """Send power off command to remote.""" + """Send power off command to remote and close.""" @abstractmethod async def async_close_remote(self) -> None: @@ -151,6 +176,11 @@ class SamsungTVBridge(ABC): if self._reauth_callback is not None: self._reauth_callback() + def _notify_reload_callback(self) -> None: + """Notify reload callback.""" + if self._reload_callback is not None: + self._reload_callback() + def _notify_update_config_entry(self, updates: Mapping[str, Any]) -> None: """Notify update config callback.""" if self._update_config_entry is not None: @@ -348,6 +378,7 @@ class SamsungTVWSBridge(SamsungTVBridge): # Ensure we get an updated value info = await self.async_device_info() return info is not None and info["device"]["PowerState"] == "on" + LOGGER.debug("Checking if TV %s is on using websocket", self.host) if remote := await self._async_get_remote(): return remote.is_alive() @@ -406,7 +437,7 @@ class SamsungTVWSBridge(SamsungTVBridge): """Try to gather infos of this TV.""" if self._rest_api is None: assert self.port - self._rest_api = SamsungTVAsyncRest( + rest_api = SamsungTVAsyncRest( host=self.host, session=async_get_clientsession(self.hass), port=self.port, @@ -414,7 +445,7 @@ class SamsungTVWSBridge(SamsungTVBridge): ) with contextlib.suppress(HttpApiError, AsyncioTimeoutError): - device_info: dict[str, Any] = await self._rest_api.rest_device_info() + device_info: dict[str, Any] = await rest_api.rest_device_info() LOGGER.debug("Device info on %s is: %s", self.host, device_info) self._device_info = device_info return device_info @@ -513,8 +544,7 @@ class SamsungTVWSBridge(SamsungTVBridge): self._notify_update_config_entry({CONF_TOKEN: self.token}) return self._remote - @staticmethod - def _remote_event(event: str, response: Any) -> None: + def _remote_event(self, event: str, response: Any) -> None: """Received event from remote websocket.""" if event == MS_ERROR_EVENT: # { 'event': 'ms.error', @@ -523,10 +553,17 @@ class SamsungTVWSBridge(SamsungTVBridge): message := data.get("message") ) == "unrecognized method value : ms.remote.control": LOGGER.error( - "Your TV seems to be unsupported by " - "SamsungTVWSBridge and may need a PIN: '%s'", + "Your TV seems to be unsupported by SamsungTVWSBridge" + " and needs a PIN: '%s'. Reloading", message, ) + self._notify_update_config_entry( + { + CONF_METHOD: METHOD_ENCRYPTED_WEBSOCKET, + CONF_PORT: ENCRYPTED_WEBSOCKET_PORT, + } + ) + self._notify_reload_callback() async def async_power_off(self) -> None: """Send power off command to remote.""" @@ -548,3 +585,162 @@ class SamsungTVWSBridge(SamsungTVBridge): LOGGER.debug( "Error closing connection to %s: %s", self.host, err.__repr__() ) + + +class SamsungTVEncryptedBridge(SamsungTVBridge): + """The Bridge for Encrypted WebSocket TVs (J/H models).""" + + def __init__( + self, + hass: HomeAssistant, + method: str, + host: str, + port: int | None = None, + entry_data: Mapping[str, Any] | None = None, + ) -> None: + """Initialize Bridge.""" + super().__init__(hass, method, host, port) + if entry_data: + self.token = entry_data.get(CONF_TOKEN) + self.session_id = entry_data.get(CONF_SESSION_ID) + self._rest_api_port: int | None = None + self._device_info: dict[str, Any] | None = None + self._remote: SamsungTVEncryptedWSAsyncRemote | None = None + self._remote_lock = asyncio.Lock() + + async def async_get_app_list(self) -> dict[str, str]: + """Get installed app list.""" + return {} + + async def async_is_on(self) -> bool: + """Tells if the TV is on.""" + LOGGER.debug("Checking if TV %s is on using websocket", self.host) + if remote := await self._async_get_remote(): + return remote.is_alive() + return False + + async def async_try_connect(self) -> str: + """Try to connect to the Websocket TV.""" + self.port = ENCRYPTED_WEBSOCKET_PORT + 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: TIMEOUT_REQUEST, + } + + try: + LOGGER.debug("Try config: %s", config) + async with SamsungTVEncryptedWSAsyncRemote( + host=self.host, + port=self.port, + web_session=async_get_clientsession(self.hass), + token=self.token or "", + session_id=self.session_id or "", + timeout=TIMEOUT_REQUEST, + ) as remote: + await remote.start_listening() + LOGGER.debug("Working config: %s", config) + return RESULT_SUCCESS + except WebSocketException as err: + LOGGER.debug("Working but unsupported config: %s, error: %s", config, err) + return RESULT_NOT_SUPPORTED + except (OSError, AsyncioTimeoutError, ConnectionFailure) as err: + LOGGER.debug("Failing config: %s, error: %s", config, err) + + return RESULT_CANNOT_CONNECT + + async def async_device_info(self) -> dict[str, Any] | None: + """Try to gather infos of this TV.""" + # Default to try all ports + rest_api_ports: Iterable[int] = WEBSOCKET_PORTS + if self._rest_api_port: + # We have already made a successful call to the REST api + rest_api_ports = (self._rest_api_port,) + + for rest_api_port in rest_api_ports: + assert self.port + rest_api = SamsungTVAsyncRest( + host=self.host, + session=async_get_clientsession(self.hass), + port=self.port, + timeout=TIMEOUT_WEBSOCKET, + ) + + with contextlib.suppress(HttpApiError, AsyncioTimeoutError): + device_info: dict[str, Any] = await rest_api.rest_device_info() + LOGGER.debug("Device info on %s is: %s", self.host, device_info) + self._device_info = device_info + self._rest_api_port = rest_api_port + return device_info + + return None + + async def async_send_keys(self, keys: list[str]) -> None: + """Send a list of keys using websocket protocol.""" + raise HomeAssistantError( + "Sending commands to encrypted TVs is not yet supported" + ) + + async def _async_get_remote(self) -> SamsungTVEncryptedWSAsyncRemote | None: + """Create or return a remote control instance.""" + if (remote := self._remote) and remote.is_alive(): + # If we have one then try to use it + return remote + + async with self._remote_lock: + # If we don't have one make sure we do it under the lock + # so we don't make two do due a race to get the remote + return await self._async_get_remote_under_lock() + + async def _async_get_remote_under_lock( + self, + ) -> SamsungTVEncryptedWSAsyncRemote | None: + """Create or return a remote control instance.""" + if self._remote is None or not self._remote.is_alive(): + # We need to create a new instance to reconnect. + LOGGER.debug("Create SamsungTVEncryptedBridge for %s", self.host) + assert self.port + self._remote = SamsungTVEncryptedWSAsyncRemote( + host=self.host, + port=self.port, + web_session=async_get_clientsession(self.hass), + token=self.token or "", + session_id=self.session_id or "", + timeout=TIMEOUT_WEBSOCKET, + ) + try: + # pylint:disable=[fixme] + # TODO: remove secondary timeout when library is bumped + # See https://github.com/xchwarze/samsung-tv-ws-api/pull/82 + await asyncio.wait_for( + self._remote.start_listening(), TIMEOUT_WEBSOCKET + ) + except (WebSocketException, AsyncioTimeoutError, OSError) as err: + LOGGER.debug( + "Failed to get remote for %s: %s", self.host, err.__repr__() + ) + self._remote = None + else: + LOGGER.debug("Created SamsungTVEncryptedBridge for %s", self.host) + return self._remote + + async def async_power_off(self) -> None: + """Send power off command to remote.""" + raise HomeAssistantError( + "Sending commands to encrypted TVs is not yet supported" + ) + + async def async_close_remote(self) -> None: + """Close remote object.""" + try: + if self._remote is not None: + # Close the current remote connection + await self._remote.close() + self._remote = None + except OSError as err: + LOGGER.debug( + "Error closing connection to %s: %s", self.host, err.__repr__() + ) diff --git a/homeassistant/components/samsungtv/config_flow.py b/homeassistant/components/samsungtv/config_flow.py index 812916a4654..29f46c1cd33 100644 --- a/homeassistant/components/samsungtv/config_flow.py +++ b/homeassistant/components/samsungtv/config_flow.py @@ -8,6 +8,7 @@ from typing import Any from urllib.parse import urlparse import getmac +from samsungtvws.encrypted.authenticator import SamsungTVEncryptedWSAsyncAuthenticator import voluptuous as vol from homeassistant import config_entries, data_entry_flow @@ -21,20 +22,25 @@ from homeassistant.const import ( CONF_TOKEN, ) from homeassistant.core import callback +from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.device_registry import format_mac from .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info from .const import ( CONF_MANUFACTURER, CONF_MODEL, + CONF_SESSION_ID, DEFAULT_MANUFACTURER, DOMAIN, + ENCRYPTED_WEBSOCKET_PORT, LEGACY_PORT, LOGGER, + METHOD_ENCRYPTED_WEBSOCKET, METHOD_LEGACY, METHOD_WEBSOCKET, RESULT_AUTH_MISSING, RESULT_CANNOT_CONNECT, + RESULT_INVALID_PIN, RESULT_NOT_SUPPORTED, RESULT_SUCCESS, RESULT_UNKNOWN_HOST, @@ -73,6 +79,9 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): self._id: int | None = None self._bridge: SamsungTVBridge | None = None self._device_info: dict[str, Any] | None = None + self._encrypted_authenticator: SamsungTVEncryptedWSAsyncAuthenticator | None = ( + None + ) def _get_entry_from_bridge(self) -> data_entry_flow.FlowResult: """Get device entry.""" @@ -174,6 +183,8 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): port = user_input.get(CONF_PORT) if port in WEBSOCKET_PORTS: user_input[CONF_METHOD] = METHOD_WEBSOCKET + elif port == ENCRYPTED_WEBSOCKET_PORT: + user_input[CONF_METHOD] = METHOD_ENCRYPTED_WEBSOCKET elif port == LEGACY_PORT: user_input[CONF_METHOD] = METHOD_LEGACY user_input[CONF_MANUFACTURER] = DEFAULT_MANUFACTURER @@ -365,10 +376,11 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """Confirm reauth.""" errors = {} assert self._reauth_entry + method = self._reauth_entry.data[CONF_METHOD] if user_input is not None: bridge = SamsungTVBridge.get_bridge( self.hass, - self._reauth_entry.data[CONF_METHOD], + method, self._reauth_entry.data[CONF_HOST], ) result = await bridge.async_try_connect() @@ -387,8 +399,50 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): errors = {"base": RESULT_AUTH_MISSING} self.context["title_placeholders"] = {"device": self._title} + step_id = "reauth_confirm" + if method == METHOD_ENCRYPTED_WEBSOCKET: + step_id = "reauth_confirm_encrypted" return self.async_show_form( - step_id="reauth_confirm", + step_id=step_id, errors=errors, description_placeholders={"device": self._title}, ) + + async def async_step_reauth_confirm_encrypted( + self, user_input: dict[str, Any] | None = None + ) -> data_entry_flow.FlowResult: + """Confirm reauth (encrypted method).""" + errors = {} + assert self._reauth_entry + if self._encrypted_authenticator is None: + self._encrypted_authenticator = SamsungTVEncryptedWSAsyncAuthenticator( + self._reauth_entry.data[CONF_HOST], + web_session=async_get_clientsession(self.hass), + ) + await self._encrypted_authenticator.start_pairing() + + if user_input is not None and (pin := user_input.get("pin")): + if token := await self._encrypted_authenticator.try_pin(pin): + session_id = ( + await self._encrypted_authenticator.get_session_id_and_close() + ) + new_data = { + **self._reauth_entry.data, + CONF_TOKEN: token, + CONF_SESSION_ID: session_id, + } + self.hass.config_entries.async_update_entry( + self._reauth_entry, data=new_data + ) + await self.hass.config_entries.async_reload(self._reauth_entry.entry_id) + return self.async_abort(reason="reauth_successful") + + errors = {"base": RESULT_INVALID_PIN} + + self.context["title_placeholders"] = {"device": self._title} + return self.async_show_form( + step_id="reauth_confirm_encrypted", + errors=errors, + description_placeholders={"device": self._title}, + data_schema=vol.Schema({vol.Required("pin"): str}), + ) diff --git a/homeassistant/components/samsungtv/const.py b/homeassistant/components/samsungtv/const.py index f2571372b1f..498b83a0539 100644 --- a/homeassistant/components/samsungtv/const.py +++ b/homeassistant/components/samsungtv/const.py @@ -16,18 +16,22 @@ CONF_DESCRIPTION = "description" CONF_MANUFACTURER = "manufacturer" CONF_MODEL = "model" CONF_ON_ACTION = "turn_on_action" +CONF_SESSION_ID = "session_id" RESULT_AUTH_MISSING = "auth_missing" +RESULT_INVALID_PIN = "invalid_pin" RESULT_SUCCESS = "success" RESULT_CANNOT_CONNECT = "cannot_connect" RESULT_NOT_SUPPORTED = "not_supported" RESULT_UNKNOWN_HOST = "unknown" METHOD_LEGACY = "legacy" +METHOD_ENCRYPTED_WEBSOCKET = "encrypted" METHOD_WEBSOCKET = "websocket" TIMEOUT_REQUEST = 31 TIMEOUT_WEBSOCKET = 5 LEGACY_PORT = 55000 +ENCRYPTED_WEBSOCKET_PORT = 8000 WEBSOCKET_PORTS = (8002, 8001) diff --git a/homeassistant/components/samsungtv/diagnostics.py b/homeassistant/components/samsungtv/diagnostics.py index ff792fff3e3..319e08827cf 100644 --- a/homeassistant/components/samsungtv/diagnostics.py +++ b/homeassistant/components/samsungtv/diagnostics.py @@ -9,9 +9,9 @@ from homeassistant.const import CONF_TOKEN from homeassistant.core import HomeAssistant from .bridge import SamsungTVBridge -from .const import DOMAIN +from .const import CONF_SESSION_ID, DOMAIN -TO_REDACT = {CONF_TOKEN} +TO_REDACT = {CONF_TOKEN, CONF_SESSION_ID} async def async_get_config_entry_diagnostics( diff --git a/homeassistant/components/samsungtv/media_player.py b/homeassistant/components/samsungtv/media_player.py index baea046c3dc..a293321d311 100644 --- a/homeassistant/components/samsungtv/media_player.py +++ b/homeassistant/components/samsungtv/media_player.py @@ -26,7 +26,14 @@ from homeassistant.components.media_player.const import ( SUPPORT_VOLUME_STEP, ) from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry -from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, STATE_OFF, STATE_ON +from homeassistant.const import ( + CONF_HOST, + CONF_MAC, + CONF_METHOD, + CONF_NAME, + STATE_OFF, + STATE_ON, +) from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_component import homeassistant.helpers.config_validation as cv @@ -44,6 +51,7 @@ from .const import ( DEFAULT_NAME, DOMAIN, LOGGER, + METHOD_ENCRYPTED_WEBSOCKET, ) SOURCES = {"TV": "KEY_TV", "HDMI": "KEY_HDMI"} @@ -112,11 +120,14 @@ class SamsungTVDevice(MediaPlayerEntity): self._attr_source_list = list(SOURCES) self._app_list: dict[str, str] | None = None - if self._on_script or self._mac: - self._attr_supported_features = SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON - else: + if config_entry.data.get(CONF_METHOD) != METHOD_ENCRYPTED_WEBSOCKET: + # Encrypted websockets currently only support ON/OFF status self._attr_supported_features = SUPPORT_SAMSUNGTV + if self._on_script or self._mac: + # Add turn-on if on_script or mac is available + self._attr_supported_features |= SUPPORT_TURN_ON + self._attr_device_info = DeviceInfo( name=self.name, manufacturer=config_entry.data.get(CONF_MANUFACTURER), diff --git a/homeassistant/components/samsungtv/strings.json b/homeassistant/components/samsungtv/strings.json index f413a7f1219..f64620638bf 100644 --- a/homeassistant/components/samsungtv/strings.json +++ b/homeassistant/components/samsungtv/strings.json @@ -13,11 +13,15 @@ "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." + "description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds or input PIN." + }, + "reauth_confirm_encrypted": { + "description": "Please enter the PIN displayed on {device}." } }, "error": { - "auth_missing": "[%key:component::samsungtv::config::abort::auth_missing%]" + "auth_missing": "[%key:component::samsungtv::config::abort::auth_missing%]", + "invalid_pin": "PIN is invalid, please try again." }, "abort": { "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]", diff --git a/homeassistant/components/samsungtv/translations/en.json b/homeassistant/components/samsungtv/translations/en.json index 4648f930e9b..2e3dd88bec7 100644 --- a/homeassistant/components/samsungtv/translations/en.json +++ b/homeassistant/components/samsungtv/translations/en.json @@ -12,7 +12,8 @@ "unknown": "Unexpected error" }, "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." + "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.", + "invalid_pin": "PIN is invalid, please try again." }, "flow_title": "{device}", "step": { @@ -21,7 +22,10 @@ "title": "Samsung TV" }, "reauth_confirm": { - "description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds." + "description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds or input PIN." + }, + "reauth_confirm_encrypted": { + "description": "Please enter the PIN displayed on {device}." }, "user": { "data": { diff --git a/tests/components/samsungtv/__init__.py b/tests/components/samsungtv/__init__.py index 4ad1622c6ca..7cace81f7a6 100644 --- a/tests/components/samsungtv/__init__.py +++ b/tests/components/samsungtv/__init__.py @@ -1 +1,23 @@ """Tests for the samsungtv component.""" + + +from homeassistant.components.samsungtv.const import DOMAIN +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.typing import ConfigType +from homeassistant.setup import async_setup_component + +from tests.common import MockConfigEntry + + +async def setup_samsungtv_entry(hass: HomeAssistant, data: ConfigType) -> ConfigEntry: + """Set up mock Samsung TV from config entry data.""" + entry = MockConfigEntry( + domain=DOMAIN, data=data, entry_id="123456", unique_id="any" + ) + entry.add_to_hass(hass) + + await async_setup_component(hass, DOMAIN, {}) + await hass.async_block_till_done() + + return entry diff --git a/tests/components/samsungtv/conftest.py b/tests/components/samsungtv/conftest.py index c7733a3652a..3d9ea74e346 100644 --- a/tests/components/samsungtv/conftest.py +++ b/tests/components/samsungtv/conftest.py @@ -9,6 +9,7 @@ from unittest.mock import AsyncMock, Mock, patch import pytest from samsungctl import Remote from samsungtvws.async_remote import SamsungTVWSAsyncRemote +from samsungtvws.encrypted.remote import SamsungTVEncryptedWSAsyncRemote import homeassistant.util.dt as dt_util @@ -77,6 +78,32 @@ def remotews_fixture() -> Mock: yield remotews +@pytest.fixture(name="remoteencws") +def remoteencws_fixture() -> Mock: + """Patch the samsungtvws SamsungTVEncryptedWSAsyncRemote.""" + remoteencws = Mock(SamsungTVEncryptedWSAsyncRemote) + remoteencws.__aenter__ = AsyncMock(return_value=remoteencws) + remoteencws.__aexit__ = AsyncMock() + + def _start_listening( + ws_event_callback: Callable[[str, Any], Awaitable[None] | None] | None = None + ): + remoteencws.ws_event_callback = ws_event_callback + + def _mock_ws_event_callback(event: str, response: Any): + if remoteencws.ws_event_callback: + remoteencws.ws_event_callback(event, response) + + remoteencws.start_listening.side_effect = _start_listening + remoteencws.raise_mock_ws_event_callback = Mock(side_effect=_mock_ws_event_callback) + + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote", + ) as remotews_class: + remotews_class.return_value = remoteencws + yield remoteencws + + @pytest.fixture(name="delay") def delay_fixture() -> Mock: """Patch the delay script function.""" diff --git a/tests/components/samsungtv/const.py b/tests/components/samsungtv/const.py index 7f9f175c171..11e92481f21 100644 --- a/tests/components/samsungtv/const.py +++ b/tests/components/samsungtv/const.py @@ -1,4 +1,29 @@ """Constants for the samsungtv tests.""" +from homeassistant.components.samsungtv.const import CONF_SESSION_ID +from homeassistant.const import ( + CONF_HOST, + CONF_IP_ADDRESS, + CONF_MAC, + CONF_METHOD, + CONF_NAME, + CONF_PORT, + CONF_TOKEN, +) + +MOCK_CONFIG_ENCRYPTED_WS = { + CONF_HOST: "fake_host", + CONF_NAME: "fake", + CONF_PORT: 8000, +} +MOCK_ENTRYDATA_ENCRYPTED_WS = { + **MOCK_CONFIG_ENCRYPTED_WS, + CONF_IP_ADDRESS: "test", + CONF_METHOD: "encrypted", + CONF_MAC: "aa:bb:cc:dd:ee:ff", + CONF_TOKEN: "037739871315caef138547b03e348b72", + CONF_SESSION_ID: "2", +} + SAMPLE_APP_LIST = [ { "appId": "111299001912", @@ -73,3 +98,32 @@ SAMPLE_DEVICE_INFO_FRAME = { "uri": "https://1.2.3.4:8002/api/v2/", "version": "2.0.25", } + +SAMPLE_DEVICE_INFO_UE48JU6400 = { + "id": "uuid:223da676-497a-4e06-9507-5e27ec4f0fb3", + "name": "[TV] TV-UE48JU6470", + "version": "2.0.25", + "device": { + "type": "Samsung SmartTV", + "duid": "uuid:223da676-497a-4e06-9507-5e27ec4f0fb3", + "model": "15_HAWKM_UHD_2D", + "modelName": "UE48JU6400", + "description": "Samsung DTV RCR", + "networkType": "wired", + "ssid": "", + "ip": "1.2.3.4", + "firmwareVersion": "Unknown", + "name": "[TV] TV-UE48JU6470", + "id": "uuid:223da676-497a-4e06-9507-5e27ec4f0fb3", + "udn": "uuid:223da676-497a-4e06-9507-5e27ec4f0fb3", + "resolution": "1920x1080", + "countryCode": "AT", + "msfVersion": "2.0.25", + "smartHubAgreement": "true", + "wifiMac": "aa:bb:ww:ii:ff:ii", + "developerMode": "0", + "developerIP": "", + }, + "type": "Samsung SmartTV", + "uri": "https://1.2.3.4:8002/api/v2/", +} diff --git a/tests/components/samsungtv/test_config_flow.py b/tests/components/samsungtv/test_config_flow.py index 8f1a22395fb..72f23b01350 100644 --- a/tests/components/samsungtv/test_config_flow.py +++ b/tests/components/samsungtv/test_config_flow.py @@ -18,9 +18,11 @@ from homeassistant.components import dhcp, ssdp, zeroconf from homeassistant.components.samsungtv.const import ( CONF_MANUFACTURER, CONF_MODEL, + CONF_SESSION_ID, DEFAULT_MANUFACTURER, DOMAIN, LEGACY_PORT, + METHOD_ENCRYPTED_WEBSOCKET, METHOD_LEGACY, METHOD_WEBSOCKET, RESULT_AUTH_MISSING, @@ -49,7 +51,13 @@ from homeassistant.const import ( from homeassistant.core import HomeAssistant from homeassistant.setup import async_setup_component -from .const import SAMPLE_APP_LIST, SAMPLE_DEVICE_INFO_FRAME +from . import setup_samsungtv_entry +from .const import ( + MOCK_CONFIG_ENCRYPTED_WS, + MOCK_ENTRYDATA_ENCRYPTED_WS, + SAMPLE_APP_LIST, + SAMPLE_DEVICE_INFO_FRAME, +) from tests.common import MockConfigEntry @@ -484,6 +492,9 @@ async def test_ssdp_websocket_not_supported( with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"), + ), patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening", + side_effect=WebSocketProtocolError("Boom"), ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote", ) as remotews, patch.object( @@ -645,12 +656,16 @@ async def test_import_legacy(hass: HomeAssistant) -> None: async def test_import_legacy_without_name(hass: HomeAssistant, rest_api: Mock) -> None: """Test importing from yaml without a name.""" rest_api.rest_device_info.return_value = None - result = await hass.config_entries.flow.async_init( - DOMAIN, - context={"source": config_entries.SOURCE_IMPORT}, - data=MOCK_IMPORT_DATA_WITHOUT_NAME, - ) - await hass.async_block_till_done() + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening", + side_effect=WebSocketProtocolError("Boom"), + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_IMPORT}, + data=MOCK_IMPORT_DATA_WITHOUT_NAME, + ) + await hass.async_block_till_done() assert result["type"] == "create_entry" assert result["title"] == "fake_host" assert result["data"][CONF_HOST] == "fake_host" @@ -682,6 +697,26 @@ async def test_import_websocket(hass: HomeAssistant): assert result["result"].unique_id is None +@pytest.mark.usefixtures("remoteencws") +async def test_import_websocket_encrypted(hass: HomeAssistant): + """Test importing from yaml with hostname.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_IMPORT}, + data=MOCK_CONFIG_ENCRYPTED_WS, + ) + await hass.async_block_till_done() + + assert result["type"] == "create_entry" + assert result["title"] == "fake" + assert result["data"][CONF_METHOD] == METHOD_ENCRYPTED_WEBSOCKET + assert result["data"][CONF_PORT] == 8000 + 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 + + @pytest.mark.usefixtures("remotews") async def test_import_websocket_without_port(hass: HomeAssistant): """Test importing from yaml with hostname by no port.""" @@ -826,7 +861,7 @@ async def test_zeroconf_ignores_soundbar(hass: HomeAssistant, rest_api: Mock) -> assert result["reason"] == "not_supported" -@pytest.mark.usefixtures("remote", "remotews") +@pytest.mark.usefixtures("remote", "remotews", "remoteencws") async def test_zeroconf_no_device_info(hass: HomeAssistant, rest_api: Mock) -> None: """Test starting a flow from zeroconf where device_info returns None.""" rest_api.rest_device_info.return_value = None @@ -1234,6 +1269,9 @@ async def test_update_legacy_missing_mac_from_dhcp_no_unique_id( with patch( "homeassistant.components.samsungtv.bridge.Remote.__enter__", return_value=True, + ), patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening", + side_effect=WebSocketProtocolError("Boom"), ), patch( "homeassistant.components.samsungtv.async_setup", return_value=True, @@ -1364,6 +1402,78 @@ async def test_form_reauth_websocket_not_supported(hass: HomeAssistant) -> None: assert result2["reason"] == "not_supported" +@pytest.mark.usefixtures("remoteencws") +async def test_form_reauth_encrypted(hass: HomeAssistant) -> None: + """Test reauth flow for encrypted TVs.""" + encrypted_entry_data = {**MOCK_ENTRYDATA_ENCRYPTED_WS} + del encrypted_entry_data[CONF_TOKEN] + del encrypted_entry_data[CONF_SESSION_ID] + + entry = await setup_samsungtv_entry(hass, encrypted_entry_data) + assert entry.state == config_entries.ConfigEntryState.SETUP_ERROR + flows_in_progress = [ + flow + for flow in hass.config_entries.flow.async_progress() + if flow["context"]["source"] == "reauth" + ] + assert len(flows_in_progress) == 1 + result = flows_in_progress[0] + + with patch( + "homeassistant.components.samsungtv.config_flow.SamsungTVEncryptedWSAsyncAuthenticator", + autospec=True, + ) as authenticator_mock: + authenticator_mock.return_value.try_pin.side_effect = [ + None, + "037739871315caef138547b03e348b72", + ] + authenticator_mock.return_value.get_session_id_and_close.return_value = "1" + + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + assert result["type"] == "form" + assert result["step_id"] == "reauth_confirm_encrypted" + assert result["errors"] == {} + + # First time on reauth_confirm_encrypted + # creates the authenticator, start pairing and requests PIN + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=None + ) + assert result["type"] == "form" + assert result["step_id"] == "reauth_confirm_encrypted" + + # Invalid PIN + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={"pin": "invalid"} + ) + assert result["type"] == "form" + assert result["step_id"] == "reauth_confirm_encrypted" + + # Valid PIN + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={"pin": "1234"} + ) + await hass.async_block_till_done() + assert result["type"] == "abort" + assert result["reason"] == "reauth_successful" + assert entry.state == config_entries.ConfigEntryState.LOADED + + authenticator_mock.assert_called_once() + assert authenticator_mock.call_args[0] == ("fake_host",) + + authenticator_mock.return_value.start_pairing.assert_called_once() + assert authenticator_mock.return_value.try_pin.call_count == 2 + assert authenticator_mock.return_value.try_pin.call_args_list == [ + call("invalid"), + call("1234"), + ] + authenticator_mock.return_value.get_session_id_and_close.assert_called_once() + + assert entry.data[CONF_TOKEN] == "037739871315caef138547b03e348b72" + assert entry.data[CONF_SESSION_ID] == "1" + + @pytest.mark.usefixtures("remotews") async def test_update_incorrect_udn_matching_upnp_udn_unique_id_added_from_ssdp( hass: HomeAssistant, diff --git a/tests/components/samsungtv/test_diagnostics.py b/tests/components/samsungtv/test_diagnostics.py index ba9e050d9ce..88fb98a66db 100644 --- a/tests/components/samsungtv/test_diagnostics.py +++ b/tests/components/samsungtv/test_diagnostics.py @@ -1,39 +1,30 @@ """Test samsungtv diagnostics.""" +from unittest.mock import Mock + from aiohttp import ClientSession import pytest +from samsungtvws.exceptions import HttpApiError from homeassistant.components.diagnostics import REDACTED -from homeassistant.components.samsungtv import DOMAIN -from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant -from .const import SAMPLE_DEVICE_INFO_WIFI +from . import setup_samsungtv_entry +from .const import ( + MOCK_ENTRYDATA_ENCRYPTED_WS, + SAMPLE_DEVICE_INFO_UE48JU6400, + SAMPLE_DEVICE_INFO_WIFI, +) from .test_media_player import MOCK_ENTRY_WS_WITH_MAC -from tests.common import MockConfigEntry from tests.components.diagnostics import get_diagnostics_for_config_entry -@pytest.fixture(name="config_entry") -def get_config_entry(hass: HomeAssistant) -> ConfigEntry: - """Create and register mock config entry.""" - config_entry = MockConfigEntry( - domain=DOMAIN, - data=MOCK_ENTRY_WS_WITH_MAC, - entry_id="123456", - unique_id="any", - ) - config_entry.add_to_hass(hass) - return config_entry - - @pytest.mark.usefixtures("remotews") async def test_entry_diagnostics( - hass: HomeAssistant, config_entry: ConfigEntry, hass_client: ClientSession + hass: HomeAssistant, hass_client: ClientSession ) -> None: """Test config entry diagnostics.""" - await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() + config_entry = await setup_samsungtv_entry(hass, MOCK_ENTRY_WS_WITH_MAC) assert await get_diagnostics_for_config_entry(hass, hass_client, config_entry) == { "entry": { @@ -60,3 +51,74 @@ async def test_entry_diagnostics( }, "device_info": SAMPLE_DEVICE_INFO_WIFI, } + + +@pytest.mark.usefixtures("remoteencws") +async def test_entry_diagnostics_encrypted( + hass: HomeAssistant, rest_api: Mock, hass_client: ClientSession +) -> None: + """Test config entry diagnostics.""" + rest_api.rest_device_info.return_value = SAMPLE_DEVICE_INFO_UE48JU6400 + config_entry = await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + + assert await get_diagnostics_for_config_entry(hass, hass_client, config_entry) == { + "entry": { + "data": { + "host": "fake_host", + "ip_address": "test", + "mac": "aa:bb:cc:dd:ee:ff", + "method": "encrypted", + "model": "UE48JU6400", + "name": "fake", + "port": 8000, + "token": REDACTED, + "session_id": REDACTED, + }, + "disabled_by": None, + "domain": "samsungtv", + "entry_id": "123456", + "options": {}, + "pref_disable_new_entities": False, + "pref_disable_polling": False, + "source": "user", + "title": "Mock Title", + "unique_id": "any", + "version": 2, + }, + "device_info": SAMPLE_DEVICE_INFO_UE48JU6400, + } + + +@pytest.mark.usefixtures("remoteencws") +async def test_entry_diagnostics_encrypte_offline( + hass: HomeAssistant, rest_api: Mock, hass_client: ClientSession +) -> None: + """Test config entry diagnostics.""" + rest_api.rest_device_info.side_effect = HttpApiError + config_entry = await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + + assert await get_diagnostics_for_config_entry(hass, hass_client, config_entry) == { + "entry": { + "data": { + "host": "fake_host", + "ip_address": "test", + "mac": "aa:bb:cc:dd:ee:ff", + "method": "encrypted", + "name": "fake", + "port": 8000, + "token": REDACTED, + "session_id": REDACTED, + }, + "disabled_by": None, + "domain": "samsungtv", + "entry_id": "123456", + "options": {}, + "pref_disable_new_entities": False, + "pref_disable_polling": False, + "source": "user", + "title": "Mock Title", + "unique_id": "any", + "version": 2, + }, + "device_info": None, + } diff --git a/tests/components/samsungtv/test_init.py b/tests/components/samsungtv/test_init.py index 14462b1aaf3..6fec07fab84 100644 --- a/tests/components/samsungtv/test_init.py +++ b/tests/components/samsungtv/test_init.py @@ -80,6 +80,9 @@ async def test_setup_from_yaml_without_port_device_offline(hass: HomeAssistant) """Test import from yaml when the device is offline.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError + ), patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening", + side_effect=OSError, ), patch( "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote.open", side_effect=OSError, diff --git a/tests/components/samsungtv/test_media_player.py b/tests/components/samsungtv/test_media_player.py index 74849fb9fee..dac0c43bb1b 100644 --- a/tests/components/samsungtv/test_media_player.py +++ b/tests/components/samsungtv/test_media_player.py @@ -8,6 +8,7 @@ import pytest from samsungctl import exceptions from samsungtvws.async_remote import SamsungTVWSAsyncRemote from samsungtvws.command import SamsungTVSleepCommand +from samsungtvws.encrypted.remote import SamsungTVEncryptedWSAsyncRemote from samsungtvws.exceptions import ConnectionFailure, HttpApiError from samsungtvws.remote import ChannelEmitCommand, SendRemoteKey from websockets.exceptions import ConnectionClosedError, WebSocketException @@ -29,6 +30,9 @@ from homeassistant.components.media_player.const import ( from homeassistant.components.samsungtv.const import ( CONF_ON_ACTION, DOMAIN as SAMSUNGTV_DOMAIN, + ENCRYPTED_WEBSOCKET_PORT, + METHOD_ENCRYPTED_WEBSOCKET, + METHOD_WEBSOCKET, TIMEOUT_WEBSOCKET, ) from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV @@ -65,7 +69,12 @@ from homeassistant.helpers.typing import ConfigType from homeassistant.setup import async_setup_component import homeassistant.util.dt as dt_util -from .const import SAMPLE_APP_LIST, SAMPLE_DEVICE_INFO_FRAME +from . import setup_samsungtv_entry +from .const import ( + MOCK_ENTRYDATA_ENCRYPTED_WS, + SAMPLE_APP_LIST, + SAMPLE_DEVICE_INFO_FRAME, +) from tests.common import MockConfigEntry, async_fire_time_changed @@ -119,6 +128,7 @@ MOCK_ENTRY_WS_WITH_MAC = { CONF_TOKEN: "123456789", } + ENTITY_ID_NOTURNON = f"{DOMAIN}.fake_noturnon" MOCK_CONFIG_NOTURNON = { SAMSUNGTV_DOMAIN: [ @@ -221,6 +231,30 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non remote_class.assert_called_once_with(**MOCK_CALLS_WS) +async def test_setup_encrypted_websocket( + hass: HomeAssistant, mock_now: datetime +) -> None: + """Test setup of platform from config entry.""" + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote" + ) as remote_class: + remote = Mock(SamsungTVEncryptedWSAsyncRemote) + remote.__aenter__ = AsyncMock(return_value=remote) + remote.__aexit__ = AsyncMock() + remote_class.return_value = remote + + await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + + next_update = mock_now + timedelta(minutes=5) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + + state = hass.states.get(ENTITY_ID) + assert state + remote_class.assert_called_once() + + @pytest.mark.usefixtures("remote") async def test_update_on(hass: HomeAssistant, mock_now: datetime) -> None: """Testing update tv on.""" @@ -343,6 +377,30 @@ async def test_update_off_ws_with_power_state( remotews.start_listening.assert_not_called() +async def test_update_off_encryptedws( + hass: HomeAssistant, remoteencws: Mock, rest_api: Mock, mock_now: datetime +) -> None: + """Testing update tv off.""" + await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + + rest_api.rest_device_info.assert_called_once() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + remoteencws.start_listening = Mock(side_effect=WebSocketException("Boom")) + remoteencws.is_alive.return_value = False + + next_update = mock_now + timedelta(minutes=5) + with patch("homeassistant.util.dt.utcnow", return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_OFF + rest_api.rest_device_info.assert_called_once() + + @pytest.mark.usefixtures("remote") async def test_update_access_denied(hass: HomeAssistant, mock_now: datetime) -> None: """Testing update tv access denied exception.""" @@ -543,6 +601,21 @@ async def test_send_key_websocketexception(hass: HomeAssistant, remotews: Mock) assert state.state == STATE_ON +async def test_send_key_websocketexception_encrypted( + hass: HomeAssistant, remoteencws: Mock +) -> None: + """Testing unhandled response exception.""" + await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + remoteencws.send_commands = Mock(side_effect=WebSocketException("Boom")) + with pytest.raises(HomeAssistantError) as exc_info: + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert exc_info.match("media_player.fake does not support this service.") + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + async def test_send_key_os_error_ws(hass: HomeAssistant, remotews: Mock) -> None: """Testing unhandled response exception.""" await setup_samsungtv(hass, MOCK_CONFIGWS) @@ -554,6 +627,21 @@ async def test_send_key_os_error_ws(hass: HomeAssistant, remotews: Mock) -> None assert state.state == STATE_ON +async def test_send_key_os_error_ws_encrypted( + hass: HomeAssistant, remoteencws: Mock +) -> None: + """Testing unhandled response exception.""" + await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + remoteencws.send_commands = Mock(side_effect=OSError("Boom")) + with pytest.raises(HomeAssistantError) as exc_info: + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert exc_info.match("media_player.fake does not support this service.") + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_ON + + async def test_send_key_os_error(hass: HomeAssistant, remote: Mock) -> None: """Testing broken pipe Exception.""" await setup_samsungtv(hass, MOCK_CONFIG) @@ -714,6 +802,28 @@ async def test_turn_off_websocket_frame( assert commands[2].params["DataOfCmd"] == "KEY_POWER" +async def test_turn_off_encrypted_websocket( + hass: HomeAssistant, remoteencws: Mock, caplog: pytest.LogCaptureFixture +) -> None: + """Test for turn_off.""" + await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + + remoteencws.send_commands.reset_mock() + + with pytest.raises(HomeAssistantError) as exc_info: + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert exc_info.match("media_player.fake does not support this service.") + + # commands not sent : power off in progress + with pytest.raises(HomeAssistantError) as exc_info: + assert await hass.services.async_call( + DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert exc_info.match("media_player.fake does not support this service.") + + async def test_turn_off_legacy(hass: HomeAssistant, remote: Mock) -> None: """Test for turn_off.""" await setup_samsungtv(hass, MOCK_CONFIG_NOTURNON) @@ -751,6 +861,20 @@ async def test_turn_off_ws_os_error( assert "Error closing connection" in caplog.text +async def test_turn_off_encryptedws_os_error( + hass: HomeAssistant, remoteencws: Mock, caplog: pytest.LogCaptureFixture +) -> None: + """Test for turn_off with OSError.""" + caplog.set_level(logging.DEBUG) + await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS) + remoteencws.close = Mock(side_effect=OSError("BOOM")) + with pytest.raises(HomeAssistantError) as exc_info: + assert await hass.services.async_call( + DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True + ) + assert exc_info.match("media_player.fake does not support this service.") + + async def test_volume_up(hass: HomeAssistant, remote: Mock) -> None: """Test for volume_up.""" await setup_samsungtv(hass, MOCK_CONFIG) @@ -1074,11 +1198,10 @@ async def test_websocket_unsupported_remote_control( hass: HomeAssistant, remotews: Mock, caplog: pytest.LogCaptureFixture ) -> None: """Test for turn_off.""" - with patch( - "homeassistant.components.samsungtv.bridge.Remote", - side_effect=[OSError("Boom"), DEFAULT_MOCK], - ): - await setup_samsungtv(hass, MOCK_CONFIGWS) + entry = await setup_samsungtv_entry(hass, MOCK_ENTRY_WS) + + assert entry.data[CONF_METHOD] == METHOD_WEBSOCKET + assert entry.data[CONF_PORT] == 8001 remotews.send_command.reset_mock() @@ -1102,6 +1225,18 @@ async def test_websocket_unsupported_remote_control( # error logged assert ( - "Your TV seems to be unsupported by SamsungTVWSBridge and may need a PIN: " + "Your TV seems to be unsupported by SamsungTVWSBridge and needs a PIN: " "'unrecognized method value : ms.remote.control'" in caplog.text ) + + # ensure reauth triggered, and method/port updated + await hass.async_block_till_done() + assert [ + flow + for flow in hass.config_entries.flow.async_progress() + if flow["context"]["source"] == "reauth" + ] + assert entry.data[CONF_METHOD] == METHOD_ENCRYPTED_WEBSOCKET + assert entry.data[CONF_PORT] == ENCRYPTED_WEBSOCKET_PORT + state = hass.states.get(ENTITY_ID) + assert state.state == STATE_UNAVAILABLE