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

Use async rest api in SamsungTV (#67369) 

Co-authored-by: epenet <epenet@users.noreply.github.com>
pull/67446/head
epenet 2022-02-28 20:53:42 +01:00 committed by GitHub
parent 508ed257d4
commit 1556868d56
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 121 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,13 +2,14 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from asyncio.exceptions import TimeoutError as AsyncioTimeoutError
import contextlib
from typing import Any
from requests.exceptions import Timeout as RequestsTimeout
from samsungctl import Remote
from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledResponse
from samsungtvws import SamsungTVWS
from samsungtvws.async_rest import SamsungTVAsyncRest
from samsungtvws.exceptions import ConnectionFailure, HttpApiError
from websocket import WebSocketException
@ -21,6 +22,7 @@ from homeassistant.const import (
CONF_TIMEOUT,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import format_mac
from .const import (
@ -294,6 +296,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
"""Initialize Bridge."""
super().__init__(hass, method, host, port)
self.token = token
self._rest_api: SamsungTVAsyncRest | None = None
self._app_list: dict[str, str] | None = None
self._remote: SamsungTVWS | None = None
@ -375,14 +378,21 @@ class SamsungTVWSBridge(SamsungTVBridge):
async def async_device_info(self) -> dict[str, Any] | None:
"""Try to gather infos of this TV."""
return await self.hass.async_add_executor_job(self._device_info)
if not self.port:
return None
def _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()
return device_info
if self._rest_api is None:
self._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 self._rest_api.rest_device_info()
LOGGER.debug("Device info on %s is: %s", self.host, device_info)
return device_info
return None
@ -416,7 +426,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
# Different reasons, e.g. hostname not resolveable
pass
def _get_remote(self, avoid_open: bool = False) -> SamsungTVWS:
def _get_remote(self) -> SamsungTVWS:
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
@ -431,8 +441,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
timeout=TIMEOUT_WEBSOCKET,
name=VALUE_CONF_NAME,
)
if not avoid_open:
self._remote.open()
self._remote.open()
# 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 as err:

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

@ -32,16 +32,14 @@ def remote_fixture() -> Mock:
yield remote
@pytest.fixture(name="remotews")
def remotews_fixture() -> Mock:
"""Patch the samsungtvws SamsungTVWS."""
@pytest.fixture(name="rest_api", autouse=True)
def rest_api_fixture() -> Mock:
"""Patch the samsungtvws SamsungTVAsyncRest."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews_class:
remotews = Mock(SamsungTVWS)
remotews.__enter__ = Mock(return_value=remotews)
remotews.__exit__ = Mock()
remotews.rest_device_info.return_value = {
"homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest",
autospec=True,
) as rest_api_class:
rest_api_class.return_value.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
@ -51,51 +49,24 @@ def remotews_fixture() -> Mock:
"networkType": "wireless",
},
}
yield rest_api_class.return_value
@pytest.fixture(name="remotews")
def remotews_fixture() -> Mock:
"""Patch the samsungtvws SamsungTVWS."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews_class:
remotews = Mock(SamsungTVWS)
remotews.__enter__ = Mock(return_value=remotews)
remotews.__exit__ = Mock()
remotews.app_list.return_value = SAMPLE_APP_LIST
remotews.token = "FAKE_TOKEN"
remotews_class.return_value = remotews
yield remotews
@pytest.fixture(name="remotews_no_device_info")
def remotews_no_device_info_fixture() -> Mock:
"""Patch the samsungtvws SamsungTVWS."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews_class:
remotews = Mock(SamsungTVWS)
remotews.__enter__ = Mock(return_value=remotews)
remotews.__exit__ = Mock()
remotews.rest_device_info.return_value = None
remotews.token = "FAKE_TOKEN"
remotews_class.return_value = remotews
yield remotews
@pytest.fixture(name="remotews_soundbar")
def remotews_soundbar_fixture() -> Mock:
"""Patch the samsungtvws SamsungTVWS."""
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews_class:
remotews = Mock(SamsungTVWS)
remotews.__enter__ = Mock(return_value=remotews)
remotews.__exit__ = Mock()
remotews.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"wifiMac": "aa:bb:cc:dd:ee:ff",
"mac": "aa:bb:cc:dd:ee:ff",
"name": "[TV] Living Room",
"type": "Samsung SoundBar",
},
}
remotews.token = "FAKE_TOKEN"
remotews_class.return_value = remotews
yield remotews
@pytest.fixture(name="delay")
def delay_fixture() -> Mock:
"""Patch the delay script function."""

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

@ -1,6 +1,6 @@
"""Tests for Samsung TV config flow."""
import socket
from unittest.mock import Mock, call, patch
from unittest.mock import ANY, AsyncMock, Mock, call, patch
import pytest
from samsungctl.exceptions import AccessDenied, UnhandledResponse
@ -178,10 +178,9 @@ AUTODETECT_WEBSOCKET_SSL = {
}
DEVICEINFO_WEBSOCKET_SSL = {
"host": "fake_host",
"name": "HomeAssistant",
"session": ANY,
"port": 8002,
"timeout": TIMEOUT_WEBSOCKET,
"token": "123456789",
}
@ -456,8 +455,11 @@ async def test_ssdp_websocket_success_populates_mac_address(
assert result["result"].unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de"
async def test_ssdp_websocket_not_supported(hass: HomeAssistant) -> None:
async def test_ssdp_websocket_not_supported(
hass: HomeAssistant, rest_api: Mock
) -> None:
"""Test starting a flow from discovery for not supported device."""
rest_api.rest_device_info.return_value = None
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=OSError("Boom"),
@ -630,9 +632,10 @@ async def test_import_legacy(hass: HomeAssistant, no_mac_address: Mock) -> None:
assert entries[0].data[CONF_PORT] == LEGACY_PORT
@pytest.mark.usefixtures("remote", "remotews_no_device_info", "no_mac_address")
async def test_import_legacy_without_name(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("remote", "remotews", "no_mac_address")
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},
@ -762,9 +765,19 @@ async def test_zeroconf(hass: HomeAssistant) -> None:
assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
@pytest.mark.usefixtures("remotews_soundbar")
async def test_zeroconf_ignores_soundbar(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("remotews")
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 = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"wifiMac": "aa:bb:cc:dd:ee:ff",
"mac": "aa:bb:cc:dd:ee:ff",
"name": "[TV] Living Room",
"type": "Samsung SoundBar",
},
}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
@ -775,9 +788,10 @@ async def test_zeroconf_ignores_soundbar(hass: HomeAssistant) -> None:
assert result["reason"] == "not_supported"
@pytest.mark.usefixtures("remote", "remotews_no_device_info")
async def test_zeroconf_no_device_info(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("remote", "remotews")
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
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_ZEROCONF},
@ -815,23 +829,29 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=OSError("Boom"),
), patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remotews:
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews, patch(
"homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest",
) as rest_api_class:
remote = Mock(SamsungTVWS)
remote.__enter__ = Mock(return_value=remote)
remote.__exit__ = Mock(return_value=False)
remote.app_list.return_value = SAMPLE_APP_LIST
remote.rest_device_info.return_value = {
"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",
"name": "[TV] Living Room",
"type": "Samsung SmartTV",
},
}
rest_api_class.return_value.rest_device_info = AsyncMock(
return_value={
"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",
"name": "[TV] Living Room",
"type": "Samsung SmartTV",
},
}
)
remote.token = "123456789"
remotews.return_value = remote
@ -841,11 +861,8 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None:
assert result["type"] == "create_entry"
assert result["data"][CONF_METHOD] == "websocket"
assert result["data"][CONF_TOKEN] == "123456789"
assert remotews.call_count == 2
assert remotews.call_args_list == [
call(**AUTODETECT_WEBSOCKET_SSL),
call(**DEVICEINFO_WEBSOCKET_SSL),
]
remotews.assert_called_once_with(**AUTODETECT_WEBSOCKET_SSL)
rest_api_class.assert_called_once_with(**DEVICEINFO_WEBSOCKET_SSL)
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(DOMAIN)
@ -861,22 +878,27 @@ async def test_websocket_no_mac(hass: HomeAssistant) -> None:
), patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS"
) as remotews, patch(
"homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest",
) as rest_api_class, patch(
"getmac.get_mac_address", return_value="gg:hh:ii:ll:mm:nn"
):
remote = Mock(SamsungTVWS)
remote.__enter__ = Mock(return_value=remote)
remote.__exit__ = Mock(return_value=False)
remote.app_list.return_value = SAMPLE_APP_LIST
remote.rest_device_info.return_value = {
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"networkType": "lan",
"udn": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"name": "[TV] Living Room",
"type": "Samsung SmartTV",
},
}
rest_api_class.return_value.rest_device_info = AsyncMock(
return_value={
"id": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"device": {
"modelName": "82GXARRS",
"networkType": "lan",
"udn": "uuid:be9554b9-c9fb-41f4-8920-22da015376a4",
"name": "[TV] Living Room",
"type": "Samsung SmartTV",
},
}
)
remote.token = "123456789"
remotews.return_value = remote
@ -887,11 +909,8 @@ async def test_websocket_no_mac(hass: HomeAssistant) -> None:
assert result["data"][CONF_METHOD] == "websocket"
assert result["data"][CONF_TOKEN] == "123456789"
assert result["data"][CONF_MAC] == "gg:hh:ii:ll:mm:nn"
assert remotews.call_count == 2
assert remotews.call_args_list == [
call(**AUTODETECT_WEBSOCKET_SSL),
call(**DEVICEINFO_WEBSOCKET_SSL),
]
remotews.assert_called_once_with(**AUTODETECT_WEBSOCKET_SSL)
rest_api_class.assert_called_once_with(**DEVICEINFO_WEBSOCKET_SSL)
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(DOMAIN)
@ -1166,18 +1185,16 @@ async def test_update_legacy_missing_mac_from_dhcp(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("remote")
async def test_update_legacy_missing_mac_from_dhcp_no_unique_id(
hass: HomeAssistant,
hass: HomeAssistant, rest_api: Mock
) -> None:
"""Test missing mac added when there is no unique id."""
rest_api.rest_device_info.side_effect = HttpApiError
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_LEGACY_ENTRY,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.samsungtv.bridge.SamsungTVWS.rest_device_info",
side_effect=HttpApiError,
), patch(
"homeassistant.components.samsungtv.bridge.Remote.__enter__",
return_value=True,
), patch(

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

@ -159,31 +159,20 @@ async def test_setup_without_turnon(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("remotews")
async def test_setup_websocket(hass: HomeAssistant) -> None:
"""Test setup of platform."""
with patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remote_class:
remote = Mock(SamsungTVWS)
remote.__enter__ = Mock(return_value=remote)
remote.__exit__ = Mock()
remote.app_list.return_value = SAMPLE_APP_LIST
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.token = "123456789"
remote_class.return_value = remote
await setup_samsungtv(hass, MOCK_CONFIGWS)
assert remote_class.call_count == 2
assert remote_class.call_args_list == [
call(**MOCK_CALLS_WS),
call(**MOCK_CALLS_WS),
]
assert remote_class.call_count == 1
assert remote_class.call_args_list == [call(**MOCK_CALLS_WS)]
assert hass.states.get(ENTITY_ID)
await hass.async_block_till_done()
@ -193,7 +182,9 @@ async def test_setup_websocket(hass: HomeAssistant) -> None:
assert config_entries[0].data[CONF_MAC] == "aa:bb:cc:dd:ee:ff"
async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> None:
async def test_setup_websocket_2(
hass: HomeAssistant, mock_now: datetime, rest_api: Mock
) -> None:
"""Test setup of platform from config entry."""
entity_id = f"{DOMAIN}.fake"
@ -208,21 +199,21 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non
assert len(config_entries) == 1
assert entry is config_entries[0]
rest_api.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",
},
}
with patch("homeassistant.components.samsungtv.bridge.SamsungTVWS") as remote_class:
remote = Mock(SamsungTVWS)
remote.__enter__ = Mock(return_value=remote)
remote.__exit__ = Mock()
remote.app_list.return_value = SAMPLE_APP_LIST
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.token = "987654321"
remote_class.return_value = remote
assert await async_setup_component(hass, SAMSUNGTV_DOMAIN, {})
@ -237,7 +228,7 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non
state = hass.states.get(entity_id)
assert state
assert remote_class.call_count == 3
assert remote_class.call_count == 2
assert remote_class.call_args_list[0] == call(**MOCK_CALLS_WS)