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

Expose Samsung wrapper as async (#67042) 

Co-authored-by: epenet <epenet@users.noreply.github.com>
pull/67068/head
epenet 2022-02-22 19:31:16 +01:00 committed by GitHub
parent d25a46d68d
commit a60c37cdb8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 147 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -100,10 +100,11 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@callback
def _async_get_device_bridge(
data: dict[str, Any]
hass: HomeAssistant, data: dict[str, Any]
) -> SamsungTVLegacyBridge | SamsungTVWSBridge:
"""Get device bridge."""
return SamsungTVBridge.get_bridge(
hass,
data[CONF_METHOD],
data[CONF_HOST],
data[CONF_PORT],
@ -131,9 +132,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
bridge.register_new_token_callback(new_token_callback)
def stop_bridge(event: Event) -> None:
async def stop_bridge(event: Event) -> None:
"""Stop SamsungTV bridge connection."""
bridge.stop()
await bridge.async_stop()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_bridge)
@ -169,14 +170,14 @@ async def _async_create_bridge_with_updated_data(
updated_data[CONF_PORT] = port
updated_data[CONF_METHOD] = method
bridge = _async_get_device_bridge({**entry.data, **updated_data})
bridge = _async_get_device_bridge(hass, {**entry.data, **updated_data})
mac = entry.data.get(CONF_MAC)
if not 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)
mac = await bridge.async_mac_from_device()
if not mac:
mac = await hass.async_add_executor_job(
@ -196,7 +197,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN][entry.entry_id].stop()
await hass.data[DOMAIN][entry.entry_id].async_stop()
return unload_ok

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

@ -54,25 +54,18 @@ async def async_get_device_info(
hass: HomeAssistant,
bridge: SamsungTVWSBridge | SamsungTVLegacyBridge | None,
host: str,
) -> tuple[int | None, str | None, dict[str, Any] | None]:
"""Fetch the port, method, and device info."""
return await hass.async_add_executor_job(_get_device_info, bridge, host)
def _get_device_info(
bridge: SamsungTVWSBridge | SamsungTVLegacyBridge, host: str
) -> tuple[int | None, str | None, dict[str, Any] | None]:
"""Fetch the port, method, and device info."""
if bridge and bridge.port:
return bridge.port, bridge.method, bridge.device_info()
return bridge.port, bridge.method, await bridge.async_device_info()
for port in WEBSOCKET_PORTS:
bridge = SamsungTVBridge.get_bridge(METHOD_WEBSOCKET, host, port)
if info := bridge.device_info():
bridge = SamsungTVBridge.get_bridge(hass, METHOD_WEBSOCKET, host, port)
if info := await bridge.async_device_info():
return port, METHOD_WEBSOCKET, info
bridge = SamsungTVBridge.get_bridge(METHOD_LEGACY, host, LEGACY_PORT)
result = bridge.try_connect()
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, None
@ -84,15 +77,22 @@ class SamsungTVBridge(ABC):
@staticmethod
def get_bridge(
method: str, host: str, port: int | None = None, token: str | None = None
hass: HomeAssistant,
method: str,
host: str,
port: int | None = None,
token: str | None = None,
) -> SamsungTVLegacyBridge | SamsungTVWSBridge:
"""Get Bridge instance."""
if method == METHOD_LEGACY or port == LEGACY_PORT:
return SamsungTVLegacyBridge(method, host, port)
return SamsungTVWSBridge(method, host, port, token)
return SamsungTVLegacyBridge(hass, method, host, port)
return SamsungTVWSBridge(hass, method, host, port, token)
def __init__(self, method: str, host: str, port: int | None = None) -> None:
def __init__(
self, hass: HomeAssistant, method: str, host: str, port: int | None = None
) -> None:
"""Initialize Bridge."""
self.hass = hass
self.port = port
self.method = method
self.host = host
@ -110,28 +110,29 @@ class SamsungTVBridge(ABC):
self._new_token_callback = func
@abstractmethod
def try_connect(self) -> str | None:
async def async_try_connect(self) -> str | None:
"""Try to connect to the TV."""
@abstractmethod
def device_info(self) -> dict[str, Any] | None:
async def async_device_info(self) -> dict[str, Any] | None:
"""Try to gather infos of this TV."""
@abstractmethod
def mac_from_device(self) -> str | None:
async def async_mac_from_device(self) -> str | None:
"""Try to fetch the mac address of the TV."""
@abstractmethod
def get_app_list(self) -> dict[str, str] | None:
async def async_get_app_list(self) -> dict[str, str] | None:
"""Get installed app list."""
def is_on(self) -> bool:
async def async_is_on(self) -> bool:
"""Tells if the TV is on."""
if self._remote is not None:
self.close_remote()
await self.async_close_remote()
try:
return self._get_remote() is not None
remote = await self.hass.async_add_executor_job(self._get_remote)
return remote is not None
except (
UnhandledResponse,
AccessDenied,
@ -143,14 +144,14 @@ class SamsungTVBridge(ABC):
# Different reasons, e.g. hostname not resolveable
return False
def send_key(self, key: str, key_type: str | None = None) -> None:
async def async_send_key(self, key: str, key_type: str | None = None) -> None:
"""Send a key to the tv and handles exceptions."""
try:
# recreate connection if connection was dead
retry_count = 1
for _ in range(retry_count + 1):
try:
self._send_key(key, key_type)
await self._async_send_key(key, key_type)
break
except (
ConnectionClosed,
@ -168,19 +169,19 @@ class SamsungTVBridge(ABC):
pass
@abstractmethod
def _send_key(self, key: str, key_type: str | None = None) -> None:
async def _async_send_key(self, key: str, key_type: str | None = None) -> None:
"""Send the key."""
@abstractmethod
def _get_remote(self, avoid_open: bool = False) -> Remote:
def _get_remote(self, avoid_open: bool = False) -> Remote | SamsungTVWS:
"""Get Remote object."""
def close_remote(self) -> None:
async def async_close_remote(self) -> None:
"""Close remote object."""
try:
if self._remote is not None:
# Close the current remote connection
self._remote.close()
await self.hass.async_add_executor_job(self._remote.close)
self._remote = None
except OSError:
LOGGER.debug("Could not establish connection")
@ -199,9 +200,11 @@ class SamsungTVBridge(ABC):
class SamsungTVLegacyBridge(SamsungTVBridge):
"""The Bridge for Legacy TVs."""
def __init__(self, method: str, host: str, port: int | None) -> None:
def __init__(
self, hass: HomeAssistant, method: str, host: str, port: int | None
) -> None:
"""Initialize Bridge."""
super().__init__(method, host, LEGACY_PORT)
super().__init__(hass, method, host, LEGACY_PORT)
self.config = {
CONF_NAME: VALUE_CONF_NAME,
CONF_DESCRIPTION: VALUE_CONF_NAME,
@ -212,15 +215,19 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
CONF_TIMEOUT: 1,
}
def mac_from_device(self) -> None:
async def async_mac_from_device(self) -> None:
"""Try to fetch the mac address of the TV."""
return None
def get_app_list(self) -> dict[str, str]:
async def async_get_app_list(self) -> dict[str, str]:
"""Get installed app list."""
return {}
def try_connect(self) -> str:
async def async_try_connect(self) -> str:
"""Try to connect to the Legacy TV."""
return await self.hass.async_add_executor_job(self._try_connect)
def _try_connect(self) -> str:
"""Try to connect to the Legacy TV."""
config = {
CONF_NAME: VALUE_CONF_NAME,
@ -247,7 +254,7 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
LOGGER.debug("Failing config: %s, error: %s", config, err)
return RESULT_CANNOT_CONNECT
def device_info(self) -> None:
async def async_device_info(self) -> None:
"""Try to gather infos of this device."""
return None
@ -269,34 +276,47 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
pass
return self._remote
def _send_key(self, key: str, key_type: str | None = None) -> None:
async def _async_send_key(self, key: str, key_type: str | None = None) -> None:
"""Send the key using legacy protocol."""
return await self.hass.async_add_executor_job(self._send_key, key)
def _send_key(self, key: str) -> None:
"""Send the key using legacy protocol."""
if remote := self._get_remote():
remote.control(key)
def stop(self) -> None:
async def async_stop(self) -> None:
"""Stop Bridge."""
LOGGER.debug("Stopping SamsungTVLegacyBridge")
self.close_remote()
await self.async_close_remote()
class SamsungTVWSBridge(SamsungTVBridge):
"""The Bridge for WebSocket TVs."""
def __init__(
self, method: str, host: str, port: int | None = None, token: str | None = None
self,
hass: HomeAssistant,
method: str,
host: str,
port: int | None = None,
token: str | None = None,
) -> None:
"""Initialize Bridge."""
super().__init__(method, host, port)
super().__init__(hass, method, host, port)
self.token = token
self._app_list: dict[str, str] | None = None
def mac_from_device(self) -> str | None:
async def async_mac_from_device(self) -> str | None:
"""Try to fetch the mac address of the TV."""
info = self.device_info()
info = await self.async_device_info()
return mac_from_device_info(info) if info else None
def get_app_list(self) -> dict[str, str] | None:
async def async_get_app_list(self) -> dict[str, str] | None:
"""Get installed app list."""
return await self.hass.async_add_executor_job(self._get_app_list)
def _get_app_list(self) -> dict[str, str] | None:
"""Get installed app list."""
if self._app_list is None:
if remote := self._get_remote():
@ -308,7 +328,11 @@ class SamsungTVWSBridge(SamsungTVBridge):
return self._app_list
def try_connect(self) -> str:
async def async_try_connect(self) -> str:
"""Try to connect to the Websocket TV."""
return await self.hass.async_add_executor_job(self._try_connect)
def _try_connect(self) -> str:
"""Try to connect to the Websocket TV."""
for self.port in WEBSOCKET_PORTS:
config = {
@ -350,15 +374,21 @@ class SamsungTVWSBridge(SamsungTVBridge):
return RESULT_CANNOT_CONNECT
def device_info(self) -> dict[str, Any] | None:
async def async_device_info(self) -> dict[str, Any] | None:
"""Try to gather infos of this TV."""
if remote := self._get_remote(avoid_open=True):
with contextlib.suppress(HttpApiError, RequestsTimeout):
device_info: dict[str, Any] = remote.rest_device_info()
device_info: dict[str, Any] = await self.hass.async_add_executor_job(
remote.rest_device_info
)
return device_info
return None
async def _async_send_key(self, key: str, key_type: str | None = None) -> None:
"""Send the key using websocket protocol."""
return await self.hass.async_add_executor_job(self._send_key, key, key_type)
def _send_key(self, key: str, key_type: str | None = None) -> None:
"""Send the key using websocket protocol."""
if key == "KEY_POWEROFF":
@ -369,7 +399,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
else:
remote.send_key(key)
def _get_remote(self, avoid_open: bool = False) -> Remote:
def _get_remote(self, avoid_open: bool = False) -> SamsungTVWS:
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
@ -388,11 +418,16 @@ class SamsungTVWSBridge(SamsungTVBridge):
self._remote.open("samsung.remote.control")
# This is only happening when the auth was switched to DENY
# A removed auth will lead to socket timeout because waiting for auth popup is just an open socket
except ConnectionFailure:
except ConnectionFailure as err:
LOGGER.debug("ConnectionFailure %s", err.__repr__())
self._notify_reauth_callback()
except (WebSocketException, OSError):
except (WebSocketException, OSError) as err:
LOGGER.debug("WebSocketException, OSError %s", err.__repr__())
self._remote = None
else:
LOGGER.debug(
"Created SamsungTVWSBridge for %s (%s)", CONF_NAME, self.host
)
if self.token != self._remote.token:
LOGGER.debug(
"SamsungTVWSBridge has provided a new token %s",
@ -402,7 +437,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
self._notify_new_token_callback()
return self._remote
def stop(self) -> None:
async def async_stop(self) -> None:
"""Stop Bridge."""
LOGGER.debug("Stopping SamsungTVWSBridge")
self.close_remote()
await self.async_close_remote()

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

@ -124,11 +124,11 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
updates[CONF_MAC] = self._mac
self._abort_if_unique_id_configured(updates=updates)
def _try_connect(self) -> None:
async def _try_connect(self) -> None:
"""Try to connect and check auth."""
for method in SUPPORTED_METHODS:
self._bridge = SamsungTVBridge.get_bridge(method, self._host)
result = self._bridge.try_connect()
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:
@ -203,7 +203,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""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.hass.async_add_executor_job(self._try_connect)
await self._try_connect()
assert self._bridge
self._async_abort_entries_match({CONF_HOST: self._host})
if self._bridge.method != METHOD_LEGACY:
@ -309,7 +309,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle user-confirmation of discovered node."""
if user_input is not None:
await self.hass.async_add_executor_job(self._try_connect)
await self._try_connect()
assert self._bridge
return self._get_entry_from_bridge()
@ -341,9 +341,11 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
assert self._reauth_entry
if user_input is not None:
bridge = SamsungTVBridge.get_bridge(
self._reauth_entry.data[CONF_METHOD], self._reauth_entry.data[CONF_HOST]
self.hass,
self._reauth_entry.data[CONF_METHOD],
self._reauth_entry.data[CONF_HOST],
)
result = await self.hass.async_add_executor_job(bridge.try_connect)
result = await bridge.async_try_connect()
if result == RESULT_SUCCESS:
new_data = dict(self._reauth_entry.data)
new_data[CONF_TOKEN] = bridge.token

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

@ -23,5 +23,5 @@ async def async_get_config_entry_diagnostics(
]
return {
"entry": async_redact_data(entry.as_dict(), TO_REDACT),
"device_info": await hass.async_add_executor_job(bridge.device_info),
"device_info": await bridge.async_device_info(),
}

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

@ -153,30 +153,32 @@ class SamsungTVDevice(MediaPlayerEntity):
)
)
def update(self) -> None:
async def async_update(self) -> None:
"""Update state of device."""
if self._auth_failed or self.hass.is_stopping:
return
if self._power_off_in_progress():
self._attr_state = STATE_OFF
else:
self._attr_state = STATE_ON if self._bridge.is_on() else STATE_OFF
self._attr_state = (
STATE_ON if await self._bridge.async_is_on() else STATE_OFF
)
if self._attr_state == STATE_ON and self._app_list is None:
self._app_list = {} # Ensure that we don't update it twice in parallel
self._update_app_list()
await self._async_update_app_list()
def _update_app_list(self) -> None:
self._app_list = self._bridge.get_app_list()
async def _async_update_app_list(self) -> None:
self._app_list = await self._bridge.async_get_app_list()
if self._app_list is not None:
self._attr_source_list.extend(self._app_list)
def send_key(self, key: str, key_type: str | None = None) -> None:
async def _async_send_key(self, key: str, key_type: str | None = None) -> None:
"""Send a key to the tv and handles exceptions."""
if self._power_off_in_progress() and key != "KEY_POWEROFF":
LOGGER.info("TV is powering off, not sending command: %s", key)
return
self._bridge.send_key(key, key_type)
await self._bridge.async_send_key(key, key_type)
def _power_off_in_progress(self) -> bool:
return (
@ -196,57 +198,57 @@ class SamsungTVDevice(MediaPlayerEntity):
or self._power_off_in_progress()
)
def turn_off(self) -> None:
async def async_turn_off(self) -> None:
"""Turn off media player."""
self._end_of_power_off = dt_util.utcnow() + SCAN_INTERVAL_PLUS_OFF_TIME
self.send_key("KEY_POWEROFF")
await self._async_send_key("KEY_POWEROFF")
# Force closing of remote session to provide instant UI feedback
self._bridge.close_remote()
await self._bridge.async_close_remote()
def volume_up(self) -> None:
async def async_volume_up(self) -> None:
"""Volume up the media player."""
self.send_key("KEY_VOLUP")
await self._async_send_key("KEY_VOLUP")
def volume_down(self) -> None:
async def async_volume_down(self) -> None:
"""Volume down media player."""
self.send_key("KEY_VOLDOWN")
await self._async_send_key("KEY_VOLDOWN")
def mute_volume(self, mute: bool) -> None:
async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command."""
self.send_key("KEY_MUTE")
await self._async_send_key("KEY_MUTE")
def media_play_pause(self) -> None:
async def async_media_play_pause(self) -> None:
"""Simulate play pause media player."""
if self._playing:
self.media_pause()
await self.async_media_pause()
else:
self.media_play()
await self.async_media_play()
def media_play(self) -> None:
async def async_media_play(self) -> None:
"""Send play command."""
self._playing = True
self.send_key("KEY_PLAY")
await self._async_send_key("KEY_PLAY")
def media_pause(self) -> None:
async def async_media_pause(self) -> None:
"""Send media pause command to media player."""
self._playing = False
self.send_key("KEY_PAUSE")
await self._async_send_key("KEY_PAUSE")
def media_next_track(self) -> None:
async def async_media_next_track(self) -> None:
"""Send next track command."""
self.send_key("KEY_CHUP")
await self._async_send_key("KEY_CHUP")
def media_previous_track(self) -> None:
async def async_media_previous_track(self) -> None:
"""Send the previous track command."""
self.send_key("KEY_CHDOWN")
await self._async_send_key("KEY_CHDOWN")
async def async_play_media(
self, media_type: str, media_id: str, **kwargs: Any
) -> None:
"""Support changing a channel."""
if media_type == MEDIA_TYPE_APP:
await self.hass.async_add_executor_job(self.send_key, media_id, "run_app")
await self._async_send_key(media_id, "run_app")
return
if media_type != MEDIA_TYPE_CHANNEL:
@ -261,9 +263,9 @@ class SamsungTVDevice(MediaPlayerEntity):
return
for digit in media_id:
await self.hass.async_add_executor_job(self.send_key, f"KEY_{digit}")
await self._async_send_key(f"KEY_{digit}")
await asyncio.sleep(KEY_PRESS_TIMEOUT)
await self.hass.async_add_executor_job(self.send_key, "KEY_ENTER")
await self._async_send_key("KEY_ENTER")
def _wake_on_lan(self) -> None:
"""Wake the device via wake on lan."""
@ -279,14 +281,14 @@ class SamsungTVDevice(MediaPlayerEntity):
elif self._mac:
await self.hass.async_add_executor_job(self._wake_on_lan)
def select_source(self, source: str) -> None:
async def async_select_source(self, source: str) -> None:
"""Select input source."""
if self._app_list and source in self._app_list:
self.send_key(self._app_list[source], "run_app")
await self._async_send_key(self._app_list[source], "run_app")
return
if source in SOURCES:
self.send_key(SOURCES[source])
await self._async_send_key(SOURCES[source])
return
LOGGER.error("Unsupported source")

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

@ -322,7 +322,7 @@ async def test_ssdp(hass: HomeAssistant, no_mac_address: Mock) -> None:
no_mac_address.return_value = "aa:bb:cc:dd:ee:ff"
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.device_info",
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=MOCK_DEVICE_INFO,
):
# confirm to add the entry
@ -351,7 +351,7 @@ async def test_ssdp_noprefix(hass: HomeAssistant, no_mac_address: Mock) -> None:
no_mac_address.return_value = "aa:bb:cc:dd:ee:ff"
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.device_info",
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=MOCK_DEVICE_INFO_2,
):
# confirm to add the entry
@ -399,7 +399,7 @@ async def test_ssdp_legacy_missing_auth(hass: HomeAssistant) -> None:
# missing authentication
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVLegacyBridge.try_connect",
"homeassistant.components.samsungtv.bridge.SamsungTVLegacyBridge.async_try_connect",
return_value=RESULT_AUTH_MISSING,
):
result = await hass.config_entries.flow.async_configure(
@ -421,7 +421,7 @@ async def test_ssdp_legacy_not_supported(hass: HomeAssistant) -> None:
assert result["step_id"] == "confirm"
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVLegacyBridge.try_connect",
"homeassistant.components.samsungtv.bridge.SamsungTVLegacyBridge.async_try_connect",
return_value=RESULT_NOT_SUPPORTED,
):
# device not supported
@ -498,7 +498,7 @@ async def test_ssdp_not_successful(hass: HomeAssistant) -> None:
"homeassistant.components.samsungtv.bridge.SamsungTVWS.open",
side_effect=OSError("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.device_info",
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=MOCK_DEVICE_INFO,
):
@ -527,7 +527,7 @@ async def test_ssdp_not_successful_2(hass: HomeAssistant) -> None:
"homeassistant.components.samsungtv.bridge.SamsungTVWS.open",
side_effect=ConnectionFailure("Boom"),
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.device_info",
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=MOCK_DEVICE_INFO,
):
@ -554,7 +554,7 @@ async def test_ssdp_already_in_progress(
no_mac_address.return_value = "aa:bb:cc:dd:ee:ff"
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.device_info",
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=MOCK_DEVICE_INFO,
):
@ -581,7 +581,7 @@ async def test_ssdp_already_configured(
no_mac_address.return_value = "aa:bb:cc:dd:ee:ff"
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.device_info",
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=MOCK_DEVICE_INFO,
):

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

@ -83,7 +83,7 @@ async def test_setup_from_yaml_without_port_device_offline(hass: HomeAssistant)
"homeassistant.components.samsungtv.bridge.SamsungTVWS.open",
side_effect=OSError,
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.device_info",
"homeassistant.components.samsungtv.bridge.SamsungTVWSBridge.async_device_info",
return_value=None,
):
await async_setup_component(hass, SAMSUNGTV_DOMAIN, MOCK_CONFIG)