diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py index 470d7e80584671..4363e7d8c446f0 100644 --- a/homeassistant/components/samsungtv/__init__.py +++ b/homeassistant/components/samsungtv/__init__.py @@ -173,7 +173,7 @@ async def _async_create_bridge_with_updated_data( else: # When we imported from yaml we didn't setup the method # because we didn't know it - port, method, info = await async_get_device_info(hass, None, host) + _result, port, method, info = await async_get_device_info(hass, host) load_info_attempted = True if not port or not method: raise ConfigEntryNotReady( diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py index b89c76f028e102..4e53b0fd0a5459 100644 --- a/homeassistant/components/samsungtv/bridge.py +++ b/homeassistant/components/samsungtv/bridge.py @@ -23,7 +23,12 @@ MS_ERROR_EVENT, parse_installed_app, ) -from samsungtvws.exceptions import ConnectionFailure, HttpApiError +from samsungtvws.exceptions import ( + ConnectionFailure, + HttpApiError, + ResponseError, + UnauthorizedError, +) from samsungtvws.remote import ChannelEmitCommand, SendRemoteKey from websockets.exceptions import ConnectionClosedError, WebSocketException @@ -54,6 +59,7 @@ RESULT_CANNOT_CONNECT, RESULT_NOT_SUPPORTED, RESULT_SUCCESS, + SUCCESSFUL_RESULTS, TIMEOUT_REQUEST, TIMEOUT_WEBSOCKET, VALUE_CONF_ID, @@ -66,6 +72,8 @@ ENCRYPTED_MODEL_USES_POWER_OFF = {"H6400"} ENCRYPTED_MODEL_USES_POWER = {"JU6400", "JU641D"} +REST_EXCEPTIONS = (HttpApiError, AsyncioTimeoutError, ResponseError) + def mac_from_device_info(info: dict[str, Any]) -> str | None: """Extract the mac address from the device info.""" @@ -76,36 +84,39 @@ def mac_from_device_info(info: dict[str, Any]) -> str | None: async def async_get_device_info( hass: HomeAssistant, - bridge: SamsungTVBridge | None, host: str, -) -> tuple[int | None, str | None, dict[str, Any] | None]: +) -> tuple[str, 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 + # Try the websocket ssl and non-ssl 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() + LOGGER.debug( + "Fetching rest info via %s was successful: %s, checking for encrypted", + port, + info, + ) + encrypted_bridge = SamsungTVEncryptedBridge( + hass, METHOD_ENCRYPTED_WEBSOCKET, host, ENCRYPTED_WEBSOCKET_PORT + ) + result = await encrypted_bridge.async_try_connect() + if result != RESULT_CANNOT_CONNECT: + return ( + result, + ENCRYPTED_WEBSOCKET_PORT, + METHOD_ENCRYPTED_WEBSOCKET, + info, + ) + return RESULT_SUCCESS, port, METHOD_WEBSOCKET, 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() + if result in SUCCESSFUL_RESULTS: + return result, LEGACY_PORT, METHOD_LEGACY, await bridge.async_device_info() # Failed to get info - return None, None, None + return result, None, None, None class SamsungTVBridge(ABC): @@ -433,8 +444,11 @@ async def async_try_connect(self) -> str: "Working but unsupported config: %s, error: %s", config, err ) result = RESULT_NOT_SUPPORTED - except (OSError, AsyncioTimeoutError, ConnectionFailure) as err: - LOGGER.debug("Failing config: %s, error: %s", config, err) + except UnauthorizedError as err: + LOGGER.debug("Failing config: %s, %s error: %s", config, type(err), err) + return RESULT_AUTH_MISSING + except (ConnectionFailure, OSError, AsyncioTimeoutError) as err: + LOGGER.debug("Failing config: %s, %s error: %s", config, type(err), err) # pylint: disable=useless-else-on-loop else: if result: @@ -453,7 +467,7 @@ async def async_device_info(self) -> dict[str, Any] | None: timeout=TIMEOUT_WEBSOCKET, ) - with contextlib.suppress(HttpApiError, AsyncioTimeoutError): + with contextlib.suppress(*REST_EXCEPTIONS): 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 @@ -654,8 +668,7 @@ async def async_try_connect(self) -> str: 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, + CONF_TIMEOUT: TIMEOUT_WEBSOCKET, } try: @@ -669,13 +682,14 @@ async def async_try_connect(self) -> str: 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) + else: + LOGGER.debug("Working config: %s", config) + return RESULT_SUCCESS return RESULT_CANNOT_CONNECT @@ -696,7 +710,7 @@ async def async_device_info(self) -> dict[str, Any] | None: timeout=TIMEOUT_WEBSOCKET, ) - with contextlib.suppress(HttpApiError, AsyncioTimeoutError): + with contextlib.suppress(*REST_EXCEPTIONS): 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 diff --git a/homeassistant/components/samsungtv/config_flow.py b/homeassistant/components/samsungtv/config_flow.py index 29f46c1cd3300c..0b87e38b00a459 100644 --- a/homeassistant/components/samsungtv/config_flow.py +++ b/homeassistant/components/samsungtv/config_flow.py @@ -30,6 +30,7 @@ CONF_MANUFACTURER, CONF_MODEL, CONF_SESSION_ID, + CONF_SSDP_RENDERING_CONTROL_LOCATION, DEFAULT_MANUFACTURER, DOMAIN, ENCRYPTED_WEBSOCKET_PORT, @@ -44,20 +45,34 @@ RESULT_NOT_SUPPORTED, RESULT_SUCCESS, RESULT_UNKNOWN_HOST, + SUCCESSFUL_RESULTS, + UPNP_SVC_RENDERINGCONTROL, WEBSOCKET_PORTS, ) DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME): str}) -SUPPORTED_METHODS = [METHOD_LEGACY, METHOD_WEBSOCKET] def _strip_uuid(udn: str) -> str: return udn[5:] if udn.startswith("uuid:") else udn -def _entry_is_complete(entry: config_entries.ConfigEntry) -> bool: - """Return True if the config entry information is complete.""" - return bool(entry.unique_id and entry.data.get(CONF_MAC)) +def _entry_is_complete( + entry: config_entries.ConfigEntry, ssdp_rendering_control_location: str | None +) -> bool: + """Return True if the config entry information is complete. + + If we do not have an ssdp location we consider it complete + as some TVs will not support SSDP/UPNP + """ + return bool( + entry.unique_id + and entry.data.get(CONF_MAC) + and ( + not ssdp_rendering_control_location + or entry.data.get(CONF_SSDP_RENDERING_CONTROL_LOCATION) + ) + ) class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): @@ -72,22 +87,22 @@ def __init__(self) -> None: self._mac: str | None = None self._udn: str | None = None self._upnp_udn: str | None = None + self._ssdp_rendering_control_location: str | None = None self._manufacturer: str | None = None self._model: str | None = None + self._connect_result: str | None = None + self._method: str | None = None self._name: str | None = None self._title: str = "" 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.""" - assert self._bridge + self._authenticator: SamsungTVEncryptedWSAsyncAuthenticator | None = None - data = { + def _base_config_entry(self) -> dict[str, Any]: + """Generate the base config entry without the method.""" + assert self._bridge is not None + return { CONF_HOST: self._host, CONF_MAC: self._mac, CONF_MANUFACTURER: self._manufacturer or DEFAULT_MANUFACTURER, @@ -95,7 +110,13 @@ def _get_entry_from_bridge(self) -> data_entry_flow.FlowResult: CONF_MODEL: self._model, CONF_NAME: self._name, CONF_PORT: self._bridge.port, + CONF_SSDP_RENDERING_CONTROL_LOCATION: self._ssdp_rendering_control_location, } + + def _get_entry_from_bridge(self) -> data_entry_flow.FlowResult: + """Get device entry.""" + assert self._bridge + data = self._base_config_entry() if self._bridge.token: data[CONF_TOKEN] = self._bridge.token return self.async_create_entry( @@ -115,48 +136,66 @@ async def _async_set_unique_id_from_udn( ) -> None: """Set the unique id from the udn.""" assert self._host is not None - await self.async_set_unique_id(self._udn, raise_on_progress=raise_on_progress) + # Set the unique id without raising on progress in case + # there are two SSDP flows with for each ST + await self.async_set_unique_id(self._udn, raise_on_progress=False) if ( entry := self._async_update_existing_matching_entry() - ) and _entry_is_complete(entry): + ) and _entry_is_complete(entry, self._ssdp_rendering_control_location): raise data_entry_flow.AbortFlow("already_configured") + # Now that we have updated the config entry, we can raise + # if another one is progressing + if raise_on_progress: + await self.async_set_unique_id(self._udn, raise_on_progress=True) def _async_update_and_abort_for_matching_unique_id(self) -> None: """Abort and update host and mac if we have it.""" updates = {CONF_HOST: self._host} if self._mac: updates[CONF_MAC] = self._mac + if self._ssdp_rendering_control_location: + updates[ + CONF_SSDP_RENDERING_CONTROL_LOCATION + ] = self._ssdp_rendering_control_location self._abort_if_unique_id_configured(updates=updates) - async def _try_connect(self) -> None: - """Try to connect and check auth.""" - for method in SUPPORTED_METHODS: - self._bridge = SamsungTVBridge.get_bridge(self.hass, method, self._host) - result = await self._bridge.async_try_connect() - if result == RESULT_SUCCESS: - return - if result != RESULT_CANNOT_CONNECT: - raise data_entry_flow.AbortFlow(result) - LOGGER.debug("No working config found") - raise data_entry_flow.AbortFlow(RESULT_CANNOT_CONNECT) + async def _async_create_bridge(self) -> None: + """Create the bridge.""" + result, method, _info = await self._async_get_device_info_and_method() + if result not in SUCCESSFUL_RESULTS: + LOGGER.debug("No working config found for %s", self._host) + raise data_entry_flow.AbortFlow(result) + assert method is not None + self._bridge = SamsungTVBridge.get_bridge(self.hass, method, self._host) + return + + async def _async_get_device_info_and_method( + self, + ) -> tuple[str, str | None, dict[str, Any] | None]: + """Get device info and method only once.""" + if self._connect_result is None: + result, _, method, info = await async_get_device_info(self.hass, self._host) + self._connect_result = result + self._method = method + self._device_info = info + if not method: + LOGGER.debug("Host:%s did not return device info", self._host) + return result, None, None + return self._connect_result, self._method, self._device_info async def _async_get_and_check_device_info(self) -> bool: """Try to get the device info.""" - _port, _method, info = await async_get_device_info( - self.hass, self._bridge, self._host - ) + result, _method, info = await self._async_get_device_info_and_method() + if result not in SUCCESSFUL_RESULTS: + raise data_entry_flow.AbortFlow(result) if not info: - if not _method: - LOGGER.debug( - "Samsung host %s is not supported by either %s or %s methods", - self._host, - METHOD_LEGACY, - METHOD_WEBSOCKET, - ) - raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) return False dev_info = info.get("device", {}) + assert dev_info is not None if (device_type := dev_info.get("type")) != "Samsung SmartTV": + LOGGER.debug( + "Host:%s has type: %s which is not supported", self._host, device_type + ) raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) self._model = dev_info.get("modelName") name = dev_info.get("name") @@ -169,7 +208,6 @@ async def _async_get_and_check_device_info(self) -> bool: partial(getmac.get_mac_address, ip=self._host) ): self._mac = mac - self._device_info = info return True async def async_step_import( @@ -209,16 +247,73 @@ async def async_step_user( """Handle a flow initialized by the user.""" if user_input is not None: await self._async_set_name_host_from_input(user_input) - await self._try_connect() + await self._async_create_bridge() assert self._bridge 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() + if self._bridge.method == METHOD_ENCRYPTED_WEBSOCKET: + return await self.async_step_encrypted_pairing() + return await self.async_step_pairing({}) return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA) + async def async_step_pairing( + self, user_input: dict[str, Any] | None = None + ) -> data_entry_flow.FlowResult: + """Handle a pairing by accepting the message on the TV.""" + assert self._bridge is not None + errors: dict[str, str] = {} + if user_input is not None: + result = await self._bridge.async_try_connect() + if result == RESULT_SUCCESS: + return self._get_entry_from_bridge() + if result != RESULT_AUTH_MISSING: + raise data_entry_flow.AbortFlow(result) + errors = {"base": RESULT_AUTH_MISSING} + + self.context["title_placeholders"] = {"device": self._title} + return self.async_show_form( + step_id="pairing", + errors=errors, + description_placeholders={"device": self._title}, + data_schema=vol.Schema({}), + ) + + async def async_step_encrypted_pairing( + self, user_input: dict[str, Any] | None = None + ) -> data_entry_flow.FlowResult: + """Handle a encrypted pairing.""" + assert self._host is not None + await self._async_start_encrypted_pairing(self._host) + assert self._authenticator is not None + errors: dict[str, str] = {} + + if user_input is not None: + if ( + (pin := user_input.get("pin")) + and (token := await self._authenticator.try_pin(pin)) + and (session_id := await self._authenticator.get_session_id_and_close()) + ): + return self.async_create_entry( + data={ + **self._base_config_entry(), + CONF_TOKEN: token, + CONF_SESSION_ID: session_id, + }, + title=self._title, + ) + errors = {"base": RESULT_INVALID_PIN} + + self.context["title_placeholders"] = {"device": self._title} + return self.async_show_form( + step_id="encrypted_pairing", + errors=errors, + description_placeholders={"device": self._title}, + data_schema=vol.Schema({vol.Required("pin"): str}), + ) + @callback def _async_get_existing_matching_entry( self, @@ -254,8 +349,21 @@ def _async_update_existing_matching_entry( or (is_unique_match and self.unique_id != entry.unique_id) ): entry_kw_args["unique_id"] = self.unique_id - if self._mac and not entry.data.get(CONF_MAC): - entry_kw_args["data"] = {**entry.data, CONF_MAC: self._mac} + data = entry.data + update_ssdp_rendering_control_location = ( + self._ssdp_rendering_control_location + and data.get(CONF_SSDP_RENDERING_CONTROL_LOCATION) + != self._ssdp_rendering_control_location + ) + update_mac = self._mac and not data.get(CONF_MAC) + if update_ssdp_rendering_control_location or update_mac: + entry_kw_args["data"] = {**entry.data} + if update_ssdp_rendering_control_location: + entry_kw_args["data"][ + CONF_SSDP_RENDERING_CONTROL_LOCATION + ] = self._ssdp_rendering_control_location + if update_mac: + entry_kw_args["data"][CONF_MAC] = self._mac if entry_kw_args: LOGGER.debug("Updating existing config entry with %s", entry_kw_args) self.hass.config_entries.async_update_entry(entry, **entry_kw_args) @@ -294,6 +402,11 @@ async def async_step_ssdp( """Handle a flow initialized by ssdp discovery.""" LOGGER.debug("Samsung device found via SSDP: %s", discovery_info) model_name: str = discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME) or "" + if discovery_info.ssdp_st == UPNP_SVC_RENDERINGCONTROL: + self._ssdp_rendering_control_location = discovery_info.ssdp_location + LOGGER.debug( + "Set SSDP location to: %s", self._ssdp_rendering_control_location + ) self._udn = self._upnp_udn = _strip_uuid( discovery_info.upnp[ssdp.ATTR_UPNP_UDN] ) @@ -345,12 +458,12 @@ async def async_step_confirm( ) -> data_entry_flow.FlowResult: """Handle user-confirmation of discovered node.""" if user_input is not None: - - await self._try_connect() + await self._async_create_bridge() assert self._bridge - return self._get_entry_from_bridge() + if self._bridge.method == METHOD_ENCRYPTED_WEBSOCKET: + return await self.async_step_encrypted_pairing() + return await self.async_step_pairing({}) - self._set_confirm_only() return self.async_show_form( step_id="confirm", description_placeholders={"device": self._title} ) @@ -378,6 +491,8 @@ async def async_step_reauth_confirm( assert self._reauth_entry method = self._reauth_entry.data[CONF_METHOD] if user_input is not None: + if method == METHOD_ENCRYPTED_WEBSOCKET: + return await self.async_step_reauth_confirm_encrypted() bridge = SamsungTVBridge.get_bridge( self.hass, method, @@ -399,40 +514,42 @@ async def async_step_reauth_confirm( 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=step_id, + step_id="reauth_confirm", errors=errors, description_placeholders={"device": self._title}, ) + async def _async_start_encrypted_pairing(self, host: str) -> None: + if self._authenticator is None: + self._authenticator = SamsungTVEncryptedWSAsyncAuthenticator( + host, + web_session=async_get_clientsession(self.hass), + ) + await self._authenticator.start_pairing() + 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() + await self._async_start_encrypted_pairing(self._reauth_entry.data[CONF_HOST]) + assert self._authenticator is not None - 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, - } + if user_input is not None: + if ( + (pin := user_input.get("pin")) + and (token := await self._authenticator.try_pin(pin)) + and (session_id := await self._authenticator.get_session_id_and_close()) + ): self.hass.config_entries.async_update_entry( - self._reauth_entry, data=new_data + self._reauth_entry, + data={ + **self._reauth_entry.data, + CONF_TOKEN: token, + CONF_SESSION_ID: session_id, + }, ) await self.hass.config_entries.async_reload(self._reauth_entry.entry_id) return self.async_abort(reason="reauth_successful") diff --git a/homeassistant/components/samsungtv/const.py b/homeassistant/components/samsungtv/const.py index 498b83a05393db..ad3300cd1e4848 100644 --- a/homeassistant/components/samsungtv/const.py +++ b/homeassistant/components/samsungtv/const.py @@ -15,6 +15,7 @@ CONF_DESCRIPTION = "description" CONF_MANUFACTURER = "manufacturer" CONF_MODEL = "model" +CONF_SSDP_RENDERING_CONTROL_LOCATION = "ssdp_rendering_control_location" CONF_ON_ACTION = "turn_on_action" CONF_SESSION_ID = "session_id" @@ -34,4 +35,10 @@ LEGACY_PORT = 55000 ENCRYPTED_WEBSOCKET_PORT = 8000 -WEBSOCKET_PORTS = (8002, 8001) +WEBSOCKET_NO_SSL_PORT = 8001 +WEBSOCKET_SSL_PORT = 8002 +WEBSOCKET_PORTS = (WEBSOCKET_SSL_PORT, WEBSOCKET_NO_SSL_PORT) + +SUCCESSFUL_RESULTS = {RESULT_AUTH_MISSING, RESULT_SUCCESS} + +UPNP_SVC_RENDERINGCONTROL = "urn:schemas-upnp-org:service:RenderingControl:1" diff --git a/homeassistant/components/samsungtv/manifest.json b/homeassistant/components/samsungtv/manifest.json index 2a5b2f76da3159..44309ddf92415f 100644 --- a/homeassistant/components/samsungtv/manifest.json +++ b/homeassistant/components/samsungtv/manifest.json @@ -11,6 +11,14 @@ "ssdp": [ { "st": "urn:samsung.com:device:RemoteControlReceiver:1" + }, + { + "manufacturer": "Samsung", + "st": "urn:schemas-upnp-org:service:RenderingControl:1" + }, + { + "manufacturer": "Samsung Electronics", + "st": "urn:schemas-upnp-org:service:RenderingControl:1" } ], "zeroconf": [ diff --git a/homeassistant/components/samsungtv/strings.json b/homeassistant/components/samsungtv/strings.json index f64620638bf3fd..d0f4526335aafd 100644 --- a/homeassistant/components/samsungtv/strings.json +++ b/homeassistant/components/samsungtv/strings.json @@ -12,11 +12,17 @@ "confirm": { "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." }, + "pairing": { + "description": "[%key:component::samsungtv::config::step::confirm::description%]" + }, "reauth_confirm": { "description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds or input PIN." }, - "reauth_confirm_encrypted": { + "encrypted_pairing": { "description": "Please enter the PIN displayed on {device}." + }, + "reauth_confirm_encrypted": { + "description": "[%key:component::samsungtv::config::step::encrypted_pairing::description%]" } }, "error": { @@ -34,4 +40,4 @@ "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 2e3dd88bec7a84..c4e0e181090b58 100644 --- a/homeassistant/components/samsungtv/translations/en.json +++ b/homeassistant/components/samsungtv/translations/en.json @@ -6,7 +6,6 @@ "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", "id_missing": "This Samsung device doesn't have a SerialNumber.", - "missing_config_entry": "This Samsung device doesn't have a configuration entry.", "not_supported": "This Samsung device is currently not supported.", "reauth_successful": "Re-authentication was successful", "unknown": "Unexpected error" @@ -18,8 +17,13 @@ "flow_title": "{device}", "step": { "confirm": { - "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.", - "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." + }, + "encrypted_pairing": { + "description": "Please enter the PIN displayed on {device}." + }, + "pairing": { + "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 or input PIN." diff --git a/homeassistant/generated/ssdp.py b/homeassistant/generated/ssdp.py index 40bb9bf295ff3e..10ea82949b56ca 100644 --- a/homeassistant/generated/ssdp.py +++ b/homeassistant/generated/ssdp.py @@ -225,6 +225,14 @@ "samsungtv": [ { "st": "urn:samsung.com:device:RemoteControlReceiver:1" + }, + { + "manufacturer": "Samsung", + "st": "urn:schemas-upnp-org:service:RenderingControl:1" + }, + { + "manufacturer": "Samsung Electronics", + "st": "urn:schemas-upnp-org:service:RenderingControl:1" } ], "songpal": [ diff --git a/tests/components/samsungtv/__init__.py b/tests/components/samsungtv/__init__.py index 7cace81f7a6a9c..1358e8e0bb5302 100644 --- a/tests/components/samsungtv/__init__.py +++ b/tests/components/samsungtv/__init__.py @@ -1,5 +1,5 @@ """Tests for the samsungtv component.""" - +from __future__ import annotations from homeassistant.components.samsungtv.const import DOMAIN from homeassistant.config_entries import ConfigEntry diff --git a/tests/components/samsungtv/conftest.py b/tests/components/samsungtv/conftest.py index 3fc11d7c07a77e..b602a3a9c52aaa 100644 --- a/tests/components/samsungtv/conftest.py +++ b/tests/components/samsungtv/conftest.py @@ -12,8 +12,10 @@ from samsungtvws.command import SamsungTVCommand from samsungtvws.encrypted.remote import SamsungTVEncryptedWSAsyncRemote from samsungtvws.event import ED_INSTALLED_APP_EVENT +from samsungtvws.exceptions import ResponseError from samsungtvws.remote import ChannelEmitCommand +from homeassistant.components.samsungtv.const import WEBSOCKET_SSL_PORT import homeassistant.util.dt as dt_util from .const import SAMPLE_DEVICE_INFO_WIFI @@ -47,7 +49,7 @@ def remote_fixture() -> Mock: yield remote -@pytest.fixture(name="rest_api", autouse=True) +@pytest.fixture(name="rest_api") def rest_api_fixture() -> Mock: """Patch the samsungtvws SamsungTVAsyncRest.""" with patch( @@ -60,6 +62,52 @@ def rest_api_fixture() -> Mock: yield rest_api_class.return_value +@pytest.fixture(name="rest_api_non_ssl_only") +def rest_api_fixture_non_ssl_only() -> Mock: + """Patch the samsungtvws SamsungTVAsyncRest non-ssl only.""" + + class MockSamsungTVAsyncRest: + """Mock for a MockSamsungTVAsyncRest.""" + + def __init__(self, host, session, port, timeout): + """Mock a MockSamsungTVAsyncRest.""" + self.port = port + self.host = host + + async def rest_device_info(self): + """Mock rest_device_info to fail for ssl and work for non-ssl.""" + if self.port == WEBSOCKET_SSL_PORT: + raise ResponseError + return SAMPLE_DEVICE_INFO_WIFI + + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest", + MockSamsungTVAsyncRest, + ): + yield + + +@pytest.fixture(name="rest_api_failing") +def rest_api_failure_fixture() -> Mock: + """Patch the samsungtvws SamsungTVAsyncRest.""" + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest", + autospec=True, + ) as rest_api_class: + rest_api_class.return_value.rest_device_info.side_effect = ResponseError + yield + + +@pytest.fixture(name="remoteencws_failing") +def remoteencws_failing_fixture(): + """Patch the samsungtvws SamsungTVEncryptedWSAsyncRemote.""" + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening", + side_effect=OSError, + ): + yield + + @pytest.fixture(name="remotews") def remotews_fixture() -> Mock: """Patch the samsungtvws SamsungTVWS.""" diff --git a/tests/components/samsungtv/test_config_flow.py b/tests/components/samsungtv/test_config_flow.py index 567b53646d7567..1620b46ee23889 100644 --- a/tests/components/samsungtv/test_config_flow.py +++ b/tests/components/samsungtv/test_config_flow.py @@ -5,7 +5,12 @@ import pytest from samsungctl.exceptions import AccessDenied, UnhandledResponse from samsungtvws.async_remote import SamsungTVWSAsyncRemote -from samsungtvws.exceptions import ConnectionFailure, HttpApiError +from samsungtvws.exceptions import ( + ConnectionFailure, + HttpApiError, + ResponseError, + UnauthorizedError, +) from websockets import frames from websockets.exceptions import ( ConnectionClosedError, @@ -19,6 +24,7 @@ CONF_MANUFACTURER, CONF_MODEL, CONF_SESSION_ID, + CONF_SSDP_RENDERING_CONTROL_LOCATION, DEFAULT_MANUFACTURER, DOMAIN, LEGACY_PORT, @@ -49,6 +55,11 @@ CONF_TOKEN, ) from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import ( + RESULT_TYPE_ABORT, + RESULT_TYPE_CREATE_ENTRY, + RESULT_TYPE_FORM, +) from homeassistant.setup import async_setup_component from . import setup_samsungtv_entry @@ -88,6 +99,18 @@ ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172de", }, ) + +MOCK_SSDP_DATA_RENDERING_CONTROL_ST = ssdp.SsdpServiceInfo( + ssdp_usn="mock_usn", + ssdp_st="urn:schemas-upnp-org:service:RenderingControl:1", + ssdp_location="https://fake_host:12345/test", + upnp={ + ATTR_UPNP_FRIENDLY_NAME: "[TV] fake_name", + ATTR_UPNP_MANUFACTURER: "Samsung fake_manufacturer", + ATTR_UPNP_MODEL_NAME: "fake_model", + ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172de", + }, +) MOCK_SSDP_DATA_NOPREFIX = ssdp.SsdpServiceInfo( ssdp_usn="mock_usn", ssdp_st="mock_st", @@ -194,9 +217,15 @@ "port": 8002, "timeout": TIMEOUT_WEBSOCKET, } +DEVICEINFO_WEBSOCKET_NO_SSL = { + "host": "fake_host", + "session": ANY, + "port": 8001, + "timeout": TIMEOUT_WEBSOCKET, +} -@pytest.mark.usefixtures("remote") +@pytest.mark.usefixtures("remote", "rest_api_failing") async def test_user_legacy(hass: HomeAssistant) -> None: """Test starting a flow by user.""" # show form @@ -221,7 +250,42 @@ async def test_user_legacy(hass: HomeAssistant) -> None: assert result["result"].unique_id is None -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("rest_api_failing") +async def test_user_legacy_does_not_ok_first_time(hass: HomeAssistant) -> None: + """Test starting a flow by user.""" + # show form + with patch( + "homeassistant.components.samsungtv.bridge.Remote", + side_effect=AccessDenied("Boom"), + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "form" + assert result["step_id"] == "user" + # entry was added + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_USER_DATA + ) + + with patch("homeassistant.components.samsungtv.bridge.Remote"): + # entry was added + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], user_input={} + ) + + # legacy tv entry created + assert result3["type"] == "create_entry" + assert result3["title"] == "fake_name" + assert result3["data"][CONF_HOST] == "fake_host" + assert result3["data"][CONF_NAME] == "fake_name" + assert result3["data"][CONF_METHOD] == "legacy" + assert result3["data"][CONF_MANUFACTURER] == DEFAULT_MANUFACTURER + assert result3["data"][CONF_MODEL] is None + assert result3["result"].unique_id is None + + +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_user_websocket(hass: HomeAssistant) -> None: """Test starting a flow by user.""" with patch( @@ -249,7 +313,59 @@ async def test_user_websocket(hass: HomeAssistant) -> None: assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remoteencws", "rest_api_non_ssl_only") +async def test_user_encrypted_websocket( + hass: HomeAssistant, +) -> None: + """Test starting a flow from ssdp for a supported device populates the mac.""" + # show form + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "form" + assert result["step_id"] == "user" + + 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" + + # entry was added + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=MOCK_USER_DATA + ) + assert result2["type"] == "form" + assert result2["step_id"] == "encrypted_pairing" + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], user_input={"pin": "invalid"} + ) + assert result3["step_id"] == "encrypted_pairing" + assert result3["errors"] == {"base": "invalid_pin"} + + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], user_input={"pin": "1234"} + ) + + assert result4["type"] == "create_entry" + assert result4["title"] == "Living Room (82GXARRS)" + assert result4["data"][CONF_HOST] == "fake_host" + assert result4["data"][CONF_NAME] == "Living Room" + assert result4["data"][CONF_MAC] == "aa:bb:ww:ii:ff:ii" + assert result4["data"][CONF_MANUFACTURER] == "Samsung" + assert result4["data"][CONF_MODEL] == "82GXARRS" + assert result4["data"][CONF_SSDP_RENDERING_CONTROL_LOCATION] is None + assert result4["data"][CONF_TOKEN] == "037739871315caef138547b03e348b72" + assert result4["data"][CONF_SESSION_ID] == "1" + assert result4["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +@pytest.mark.usefixtures("rest_api_failing") async def test_user_legacy_missing_auth(hass: HomeAssistant) -> None: """Test starting a flow by user with authentication.""" with patch( @@ -260,10 +376,23 @@ async def test_user_legacy_missing_auth(hass: HomeAssistant) -> None: 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"] == RESULT_AUTH_MISSING + assert result["type"] == RESULT_TYPE_FORM + assert result["step_id"] == "pairing" + assert result["errors"] == {"base": "auth_missing"} + + with patch( + "homeassistant.components.samsungtv.bridge.Remote", + side_effect=OSError, + ): + # legacy device fails to connect after auth failed + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result2["type"] == RESULT_TYPE_ABORT + assert result2["reason"] == RESULT_CANNOT_CONNECT +@pytest.mark.usefixtures("rest_api_failing") async def test_user_legacy_not_supported(hass: HomeAssistant) -> None: """Test starting a flow by user for not supported device.""" with patch( @@ -278,6 +407,7 @@ async def test_user_legacy_not_supported(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_NOT_SUPPORTED +@pytest.mark.usefixtures("rest_api", "remoteencws_failing") async def test_user_websocket_not_supported(hass: HomeAssistant) -> None: """Test starting a flow by user for not supported device.""" with patch( @@ -295,6 +425,7 @@ async def test_user_websocket_not_supported(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_NOT_SUPPORTED +@pytest.mark.usefixtures("rest_api", "remoteencws_failing") async def test_user_websocket_access_denied( hass: HomeAssistant, caplog: pytest.LogCaptureFixture ) -> None: @@ -315,6 +446,42 @@ async def test_user_websocket_access_denied( assert "Please check the Device Connection Manager on your TV" in caplog.text +@pytest.mark.usefixtures("rest_api", "remoteencws_failing") +async def test_user_websocket_auth_retry(hass: HomeAssistant) -> None: + """Test starting a flow by user for not supported device.""" + with patch( + "homeassistant.components.samsungtv.bridge.Remote", + side_effect=OSError("Boom"), + ), patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote.open", + side_effect=UnauthorizedError, + ): + # 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"] == RESULT_TYPE_FORM + assert result["step_id"] == "pairing" + assert result["errors"] == {"base": "auth_missing"} + with patch( + "homeassistant.components.samsungtv.bridge.Remote", + side_effect=OSError("Boom"), + ), patch( + "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote.open", + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == 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_MANUFACTURER] == "Samsung" + assert result["data"][CONF_MODEL] == "82GXARRS" + assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +@pytest.mark.usefixtures("rest_api_failing") async def test_user_not_successful(hass: HomeAssistant) -> None: """Test starting a flow by user but no connection found.""" with patch( @@ -331,6 +498,7 @@ async def test_user_not_successful(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_CANNOT_CONNECT +@pytest.mark.usefixtures("rest_api_failing") async def test_user_not_successful_2(hass: HomeAssistant) -> None: """Test starting a flow by user but no connection found.""" with patch( @@ -347,75 +515,60 @@ async def test_user_not_successful_2(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_CANNOT_CONNECT -@pytest.mark.usefixtures("remote") +@pytest.mark.usefixtures("remote", "rest_api_failing") async def test_ssdp(hass: HomeAssistant) -> None: """Test starting a flow from discovery.""" - with patch( - "homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info", - return_value=MOCK_DEVICE_INFO, - ): - # confirm to add the entry - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA - ) - assert result["type"] == "form" - assert result["step_id"] == "confirm" + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" - # entry was added - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input="whatever" - ) - assert result["type"] == "create_entry" - assert result["title"] == "fake_name (fake_model)" - assert result["data"][CONF_HOST] == "fake_host" - assert result["data"][CONF_NAME] == "fake_name" - assert result["data"][CONF_MANUFACTURER] == "Samsung fake_manufacturer" - assert result["data"][CONF_MODEL] == "fake_model" - assert result["result"].unique_id == "123" + # entry was added + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input="whatever" + ) + assert result["type"] == "create_entry" + assert result["title"] == "fake_model" + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] == "fake_model" + assert result["data"][CONF_MANUFACTURER] == "Samsung fake_manufacturer" + assert result["data"][CONF_MODEL] == "fake_model" + assert result["result"].unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de" -@pytest.mark.usefixtures("remote") +@pytest.mark.usefixtures("remote", "rest_api_failing") async def test_ssdp_noprefix(hass: HomeAssistant) -> None: """Test starting a flow from discovery without prefixes.""" - with patch( - "homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info", - return_value=MOCK_DEVICE_INFO_2, - ): - # confirm to add the entry - result = await hass.config_entries.flow.async_init( - DOMAIN, - context={"source": config_entries.SOURCE_SSDP}, - data=MOCK_SSDP_DATA_NOPREFIX, - ) - assert result["type"] == "form" - assert result["step_id"] == "confirm" - - with patch( - "homeassistant.components.samsungtv.bridge.Remote.__enter__", - return_value=True, - ): - - # entry was added - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input="whatever" - ) - assert result["type"] == "create_entry" - assert result["title"] == "fake2_name (fake2_model)" - assert result["data"][CONF_HOST] == "fake2_host" - assert result["data"][CONF_NAME] == "fake2_name" - assert result["data"][CONF_MANUFACTURER] == "Samsung fake2_manufacturer" - assert result["data"][CONF_MODEL] == "fake2_model" - assert result["result"].unique_id == "345" + # confirm to add the entry + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_NOPREFIX, + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + # entry was added + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input="whatever" + ) + assert result["type"] == "create_entry" + assert result["title"] == "fake2_model" + assert result["data"][CONF_HOST] == "fake2_host" + assert result["data"][CONF_NAME] == "fake2_model" + assert result["data"][CONF_MANUFACTURER] == "Samsung fake2_manufacturer" + assert result["data"][CONF_MODEL] == "fake2_model" + assert result["result"].unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172df" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api_failing") async def test_ssdp_legacy_missing_auth(hass: HomeAssistant) -> None: """Test starting a flow from discovery with authentication.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=AccessDenied("Boom"), ): - # confirm to add the entry result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA @@ -424,48 +577,51 @@ async def test_ssdp_legacy_missing_auth(hass: HomeAssistant) -> None: assert result["step_id"] == "confirm" # missing authentication + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == RESULT_TYPE_FORM + assert result["step_id"] == "pairing" + assert result["errors"] == {"base": "auth_missing"} - with patch( - "homeassistant.components.samsungtv.bridge.SamsungTVLegacyBridge.async_try_connect", - return_value=RESULT_AUTH_MISSING, - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input="whatever" - ) - assert result["type"] == "abort" - assert result["reason"] == RESULT_AUTH_MISSING + with patch("homeassistant.components.samsungtv.bridge.Remote"): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] == RESULT_TYPE_CREATE_ENTRY + assert result["title"] == "fake_model" + assert result["data"][CONF_HOST] == "fake_host" + assert result["data"][CONF_NAME] == "fake_model" + assert result["data"][CONF_MANUFACTURER] == "Samsung fake_manufacturer" + assert result["data"][CONF_MODEL] == "fake_model" + assert result["result"].unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de" -@pytest.mark.usefixtures("remote", "remotews") + +@pytest.mark.usefixtures("remotews", "rest_api_failing") async def test_ssdp_legacy_not_supported(hass: HomeAssistant) -> None: """Test starting a flow from discovery for not supported device.""" - - # confirm to add the entry - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA - ) - assert result["type"] == "form" - assert result["step_id"] == "confirm" - with patch( "homeassistant.components.samsungtv.bridge.SamsungTVLegacyBridge.async_try_connect", return_value=RESULT_NOT_SUPPORTED, ): - # device not supported - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input="whatever" + # confirm to add the entry + 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["type"] == RESULT_TYPE_ABORT assert result["reason"] == RESULT_NOT_SUPPORTED -@pytest.mark.usefixtures("remote", "remotews") -async def test_ssdp_websocket_success_populates_mac_address( +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") +async def test_ssdp_websocket_success_populates_mac_address_and_ssdp_location( hass: HomeAssistant, ) -> None: """Test starting a flow from ssdp for a supported device populates the mac.""" result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST, ) assert result["type"] == "form" assert result["step_id"] == "confirm" @@ -480,13 +636,89 @@ async def test_ssdp_websocket_success_populates_mac_address( assert result["data"][CONF_MAC] == "aa:bb:ww:ii:ff:ii" assert result["data"][CONF_MANUFACTURER] == "Samsung fake_manufacturer" assert result["data"][CONF_MODEL] == "82GXARRS" + assert ( + result["data"][CONF_SSDP_RENDERING_CONTROL_LOCATION] + == "https://fake_host:12345/test" + ) assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -async def test_ssdp_websocket_not_supported( +@pytest.mark.usefixtures("remoteencws", "rest_api_non_ssl_only") +async def test_ssdp_encrypted_websocket_success_populates_mac_address_and_ssdp_location( + hass: HomeAssistant, +) -> None: + """Test starting a flow from ssdp for a supported device populates the mac.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST, + ) + assert result["type"] == "form" + assert result["step_id"] == "confirm" + + 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" + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result2["step_id"] == "encrypted_pairing" + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], user_input={"pin": "invalid"} + ) + assert result3["step_id"] == "encrypted_pairing" + assert result3["errors"] == {"base": "invalid_pin"} + + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], user_input={"pin": "1234"} + ) + + assert result4["type"] == "create_entry" + assert result4["title"] == "Living Room (82GXARRS)" + assert result4["data"][CONF_HOST] == "fake_host" + assert result4["data"][CONF_NAME] == "Living Room" + assert result4["data"][CONF_MAC] == "aa:bb:ww:ii:ff:ii" + assert result4["data"][CONF_MANUFACTURER] == "Samsung fake_manufacturer" + assert result4["data"][CONF_MODEL] == "82GXARRS" + assert ( + result4["data"][CONF_SSDP_RENDERING_CONTROL_LOCATION] + == "https://fake_host:12345/test" + ) + assert result4["data"][CONF_TOKEN] == "037739871315caef138547b03e348b72" + assert result4["data"][CONF_SESSION_ID] == "1" + assert result4["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +@pytest.mark.usefixtures("rest_api_non_ssl_only") +async def test_ssdp_encrypted_websocket_not_supported( + hass: HomeAssistant, +) -> None: + """Test starting a flow from ssdp for an unsupported device populates the mac.""" + with patch( + "homeassistant.components.samsungtv.bridge.SamsungTVEncryptedWSAsyncRemote.start_listening", + side_effect=WebSocketException, + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST, + ) + assert result["type"] == RESULT_TYPE_ABORT + assert result["reason"] == RESULT_NOT_SUPPORTED + + +async def test_ssdp_websocket_cannot_connect( hass: HomeAssistant, rest_api: Mock ) -> None: - """Test starting a flow from discovery for not supported device.""" + """Test starting a flow from discovery and we cannot connect.""" rest_api.rest_device_info.return_value = None with patch( "homeassistant.components.samsungtv.bridge.Remote", @@ -504,7 +736,7 @@ async def test_ssdp_websocket_not_supported( DOMAIN, context={"source": config_entries.SOURCE_SSDP}, data=MOCK_SSDP_DATA ) assert result["type"] == "abort" - assert result["reason"] == RESULT_NOT_SUPPORTED + assert result["reason"] == RESULT_CANNOT_CONNECT @pytest.mark.usefixtures("remote") @@ -521,6 +753,7 @@ async def test_ssdp_model_not_supported(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_NOT_SUPPORTED +@pytest.mark.usefixtures("remoteencws_failing") async def test_ssdp_not_successful(hass: HomeAssistant) -> None: """Test starting a flow from discovery but no device found.""" with patch( @@ -549,6 +782,7 @@ async def test_ssdp_not_successful(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_CANNOT_CONNECT +@pytest.mark.usefixtures("remoteencws_failing") async def test_ssdp_not_successful_2(hass: HomeAssistant) -> None: """Test starting a flow from discovery but no device found.""" with patch( @@ -577,7 +811,7 @@ async def test_ssdp_not_successful_2(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_CANNOT_CONNECT -@pytest.mark.usefixtures("remote") +@pytest.mark.usefixtures("remote", "remoteencws_failing") async def test_ssdp_already_in_progress(hass: HomeAssistant) -> None: """Test starting a flow from discovery twice.""" with patch( @@ -600,7 +834,7 @@ async def test_ssdp_already_in_progress(hass: HomeAssistant) -> None: assert result["reason"] == RESULT_ALREADY_IN_PROGRESS -@pytest.mark.usefixtures("remote") +@pytest.mark.usefixtures("remotews", "remoteencws_failing") async def test_ssdp_already_configured(hass: HomeAssistant) -> None: """Test starting a flow from discovery when already configured.""" with patch( @@ -615,8 +849,8 @@ async def test_ssdp_already_configured(hass: HomeAssistant) -> None: assert result["type"] == "create_entry" entry = result["result"] assert entry.data[CONF_MANUFACTURER] == DEFAULT_MANUFACTURER - assert entry.data[CONF_MODEL] is None - assert entry.unique_id is None + assert entry.data[CONF_MODEL] == "fake_model" + assert entry.unique_id == "123" # failed as already configured result2 = await hass.config_entries.flow.async_init( @@ -677,7 +911,7 @@ async def test_import_legacy_without_name(hass: HomeAssistant, rest_api: Mock) - assert entries[0].data[CONF_PORT] == LEGACY_PORT -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api") async def test_import_websocket(hass: HomeAssistant): """Test importing from yaml with hostname.""" result = await hass.config_entries.flow.async_init( @@ -716,7 +950,7 @@ async def test_import_websocket_encrypted(hass: HomeAssistant): assert result["result"].unique_id is None -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api") async def test_import_websocket_without_port(hass: HomeAssistant): """Test importing from yaml with hostname by no port.""" result = await hass.config_entries.flow.async_init( @@ -755,7 +989,7 @@ async def test_import_unknown_host(hass: HomeAssistant): assert result["reason"] == RESULT_UNKNOWN_HOST -@pytest.mark.usefixtures("remote", "remotews") +@pytest.mark.usefixtures("remotews", "rest_api_non_ssl_only", "remoteencws_failing") async def test_dhcp_wireless(hass: HomeAssistant) -> None: """Test starting a flow from dhcp.""" # confirm to add the entry @@ -782,7 +1016,7 @@ async def test_dhcp_wireless(hass: HomeAssistant) -> None: assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remote", "remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_dhcp_wired(hass: HomeAssistant, rest_api: Mock) -> None: """Test starting a flow from dhcp.""" # Even though it is named "wifiMac", it matches the mac of the wired connection @@ -811,7 +1045,7 @@ async def test_dhcp_wired(hass: HomeAssistant, rest_api: Mock) -> None: assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remote", "remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_zeroconf(hass: HomeAssistant) -> None: """Test starting a flow from zeroconf.""" result = await hass.config_entries.flow.async_init( @@ -837,7 +1071,7 @@ async def test_zeroconf(hass: HomeAssistant) -> None: assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "remoteencws_failing") async def test_zeroconf_ignores_soundbar(hass: HomeAssistant, rest_api: Mock) -> None: """Test starting a flow from zeroconf where the device is actually a soundbar.""" rest_api.rest_device_info.return_value = { @@ -874,7 +1108,7 @@ async def test_zeroconf_no_device_info(hass: HomeAssistant, rest_api: Mock) -> N assert result["reason"] == "not_supported" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_zeroconf_and_dhcp_same_time(hass: HomeAssistant) -> None: """Test starting a flow from zeroconf and dhcp.""" result = await hass.config_entries.flow.async_init( @@ -896,6 +1130,7 @@ async def test_zeroconf_and_dhcp_same_time(hass: HomeAssistant) -> None: assert result2["reason"] == "already_in_progress" +@pytest.mark.usefixtures("remoteencws_failing") async def test_autodetect_websocket(hass: HomeAssistant) -> None: """Test for send key with autodetection of protocol.""" with patch( @@ -941,6 +1176,7 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None: assert entries[0].data[CONF_MAC] == "aa:bb:cc:dd:ee:ff" +@pytest.mark.usefixtures("remoteencws_failing") async def test_websocket_no_mac(hass: HomeAssistant, mac_address: Mock) -> None: """Test for send key with autodetection of protocol.""" mac_address.return_value = "gg:ee:tt:mm:aa:cc" @@ -987,21 +1223,36 @@ async def test_websocket_no_mac(hass: HomeAssistant, mac_address: Mock) -> None: assert entries[0].data[CONF_MAC] == "gg:ee:tt:mm:aa:cc" +@pytest.mark.usefixtures("rest_api_failing") async def test_autodetect_auth_missing(hass: HomeAssistant) -> None: """Test for send key with autodetection of protocol.""" with patch( "homeassistant.components.samsungtv.bridge.Remote", - side_effect=[AccessDenied("Boom")], + side_effect=AccessDenied("Boom"), ) as remote: 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"] == RESULT_AUTH_MISSING - assert remote.call_count == 1 - assert remote.call_args_list == [call(AUTODETECT_LEGACY)] + assert result["type"] == RESULT_TYPE_FORM + assert result["step_id"] == "pairing" + assert result["errors"] == {"base": "auth_missing"} + + assert remote.call_count == 2 + assert remote.call_args_list == [ + call(AUTODETECT_LEGACY), + call(AUTODETECT_LEGACY), + ] + with patch("homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + {}, + ) + await hass.async_block_till_done() + assert result2["type"] == RESULT_TYPE_ABORT + assert result2["reason"] == RESULT_CANNOT_CONNECT +@pytest.mark.usefixtures("rest_api_failing") async def test_autodetect_not_supported(hass: HomeAssistant) -> None: """Test for send key with autodetection of protocol.""" with patch( @@ -1017,7 +1268,7 @@ async def test_autodetect_not_supported(hass: HomeAssistant) -> None: assert remote.call_args_list == [call(AUTODETECT_LEGACY)] -@pytest.mark.usefixtures("remote") +@pytest.mark.usefixtures("remote", "rest_api_failing") async def test_autodetect_legacy(hass: HomeAssistant) -> None: """Test for send key with autodetection of protocol.""" result = await hass.config_entries.flow.async_init( @@ -1032,18 +1283,13 @@ async def test_autodetect_legacy(hass: HomeAssistant) -> None: async def test_autodetect_none(hass: HomeAssistant) -> None: """Test for send key with autodetection of protocol.""" - mock_remotews = Mock() - mock_remotews.__aenter__ = AsyncMock(return_value=mock_remotews) - mock_remotews.__aexit__ = AsyncMock() - mock_remotews.open = Mock(side_effect=OSError("Boom")) - with patch( "homeassistant.components.samsungtv.bridge.Remote", side_effect=OSError("Boom"), ) as remote, patch( - "homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote", - return_value=mock_remotews, - ) as remotews: + "homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest.rest_device_info", + side_effect=ResponseError, + ) as rest_device_info: result = await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA ) @@ -1053,56 +1299,43 @@ async def test_autodetect_none(hass: HomeAssistant) -> None: assert remote.call_args_list == [ call(AUTODETECT_LEGACY), ] - assert remotews.call_count == 2 - assert remotews.call_args_list == [ - call(**AUTODETECT_WEBSOCKET_SSL), - call(**AUTODETECT_WEBSOCKET_PLAIN), - ] + assert rest_device_info.call_count == 2 -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_update_old_entry(hass: HomeAssistant) -> None: """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) + 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] == EXISTING_IP - assert not entry.unique_id + 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] == EXISTING_IP + assert not entry.unique_id - assert await async_setup_component(hass, DOMAIN, {}) is True - await hass.async_block_till_done() + 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 + # 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] + 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 == "be9554b9-c9fb-41f4-8920-22da015376a4" + # 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 == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_update_missing_mac_unique_id_added_from_dhcp( hass: HomeAssistant, ) -> None: @@ -1131,7 +1364,7 @@ async def test_update_missing_mac_unique_id_added_from_dhcp( assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_update_missing_mac_unique_id_added_from_zeroconf( hass: HomeAssistant, ) -> None: @@ -1159,11 +1392,11 @@ async def test_update_missing_mac_unique_id_added_from_zeroconf( assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") -async def test_update_missing_mac_unique_id_added_from_ssdp( +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") +async def test_update_missing_mac_unique_id_ssdp_location_added_from_ssdp( hass: HomeAssistant, ) -> None: - """Test missing mac and unique id added via ssdp.""" + """Test missing mac, ssdp_location, and unique id added via ssdp.""" entry = MockConfigEntry(domain=DOMAIN, data=MOCK_OLD_ENTRY, unique_id=None) entry.add_to_hass(hass) with patch( @@ -1185,10 +1418,131 @@ async def test_update_missing_mac_unique_id_added_from_ssdp( assert result["type"] == "abort" assert result["reason"] == "already_configured" assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii" + # Wrong st + assert CONF_SSDP_RENDERING_CONTROL_LOCATION not in entry.data assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") +async def test_update_missing_mac_unique_id_added_ssdp_location_updated_from_ssdp( + hass: HomeAssistant, +) -> None: + """Test missing mac and unique id with outdated ssdp_location with the wrong st added via ssdp.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={ + **MOCK_OLD_ENTRY, + CONF_SSDP_RENDERING_CONTROL_LOCATION: "https://1.2.3.4:555/test", + }, + unique_id=None, + ) + entry.add_to_hass(hass) + with patch( + "homeassistant.components.samsungtv.async_setup", + return_value=True, + ) as mock_setup, patch( + "homeassistant.components.samsungtv.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA, + ) + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii" + # Wrong ST, ssdp location should not change + assert ( + entry.data[CONF_SSDP_RENDERING_CONTROL_LOCATION] == "https://1.2.3.4:555/test" + ) + assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") +async def test_update_missing_mac_unique_id_added_ssdp_location_rendering_st_updated_from_ssdp( + hass: HomeAssistant, +) -> None: + """Test missing mac and unique id with outdated ssdp_location with the correct st added via ssdp.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={ + **MOCK_OLD_ENTRY, + CONF_SSDP_RENDERING_CONTROL_LOCATION: "https://1.2.3.4:555/test", + }, + unique_id=None, + ) + entry.add_to_hass(hass) + with patch( + "homeassistant.components.samsungtv.async_setup", + return_value=True, + ) as mock_setup, patch( + "homeassistant.components.samsungtv.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST, + ) + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii" + # Correct ST, ssdp location should change + assert ( + entry.data[CONF_SSDP_RENDERING_CONTROL_LOCATION] + == "https://fake_host:12345/test" + ) + assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") +async def test_update_ssdp_location_rendering_st_updated_from_ssdp( + hass: HomeAssistant, +) -> None: + """Test with outdated ssdp_location with the correct st added via ssdp.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={**MOCK_OLD_ENTRY, CONF_MAC: "aa:bb:ww:ii:ff:ii"}, + unique_id="be9554b9-c9fb-41f4-8920-22da015376a4", + ) + entry.add_to_hass(hass) + with patch( + "homeassistant.components.samsungtv.async_setup", + return_value=True, + ) as mock_setup, patch( + "homeassistant.components.samsungtv.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST, + ) + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii" + # Correct ST, ssdp location should be added + assert ( + entry.data[CONF_SSDP_RENDERING_CONTROL_LOCATION] + == "https://fake_host:12345/test" + ) + assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +@pytest.mark.usefixtures("remotews", "rest_api") async def test_update_missing_mac_added_unique_id_preserved_from_zeroconf( hass: HomeAssistant, ) -> None: @@ -1292,6 +1646,79 @@ async def test_update_legacy_missing_mac_from_dhcp_no_unique_id( assert entry.unique_id is None +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") +async def test_update_ssdp_location_unique_id_added_from_ssdp( + hass: HomeAssistant, +) -> None: + """Test missing ssdp_location, and unique id added via ssdp.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={**MOCK_OLD_ENTRY, CONF_MAC: "aa:bb:ww:ii:ff:ii"}, + unique_id=None, + ) + entry.add_to_hass(hass) + with patch( + "homeassistant.components.samsungtv.async_setup", + return_value=True, + ) as mock_setup, patch( + "homeassistant.components.samsungtv.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA, + ) + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii" + # Wrong st + assert CONF_SSDP_RENDERING_CONTROL_LOCATION not in entry.data + assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") +async def test_update_ssdp_location_unique_id_added_from_ssdp_with_rendering_control_st( + hass: HomeAssistant, +) -> None: + """Test missing ssdp_location, and unique id added via ssdp with rendering control st.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={**MOCK_OLD_ENTRY, CONF_MAC: "aa:bb:ww:ii:ff:ii"}, + unique_id=None, + ) + entry.add_to_hass(hass) + with patch( + "homeassistant.components.samsungtv.async_setup", + return_value=True, + ) as mock_setup, patch( + "homeassistant.components.samsungtv.async_setup_entry", + return_value=True, + ) as mock_setup_entry: + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_SSDP}, + data=MOCK_SSDP_DATA_RENDERING_CONTROL_ST, + ) + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + assert result["type"] == "abort" + assert result["reason"] == "already_configured" + assert entry.data[CONF_MAC] == "aa:bb:ww:ii:ff:ii" + # Correct st + assert ( + entry.data[CONF_SSDP_RENDERING_CONTROL_LOCATION] + == "https://fake_host:12345/test" + ) + assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" + + @pytest.mark.usefixtures("remote") async def test_form_reauth_legacy(hass: HomeAssistant) -> None: """Test reauthenticate legacy.""" @@ -1314,7 +1741,7 @@ async def test_form_reauth_legacy(hass: HomeAssistant) -> None: assert result2["reason"] == "reauth_successful" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api") async def test_form_reauth_websocket(hass: HomeAssistant) -> None: """Test reauthenticate websocket.""" entry = MockConfigEntry(domain=DOMAIN, data=MOCK_WS_ENTRY) @@ -1339,6 +1766,7 @@ async def test_form_reauth_websocket(hass: HomeAssistant) -> None: assert entry.state == config_entries.ConfigEntryState.LOADED +@pytest.mark.usefixtures("rest_api") async def test_form_reauth_websocket_cannot_connect( hass: HomeAssistant, remotews: Mock ) -> None: @@ -1399,7 +1827,7 @@ async def test_form_reauth_websocket_not_supported(hass: HomeAssistant) -> None: assert result2["reason"] == "not_supported" -@pytest.mark.usefixtures("remoteencws") +@pytest.mark.usefixtures("remoteencws", "rest_api") async def test_form_reauth_encrypted(hass: HomeAssistant) -> None: """Test reauth flow for encrypted TVs.""" encrypted_entry_data = {**MOCK_ENTRYDATA_ENCRYPTED_WS} @@ -1429,13 +1857,13 @@ async def test_form_reauth_encrypted(hass: HomeAssistant) -> None: result = await hass.config_entries.flow.async_configure(result["flow_id"]) assert result["type"] == "form" - assert result["step_id"] == "reauth_confirm_encrypted" + assert result["step_id"] == "reauth_confirm" 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 + result["flow_id"], user_input={} ) assert result["type"] == "form" assert result["step_id"] == "reauth_confirm_encrypted" @@ -1471,7 +1899,7 @@ async def test_form_reauth_encrypted(hass: HomeAssistant) -> None: assert entry.data[CONF_SESSION_ID] == "1" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_update_incorrect_udn_matching_upnp_udn_unique_id_added_from_ssdp( hass: HomeAssistant, ) -> None: @@ -1504,7 +1932,7 @@ async def test_update_incorrect_udn_matching_upnp_udn_unique_id_added_from_ssdp( assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_update_incorrect_udn_matching_mac_unique_id_added_from_ssdp( hass: HomeAssistant, ) -> None: @@ -1537,7 +1965,7 @@ async def test_update_incorrect_udn_matching_mac_unique_id_added_from_ssdp( assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_update_incorrect_udn_matching_mac_from_dhcp( hass: HomeAssistant, ) -> None: @@ -1571,7 +1999,7 @@ async def test_update_incorrect_udn_matching_mac_from_dhcp( assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4" -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing") async def test_no_update_incorrect_udn_not_matching_mac_from_dhcp( hass: HomeAssistant, ) -> None: diff --git a/tests/components/samsungtv/test_diagnostics.py b/tests/components/samsungtv/test_diagnostics.py index 88fb98a66dbf52..65f5126808126d 100644 --- a/tests/components/samsungtv/test_diagnostics.py +++ b/tests/components/samsungtv/test_diagnostics.py @@ -19,7 +19,7 @@ from tests.components.diagnostics import get_diagnostics_for_config_entry -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "rest_api") async def test_entry_diagnostics( hass: HomeAssistant, hass_client: ClientSession ) -> None: diff --git a/tests/components/samsungtv/test_init.py b/tests/components/samsungtv/test_init.py index 6fec07fab846ed..239840f8c8b262 100644 --- a/tests/components/samsungtv/test_init.py +++ b/tests/components/samsungtv/test_init.py @@ -56,7 +56,12 @@ } -@pytest.mark.usefixtures("remotews") +@pytest.fixture(name="autouse_rest_api", autouse=True) +def autouse_rest_api(rest_api) -> Mock: + """Enable auto use of the rest api fixture for these tests.""" + + +@pytest.mark.usefixtures("remotews", "remoteencws_failing") async def test_setup(hass: HomeAssistant) -> None: """Test Samsung TV integration is setup.""" await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) @@ -98,7 +103,7 @@ async def test_setup_from_yaml_without_port_device_offline(hass: HomeAssistant) assert config_entries_domain[0].state == ConfigEntryState.SETUP_RETRY -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "remoteencws_failing") async def test_setup_from_yaml_without_port_device_online(hass: HomeAssistant) -> None: """Test import from yaml when the device is online.""" await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) @@ -127,7 +132,7 @@ async def test_setup_duplicate_config( assert "duplicate host entries found" in caplog.text -@pytest.mark.usefixtures("remote", "remotews") +@pytest.mark.usefixtures("remote", "remotews", "remoteencws_failing") async def test_setup_duplicate_entries(hass: HomeAssistant) -> None: """Test duplicate setup of platform.""" await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG) @@ -138,7 +143,7 @@ async def test_setup_duplicate_entries(hass: HomeAssistant) -> None: assert len(hass.states.async_all("media_player")) == 1 -@pytest.mark.usefixtures("remotews") +@pytest.mark.usefixtures("remotews", "remoteencws_failing") async def test_setup_h_j_model( hass: HomeAssistant, rest_api: Mock, caplog: pytest.LogCaptureFixture ) -> None: diff --git a/tests/components/samsungtv/test_media_player.py b/tests/components/samsungtv/test_media_player.py index a0b712e575b53e..346d04e9a6501f 100644 --- a/tests/components/samsungtv/test_media_player.py +++ b/tests/components/samsungtv/test_media_player.py @@ -141,6 +141,11 @@ } +@pytest.fixture(name="autouse_rest_api", autouse=True) +def autouse_rest_api(rest_api) -> Mock: + """Enable auto use of the rest api fixture for these tests.""" + + @pytest.fixture(name="delay") def delay_fixture(): """Patch the delay script function."""