Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add basic support for SamsungTV encrypted models (#68500) 

Co-authored-by: epenet <epenet@users.noreply.github.com>
pull/68542/head
epenet 2022-03-22 11:11:41 +01:00 committed by GitHub
parent 9180243a54
commit 87378016c1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 766 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
homeassistant/components/samsungtv/__init__.py
View File

@ -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:

212
homeassistant/components/samsungtv/bridge.py
View File

@ -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__()
)

58
homeassistant/components/samsungtv/config_flow.py
View File

@ -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}),
)

4
homeassistant/components/samsungtv/const.py
View File

@ -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)

4
homeassistant/components/samsungtv/diagnostics.py
View File

@ -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(

19
homeassistant/components/samsungtv/media_player.py
View File

@ -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),

8
homeassistant/components/samsungtv/strings.json
View File

@ -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%]",

8
homeassistant/components/samsungtv/translations/en.json
View File

@ -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": {

22
tests/components/samsungtv/__init__.py
View File

@ -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

27
tests/components/samsungtv/conftest.py
View File

@ -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."""

54
tests/components/samsungtv/const.py
View File

@ -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/",
}

126
tests/components/samsungtv/test_config_flow.py
View File

@ -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,

102
tests/components/samsungtv/test_diagnostics.py
View File

@ -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,
}

3
tests/components/samsungtv/test_init.py
View File

@ -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,

149
tests/components/samsungtv/test_media_player.py
View File

@ -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