github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-08-06 09:34:49 +02:00
Code Releases Activity

Add mac address to samsungtv config entry data if missing (#51634)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
J. Nick Koston 2021-06-24 08:15:16 -10:00 committed by GitHub
parent 34a317b847
commit 5695710463
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 293 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -5,8 +5,10 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.media_player.const import DOMAIN as MP_DOMAIN
from homeassistant.config_entries import ConfigEntryNotReady
from homeassistant.const import (
CONF_HOST,
CONF_MAC,
CONF_METHOD,
CONF_NAME,
CONF_PORT,
@ -16,8 +18,16 @@ from homeassistant.const import (
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from .bridge import SamsungTVBridge
from .const import CONF_ON_ACTION, DEFAULT_NAME, DOMAIN, LOGGER
from .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info
from .const import (
CONF_ON_ACTION,
DEFAULT_NAME,
DOMAIN,
LEGACY_PORT,
LOGGER,
METHOD_LEGACY,
METHOD_WEBSOCKET,
)
def ensure_unique_hosts(value):
@ -90,13 +100,7 @@ async def async_setup_entry(hass, entry):
"""Set up the Samsung TV platform."""
# Initialize bridge
data = entry.data.copy()
bridge = _async_get_device_bridge(data)
if bridge.port is None and bridge.default_port is not None:
# For backward compat, set default port for websocket tv
data[CONF_PORT] = bridge.default_port
hass.config_entries.async_update_entry(entry, data=data)
bridge = _async_get_device_bridge(data)
bridge = await _async_create_bridge_with_updated_data(hass, entry)
def stop_bridge(event):
"""Stop SamsungTV bridge connection."""
@ -111,6 +115,46 @@ async def async_setup_entry(hass, entry):
return True
async def _async_create_bridge_with_updated_data(hass, entry):
"""Create a bridge object and update any missing data in the config entry."""
updated_data = {}
host = entry.data[CONF_HOST]
port = entry.data.get(CONF_PORT)
method = entry.data.get(CONF_METHOD)
info = None
if not port or not method:
if method == METHOD_LEGACY:
port = LEGACY_PORT
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)
if not port:
raise ConfigEntryNotReady(
"Failed to determine connection method, make sure the device is on."
)
updated_data[CONF_PORT] = port
updated_data[CONF_METHOD] = method
bridge = _async_get_device_bridge({**entry.data, **updated_data})
if not entry.data.get(CONF_MAC) and bridge.method == METHOD_WEBSOCKET:
if info:
mac = mac_from_device_info(info)
else:
mac = await hass.async_add_executor_job(bridge.mac_from_device)
if mac:
updated_data[CONF_MAC] = mac
if updated_data:
data = {**entry.data, **updated_data}
hass.config_entries.async_update_entry(entry, data=data)
return bridge
async def async_unload_entry(hass, entry):
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

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

@ -17,11 +17,14 @@ from homeassistant.const import (
CONF_TIMEOUT,
CONF_TOKEN,
)
from homeassistant.helpers.device_registry import format_mac
from .const import (
CONF_DESCRIPTION,
LEGACY_PORT,
LOGGER,
METHOD_LEGACY,
METHOD_WEBSOCKET,
RESULT_AUTH_MISSING,
RESULT_CANNOT_CONNECT,
RESULT_NOT_SUPPORTED,
@ -34,13 +37,44 @@ from .const import (
)
def mac_from_device_info(info):
"""Extract the mac address from the device info."""
dev_info = info.get("device", {})
if dev_info.get("networkType") == "wireless" and dev_info.get("wifiMac"):
return format_mac(dev_info["wifiMac"])
return None
async def async_get_device_info(hass, bridge, host):
"""Fetch the port, method, and device info."""
return await hass.async_add_executor_job(_get_device_info, bridge, host)
def _get_device_info(bridge, host):
"""Fetch the port, method, and device info."""
if bridge and bridge.port:
return bridge.port, bridge.method, bridge.device_info()
for port in WEBSOCKET_PORTS:
bridge = SamsungTVBridge.get_bridge(METHOD_WEBSOCKET, host, port)
if info := bridge.device_info():
return port, METHOD_WEBSOCKET, info
bridge = SamsungTVBridge.get_bridge(METHOD_LEGACY, host, LEGACY_PORT)
result = bridge.try_connect()
if result in (RESULT_SUCCESS, RESULT_AUTH_MISSING):
return LEGACY_PORT, METHOD_LEGACY, None
return None, None, None
class SamsungTVBridge(ABC):
"""The Base Bridge abstract class."""
@staticmethod
def get_bridge(method, host, port=None, token=None):
"""Get Bridge instance."""
if method == METHOD_LEGACY:
if method == METHOD_LEGACY or port == LEGACY_PORT:
return SamsungTVLegacyBridge(method, host, port)
return SamsungTVWSBridge(method, host, port, token)
@ -50,7 +84,6 @@ class SamsungTVBridge(ABC):
self.method = method
self.host = host
self.token = None
self.default_port = None
self._remote = None
self._callback = None
@ -66,6 +99,10 @@ class SamsungTVBridge(ABC):
def device_info(self):
"""Try to gather infos of this TV."""
@abstractmethod
def mac_from_device(self):
"""Try to fetch the mac address of the TV."""
def is_on(self):
"""Tells if the TV is on."""
if self._remote:
@ -137,7 +174,7 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
def __init__(self, method, host, port):
"""Initialize Bridge."""
super().__init__(method, host, None)
super().__init__(method, host, LEGACY_PORT)
self.config = {
CONF_NAME: VALUE_CONF_NAME,
CONF_DESCRIPTION: VALUE_CONF_NAME,
@ -148,6 +185,10 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
CONF_TIMEOUT: 1,
}
def mac_from_device(self):
"""Try to fetch the mac address of the TV."""
return None
def try_connect(self):
"""Try to connect to the Legacy TV."""
config = {
@ -212,7 +253,11 @@ class SamsungTVWSBridge(SamsungTVBridge):
"""Initialize Bridge."""
super().__init__(method, host, port)
self.token = token
self.default_port = 8001
def mac_from_device(self):
"""Try to fetch the mac address of the TV."""
info = self.device_info()
return mac_from_device_info(info) if info else None
def try_connect(self):
"""Try to connect to the Websocket TV."""

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

@ -24,7 +24,7 @@ from homeassistant.core import callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.typing import DiscoveryInfoType
from .bridge import SamsungTVBridge
from .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info
from .const import (
ATTR_PROPERTIES,
CONF_MANUFACTURER,
@ -47,23 +47,6 @@ DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME):
SUPPORTED_METHODS = [METHOD_LEGACY, METHOD_WEBSOCKET]
def _get_device_info(host):
"""Fetch device info by any websocket method."""
for port in WEBSOCKET_PORTS:
bridge = SamsungTVBridge.get_bridge(METHOD_WEBSOCKET, host, port)
if info := bridge.device_info():
return info
return None
async def async_get_device_info(hass, bridge, host):
"""Fetch device info from bridge or websocket."""
if bridge:
return await hass.async_add_executor_job(bridge.device_info)
return await hass.async_add_executor_job(_get_device_info, host)
def _strip_uuid(udn):
return udn[5:] if udn.startswith("uuid:") else udn
@ -107,7 +90,8 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_set_device_unique_id(self, raise_on_progress=True):
"""Set device unique_id."""
await self._async_get_and_check_device_info()
if not await self._async_get_and_check_device_info():
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
await self._async_set_unique_id_from_udn(raise_on_progress)
async def _async_set_unique_id_from_udn(self, raise_on_progress=True):
@ -134,9 +118,11 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_get_and_check_device_info(self):
"""Try to get the device info."""
info = await async_get_device_info(self.hass, self._bridge, self._host)
_port, _method, info = await async_get_device_info(
self.hass, self._bridge, self._host
)
if not info:
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
return False
dev_info = info.get("device", {})
device_type = dev_info.get("type")
if device_type != "Samsung SmartTV":
@ -146,9 +132,10 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._name = name.replace("[TV] ", "") if name else device_type
self._title = f"{self._name} ({self._model})"
self._udn = _strip_uuid(dev_info.get("udn", info["id"]))
if dev_info.get("networkType") == "wireless" and dev_info.get("wifiMac"):
self._mac = format_mac(dev_info.get("wifiMac"))
if mac := mac_from_device_info(info):
self._mac = mac
self._device_info = info
return True
async def async_step_import(self, user_input=None):
"""Handle configuration by yaml file."""
@ -156,11 +143,11 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# since the TV may be off at startup
await self._async_set_name_host_from_input(user_input)
self._async_abort_entries_match({CONF_HOST: self._host})
if user_input.get(CONF_PORT) in WEBSOCKET_PORTS:
port = user_input.get(CONF_PORT)
if port in WEBSOCKET_PORTS:
user_input[CONF_METHOD] = METHOD_WEBSOCKET
else:
elif port == LEGACY_PORT:
user_input[CONF_METHOD] = METHOD_LEGACY
user_input[CONF_PORT] = LEGACY_PORT
user_input[CONF_MANUFACTURER] = DEFAULT_MANUFACTURER
return self.async_create_entry(
title=self._title,
@ -225,6 +212,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_ssdp(self, discovery_info: DiscoveryInfoType):
"""Handle a flow initialized by ssdp discovery."""
LOGGER.debug("Samsung device found via SSDP: %s", discovery_info)
model_name = discovery_info.get(ATTR_UPNP_MODEL_NAME)
self._udn = _strip_uuid(discovery_info[ATTR_UPNP_UDN])
self._host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
await self._async_set_unique_id_from_udn()
@ -234,9 +222,10 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"samsung"
):
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
self._name = self._title = self._model = discovery_info.get(
ATTR_UPNP_MODEL_NAME
)
if not await self._async_get_and_check_device_info():
# If we cannot get device info for an SSDP discovery
# its likely a legacy tv.
self._name = self._title = self._model = model_name
self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm()

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

@ -21,6 +21,7 @@ def remote_fixture():
remote = Mock()
remote.__enter__ = Mock()
remote.__exit__ = Mock()
remote.port.return_value = 55000
remote_class.return_value = remote
yield remote
@ -37,6 +38,7 @@ def remotews_fixture():
remotews = Mock()
remotews.__enter__ = Mock()
remotews.__exit__ = Mock()
remotews.port.return_value = 8002
remotews.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {

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

@ -14,6 +14,7 @@ from homeassistant.components.samsungtv.const import (
CONF_MODEL,
DEFAULT_MANUFACTURER,
DOMAIN,
LEGACY_PORT,
METHOD_LEGACY,
METHOD_WEBSOCKET,
RESULT_AUTH_MISSING,
@ -362,6 +363,29 @@ async def test_ssdp_legacy_not_supported(hass: HomeAssistant, remote: Mock):
assert result["reason"] == RESULT_NOT_SUPPORTED
async def test_ssdp_websocket_success_populates_mac_address(
hass: HomeAssistant, remotews: Mock
):
"""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
)
assert result["type"] == "form"
assert result["step_id"] == "confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input="whatever"
)
assert result["type"] == "create_entry"
assert result["title"] == "Living Room (82GXARRS)"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "Living Room"
assert result["data"][CONF_MAC] == "aa:bb:cc:dd:ee:ff"
assert result["data"][CONF_MANUFACTURER] == "Samsung fake_manufacturer"
assert result["data"][CONF_MODEL] == "82GXARRS"
assert result["result"].unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de"
async def test_ssdp_websocket_not_supported(hass: HomeAssistant, remote: Mock):
"""Test starting a flow from discovery for not supported device."""
with patch(
@ -491,7 +515,7 @@ async def test_ssdp_already_configured(hass: HomeAssistant, remote: Mock):
assert entry.unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de"
async def test_import_legacy(hass: HomeAssistant):
async def test_import_legacy(hass: HomeAssistant, remote: Mock):
"""Test importing from yaml with hostname."""
with patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
@ -505,14 +529,18 @@ async def test_import_legacy(hass: HomeAssistant):
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake"
assert result["data"][CONF_METHOD] == METHOD_LEGACY
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data[CONF_METHOD] == METHOD_LEGACY
assert entries[0].data[CONF_PORT] == LEGACY_PORT
async def test_import_legacy_without_name(hass: HomeAssistant):
async def test_import_legacy_without_name(hass: HomeAssistant, remote: Mock):
"""Test importing from yaml without a name."""
with patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
@ -526,11 +554,15 @@ async def test_import_legacy_without_name(hass: HomeAssistant):
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake_host"
assert result["data"][CONF_METHOD] == METHOD_LEGACY
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data[CONF_METHOD] == METHOD_LEGACY
assert entries[0].data[CONF_PORT] == LEGACY_PORT
async def test_import_websocket(hass: HomeAssistant):
"""Test importing from yaml with hostname."""
@ -547,12 +579,38 @@ async def test_import_websocket(hass: HomeAssistant):
assert result["type"] == "create_entry"
assert result["title"] == "fake"
assert result["data"][CONF_METHOD] == METHOD_WEBSOCKET
assert result["data"][CONF_PORT] == 8002
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
async def test_import_websocket_without_port(hass: HomeAssistant, remotews: Mock):
"""Test importing from yaml with hostname by no port."""
with patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_IMPORT_WSDATA,
)
await hass.async_block_till_done()
assert result["type"] == "create_entry"
assert result["title"] == "fake"
assert result["data"][CONF_HOST] == "fake_host"
assert result["data"][CONF_NAME] == "fake"
assert result["data"][CONF_MANUFACTURER] == "Samsung"
assert result["result"].unique_id is None
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data[CONF_METHOD] == METHOD_WEBSOCKET
assert entries[0].data[CONF_PORT] == 8002
async def test_import_unknown_host(hass: HomeAssistant, remotews: Mock):
"""Test importing from yaml with hostname that does not resolve."""
with patch(
@ -687,6 +745,7 @@ async def test_autodetect_websocket(hass: HomeAssistant, remote: Mock, remotews:
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"networkType": "wireless",
"wifiMac": "aa:bb:cc:dd:ee:ff",
"udn": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"mac": "aa:bb:cc:dd:ee:ff",
@ -707,6 +766,11 @@ async def test_autodetect_websocket(hass: HomeAssistant, remote: Mock, remotews:
call(**AUTODETECT_WEBSOCKET_SSL),
call(**DEVICEINFO_WEBSOCKET_SSL),
]
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
assert entries[0].data[CONF_MAC] == "aa:bb:cc:dd:ee:ff"
async def test_autodetect_auth_missing(hass: HomeAssistant, remote: Mock):
@ -747,14 +811,14 @@ async def test_autodetect_not_supported(hass: HomeAssistant, remote: Mock):
async def test_autodetect_legacy(hass: HomeAssistant, remote: Mock):
"""Test for send key with autodetection of protocol."""
with patch("homeassistant.components.samsungtv.bridge.Remote") as remote:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
assert result["data"][CONF_METHOD] == "legacy"
assert remote.call_count == 1
assert remote.call_args_list == [call(AUTODETECT_LEGACY)]
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA
)
assert result["type"] == "create_entry"
assert result["data"][CONF_METHOD] == "legacy"
assert result["data"][CONF_NAME] == "fake_name"
assert result["data"][CONF_MAC] is None
assert result["data"][CONF_PORT] == LEGACY_PORT
async def test_autodetect_none(hass: HomeAssistant, remote: Mock, remotews: Mock):

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

@ -8,10 +8,12 @@ from homeassistant.components.samsungtv.const import (
METHOD_WEBSOCKET,
)
from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
CONF_HOST,
CONF_MAC,
CONF_METHOD,
CONF_NAME,
SERVICE_VOLUME_UP,
@ -30,6 +32,16 @@ MOCK_CONFIG = {
}
]
}
MOCK_CONFIG_WITHOUT_PORT = {
SAMSUNGTV_DOMAIN: [
{
CONF_HOST: "fake_host",
CONF_NAME: "fake",
CONF_ON_ACTION: [{"delay": "00:00:01"}],
}
]
}
REMOTE_CALL = {
"name": "HomeAssistant",
"description": "HomeAssistant",
@ -67,6 +79,41 @@ async def test_setup(hass: HomeAssistant, remote: Mock):
assert remote.call_args == call(REMOTE_CALL)
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.SamsungTVWS.open",
side_effect=OSError,
), patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
config_entries_domain = hass.config_entries.async_entries(SAMSUNGTV_DOMAIN)
assert len(config_entries_domain) == 1
assert config_entries_domain[0].state == ConfigEntryState.SETUP_RETRY
async def test_setup_from_yaml_without_port_device_online(
hass: HomeAssistant, remotews: Mock
):
"""Test import from yaml when the device is online."""
with patch(
"homeassistant.components.samsungtv.config_flow.socket.gethostbyname",
return_value="fake_host",
):
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)
await hass.async_block_till_done()
config_entries_domain = hass.config_entries.async_entries(SAMSUNGTV_DOMAIN)
assert len(config_entries_domain) == 1
assert config_entries_domain[0].data[CONF_MAC] == "aa:bb:cc:dd:ee:ff"
async def test_setup_duplicate_config(hass: HomeAssistant, remote: Mock, caplog):
"""Test duplicate setup of platform."""
DUPLICATE = {

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

@ -159,14 +159,33 @@ async def test_setup_websocket(hass, remotews, mock_now):
remote = Mock()
remote.__enter__ = Mock(return_value=enter)
remote.__exit__ = Mock()
remote.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"wifiMac": "aa:bb:cc:dd:ee:ff",
"name": "[TV] Living Room",
"type": "Samsung SmartTV",
"networkType": "wireless",
},
}
remote_class.return_value = remote
await setup_samsungtv(hass, MOCK_CONFIGWS)
assert remote_class.call_count == 1
assert remote_class.call_args_list == [call(**MOCK_CALLS_WS)]
assert remote_class.call_count == 2
assert remote_class.call_args_list == [
call(**MOCK_CALLS_WS),
call(**MOCK_CALLS_WS),
]
assert hass.states.get(ENTITY_ID)
await hass.async_block_till_done()
config_entries = hass.config_entries.async_entries(SAMSUNGTV_DOMAIN)
assert len(config_entries) == 1
assert config_entries[0].data[CONF_MAC] == "aa:bb:cc:dd:ee:ff"
async def test_setup_websocket_2(hass, mock_now):
"""Test setup of platform from config entry."""
@ -183,20 +202,37 @@ async def test_setup_websocket_2(hass, mock_now):
assert len(config_entries) == 1
assert entry is config_entries[0]
assert await async_setup_component(hass, SAMSUNGTV_DOMAIN, {})
await hass.async_block_till_done()
next_update = mock_now + timedelta(minutes=5)
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remote, patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
with patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remote_class:
enter = Mock()
type(enter).token = PropertyMock(return_value="987654321")
remote = Mock()
remote.__enter__ = Mock(return_value=enter)
remote.__exit__ = Mock()
remote.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"wifiMac": "aa:bb:cc:dd:ee:ff",
"name": "[TV] Living Room",
"type": "Samsung SmartTV",
"networkType": "wireless",
},
}
remote_class.return_value = remote
assert await async_setup_component(hass, SAMSUNGTV_DOMAIN, {})
await hass.async_block_till_done()
assert config_entries[0].data[CONF_MAC] == "aa:bb:cc:dd:ee:ff"
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
assert remote.call_count == 1
assert remote.call_args_list == [call(**MOCK_CALLS_WS)]
assert remote_class.call_count == 3
assert remote_class.call_args_list[0] == call(**MOCK_CALLS_WS)
async def test_update_on(hass, remote, mock_now):