mirror of
https://github.com/home-assistant/core.git
synced 2025-07-28 07:37:34 +00:00
Rename samsung websocket test fixtures and constants (#144719)
This commit is contained in:
parent
646c230940
commit
e493fe1105
@ -219,45 +219,47 @@ def remoteencws_failing_fixture() -> Generator[None]:
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(name="remotews")
|
||||
def remotews_fixture() -> Generator[Mock]:
|
||||
@pytest.fixture(name="remote_websocket")
|
||||
def remote_websocket_fixture() -> Generator[Mock]:
|
||||
"""Patch the samsungtvws SamsungTVWS."""
|
||||
remotews = Mock(SamsungTVWSAsyncRemote)
|
||||
remotews.__aenter__ = AsyncMock(return_value=remotews)
|
||||
remotews.__aexit__ = AsyncMock()
|
||||
remotews.token = "FAKE_TOKEN"
|
||||
remotews.app_list_data = None
|
||||
remote_websocket = Mock(SamsungTVWSAsyncRemote)
|
||||
remote_websocket.__aenter__ = AsyncMock(return_value=remote_websocket)
|
||||
remote_websocket.__aexit__ = AsyncMock()
|
||||
remote_websocket.token = "FAKE_TOKEN"
|
||||
remote_websocket.app_list_data = None
|
||||
|
||||
async def _start_listening(
|
||||
ws_event_callback: Callable[[str, Any], Awaitable[None] | None] | None = None,
|
||||
):
|
||||
remotews.ws_event_callback = ws_event_callback
|
||||
remote_websocket.ws_event_callback = ws_event_callback
|
||||
|
||||
async def _send_commands(commands: list[SamsungTVCommand]):
|
||||
if (
|
||||
len(commands) == 1
|
||||
and isinstance(commands[0], ChannelEmitCommand)
|
||||
and commands[0].params["event"] == "ed.installedApp.get"
|
||||
and remotews.app_list_data is not None
|
||||
and remote_websocket.app_list_data is not None
|
||||
):
|
||||
remotews.raise_mock_ws_event_callback(
|
||||
remote_websocket.raise_mock_ws_event_callback(
|
||||
ED_INSTALLED_APP_EVENT,
|
||||
remotews.app_list_data,
|
||||
remote_websocket.app_list_data,
|
||||
)
|
||||
|
||||
def _mock_ws_event_callback(event: str, response: Any):
|
||||
if remotews.ws_event_callback:
|
||||
remotews.ws_event_callback(event, response)
|
||||
if remote_websocket.ws_event_callback:
|
||||
remote_websocket.ws_event_callback(event, response)
|
||||
|
||||
remotews.start_listening.side_effect = _start_listening
|
||||
remotews.send_commands.side_effect = _send_commands
|
||||
remotews.raise_mock_ws_event_callback = Mock(side_effect=_mock_ws_event_callback)
|
||||
remote_websocket.start_listening.side_effect = _start_listening
|
||||
remote_websocket.send_commands.side_effect = _send_commands
|
||||
remote_websocket.raise_mock_ws_event_callback = Mock(
|
||||
side_effect=_mock_ws_event_callback
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote",
|
||||
) as remotews_class:
|
||||
remotews_class.return_value = remotews
|
||||
yield remotews
|
||||
return_value=remote_websocket,
|
||||
):
|
||||
yield remote_websocket
|
||||
|
||||
|
||||
@pytest.fixture(name="remoteencws")
|
||||
|
@ -6,6 +6,7 @@ from homeassistant.components.samsungtv.const import (
|
||||
LEGACY_PORT,
|
||||
METHOD_LEGACY,
|
||||
METHOD_WEBSOCKET,
|
||||
WEBSOCKET_SSL_PORT,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
@ -37,12 +38,11 @@ MOCK_ENTRYDATA_ENCRYPTED_WS = {
|
||||
CONF_TOKEN: "037739871315caef138547b03e348b72",
|
||||
CONF_SESSION_ID: "2",
|
||||
}
|
||||
MOCK_ENTRYDATA_WS = {
|
||||
ENTRYDATA_WEBSOCKET = {
|
||||
CONF_HOST: "10.10.12.34",
|
||||
CONF_METHOD: METHOD_WEBSOCKET,
|
||||
CONF_PORT: 8002,
|
||||
CONF_MODEL: "any",
|
||||
CONF_NAME: "any",
|
||||
CONF_PORT: WEBSOCKET_SSL_PORT,
|
||||
CONF_MODEL: "UE43LS003",
|
||||
}
|
||||
MOCK_ENTRY_WS_WITH_MAC = {
|
||||
CONF_HOST: "fake_host",
|
||||
|
@ -56,8 +56,8 @@ from homeassistant.setup import async_setup_component
|
||||
|
||||
from .const import (
|
||||
ENTRYDATA_LEGACY,
|
||||
ENTRYDATA_WEBSOCKET,
|
||||
MOCK_ENTRYDATA_ENCRYPTED_WS,
|
||||
MOCK_ENTRYDATA_WS,
|
||||
MOCK_SSDP_DATA,
|
||||
MOCK_SSDP_DATA_MAIN_TV_AGENT_ST,
|
||||
MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
|
||||
@ -193,7 +193,7 @@ async def test_user_legacy_does_not_ok_first_time(hass: HomeAssistant) -> None:
|
||||
assert result3["result"].unique_id is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_user_websocket(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow by user."""
|
||||
with patch(
|
||||
@ -518,7 +518,7 @@ async def test_ssdp_noprefix(hass: HomeAssistant) -> None:
|
||||
assert result["result"].unique_id == "068e7781-006e-1000-bbbf-84a4668d8423"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api_failing")
|
||||
async def test_ssdp_legacy_missing_auth(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow from discovery with authentication."""
|
||||
with patch(
|
||||
@ -553,7 +553,7 @@ async def test_ssdp_legacy_missing_auth(hass: HomeAssistant) -> None:
|
||||
assert result["result"].unique_id == "068e7781-006e-1000-bbbf-84a4668d8423"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api_failing")
|
||||
async def test_ssdp_legacy_not_supported(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow from discovery for not supported device."""
|
||||
with patch(
|
||||
@ -568,7 +568,7 @@ async def test_ssdp_legacy_not_supported(hass: HomeAssistant) -> None:
|
||||
assert result["reason"] == RESULT_NOT_SUPPORTED
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_ssdp_websocket_success_populates_mac_address_and_ssdp_location(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
@ -597,7 +597,7 @@ async def test_ssdp_websocket_success_populates_mac_address_and_ssdp_location(
|
||||
assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_ssdp_websocket_success_populates_mac_address_and_main_tv_ssdp_location(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
@ -711,8 +711,8 @@ async def test_ssdp_websocket_cannot_connect(hass: HomeAssistant) -> None:
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote",
|
||||
) as remotews,
|
||||
patch.object(remotews, "open", side_effect=WebSocketException("Boom")),
|
||||
) as remote_websocket,
|
||||
patch.object(remote_websocket, "open", side_effect=WebSocketException("Boom")),
|
||||
):
|
||||
# device not supported
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
@ -823,7 +823,7 @@ async def test_ssdp_already_in_progress(hass: HomeAssistant) -> None:
|
||||
assert result["reason"] == RESULT_ALREADY_IN_PROGRESS
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "remoteencws_failing")
|
||||
async def test_ssdp_already_configured(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow from discovery when already configured."""
|
||||
with patch(
|
||||
@ -851,7 +851,9 @@ async def test_ssdp_already_configured(hass: HomeAssistant) -> None:
|
||||
assert entry.unique_id == "123"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api_non_ssl_only", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures(
|
||||
"remote_websocket", "rest_api_non_ssl_only", "remoteencws_failing"
|
||||
)
|
||||
async def test_dhcp_wireless(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow from dhcp."""
|
||||
# confirm to add the entry
|
||||
@ -877,7 +879,7 @@ async def test_dhcp_wireless(hass: HomeAssistant) -> None:
|
||||
assert result["result"].unique_id == "223da676-497a-4e06-9507-5e27ec4f0fb3"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_dhcp_wired(hass: HomeAssistant, rest_api: Mock) -> None:
|
||||
"""Test starting a flow from dhcp."""
|
||||
# Even though it is named "wifiMac", it matches the mac of the wired connection
|
||||
@ -907,7 +909,9 @@ async def test_dhcp_wired(hass: HomeAssistant, rest_api: Mock) -> None:
|
||||
assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api_non_ssl_only", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures(
|
||||
"remote_websocket", "rest_api_non_ssl_only", "remoteencws_failing"
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("source1", "data1", "source2", "data2", "is_matching_result"),
|
||||
[
|
||||
@ -979,7 +983,7 @@ async def test_dhcp_zeroconf_already_in_progress(
|
||||
assert return_values == [is_matching_result]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_zeroconf(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow from zeroconf."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
@ -1004,7 +1008,7 @@ async def test_zeroconf(hass: HomeAssistant) -> None:
|
||||
assert result["result"].unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "remoteencws_failing")
|
||||
async def test_zeroconf_ignores_soundbar(hass: HomeAssistant, rest_api: Mock) -> None:
|
||||
"""Test starting a flow from zeroconf where the device is actually a soundbar."""
|
||||
rest_api.rest_device_info.return_value = {
|
||||
@ -1027,7 +1031,9 @@ async def test_zeroconf_ignores_soundbar(hass: HomeAssistant, rest_api: Mock) ->
|
||||
assert result["reason"] == RESULT_NOT_SUPPORTED
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remote_legacy", "remotews", "remoteencws", "rest_api_failing")
|
||||
@pytest.mark.usefixtures(
|
||||
"remote_legacy", "remote_websocket", "remoteencws", "rest_api_failing"
|
||||
)
|
||||
async def test_zeroconf_no_device_info(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow from zeroconf where device_info returns None."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
@ -1040,7 +1046,7 @@ async def test_zeroconf_no_device_info(hass: HomeAssistant) -> None:
|
||||
assert result["reason"] == RESULT_NOT_SUPPORTED
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_zeroconf_and_dhcp_same_time(hass: HomeAssistant) -> None:
|
||||
"""Test starting a flow from zeroconf and dhcp."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
@ -1072,7 +1078,7 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None:
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote"
|
||||
) as remotews,
|
||||
) as remote_websocket,
|
||||
patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest",
|
||||
) as rest_api_class,
|
||||
@ -1095,7 +1101,7 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None:
|
||||
}
|
||||
)
|
||||
remote.token = "123456789"
|
||||
remotews.return_value = remote
|
||||
remote_websocket.return_value = remote
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA
|
||||
@ -1103,7 +1109,7 @@ async def test_autodetect_websocket(hass: HomeAssistant) -> None:
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_METHOD] == "websocket"
|
||||
assert result["data"][CONF_TOKEN] == "123456789"
|
||||
remotews.assert_called_once_with(**AUTODETECT_WEBSOCKET_SSL)
|
||||
remote_websocket.assert_called_once_with(**AUTODETECT_WEBSOCKET_SSL)
|
||||
rest_api_class.assert_called_once_with(**DEVICEINFO_WEBSOCKET_SSL)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
@ -1123,7 +1129,7 @@ async def test_websocket_no_mac(hass: HomeAssistant, mac_address: Mock) -> None:
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVWSAsyncRemote"
|
||||
) as remotews,
|
||||
) as remote_websocket,
|
||||
patch(
|
||||
"homeassistant.components.samsungtv.bridge.SamsungTVAsyncRest",
|
||||
) as rest_api_class,
|
||||
@ -1145,7 +1151,7 @@ async def test_websocket_no_mac(hass: HomeAssistant, mac_address: Mock) -> None:
|
||||
)
|
||||
|
||||
remote.token = "123456789"
|
||||
remotews.return_value = remote
|
||||
remote_websocket.return_value = remote
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=MOCK_USER_DATA
|
||||
@ -1154,7 +1160,7 @@ async def test_websocket_no_mac(hass: HomeAssistant, mac_address: Mock) -> None:
|
||||
assert result["data"][CONF_METHOD] == "websocket"
|
||||
assert result["data"][CONF_TOKEN] == "123456789"
|
||||
assert result["data"][CONF_MAC] == "gg:ee:tt:mm:aa:cc"
|
||||
remotews.assert_called_once_with(**AUTODETECT_WEBSOCKET_SSL)
|
||||
remote_websocket.assert_called_once_with(**AUTODETECT_WEBSOCKET_SSL)
|
||||
rest_api_class.assert_called_once_with(**DEVICEINFO_WEBSOCKET_SSL)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
@ -1244,7 +1250,7 @@ async def test_autodetect_none(hass: HomeAssistant) -> None:
|
||||
assert rest_device_info.call_count == 2
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_old_entry(hass: HomeAssistant) -> None:
|
||||
"""Test update of old entry sets unique id."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=ENTRYDATA_LEGACY)
|
||||
@ -1273,7 +1279,7 @@ async def test_update_old_entry(hass: HomeAssistant) -> None:
|
||||
assert entry2.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_missing_mac_unique_id_added_from_dhcp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1295,7 +1301,7 @@ async def test_update_missing_mac_unique_id_added_from_dhcp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_incorrectly_formatted_mac_unique_id_added_from_dhcp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1319,7 +1325,7 @@ async def test_update_incorrectly_formatted_mac_unique_id_added_from_dhcp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_missing_mac_unique_id_added_from_zeroconf(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1370,7 +1376,7 @@ async def test_update_missing_model_added_from_ssdp(
|
||||
assert entry.data[CONF_MODEL] == "UE55H6400"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_missing_mac_unique_id_ssdp_location_added_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1395,7 +1401,7 @@ async def test_update_missing_mac_unique_id_ssdp_location_added_from_ssdp(
|
||||
|
||||
|
||||
@pytest.mark.usefixtures(
|
||||
"remote_legacy", "remotews", "remoteencws_failing", "rest_api_failing"
|
||||
"remote_legacy", "remote_websocket", "remoteencws_failing", "rest_api_failing"
|
||||
)
|
||||
async def test_update_zeroconf_discovery_preserved_unique_id(
|
||||
hass: HomeAssistant,
|
||||
@ -1419,7 +1425,7 @@ async def test_update_zeroconf_discovery_preserved_unique_id(
|
||||
assert entry.unique_id == "original"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_missing_mac_unique_id_added_ssdp_location_updated_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1452,7 +1458,7 @@ async def test_update_missing_mac_unique_id_added_ssdp_location_updated_from_ssd
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_missing_mac_unique_id_added_ssdp_location_rendering_st_updated_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1486,7 +1492,7 @@ async def test_update_missing_mac_unique_id_added_ssdp_location_rendering_st_upd
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_missing_mac_unique_id_added_ssdp_location_main_tv_agent_st_updated_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1524,7 +1530,7 @@ async def test_update_missing_mac_unique_id_added_ssdp_location_main_tv_agent_st
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_ssdp_location_rendering_st_updated_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1555,7 +1561,7 @@ async def test_update_ssdp_location_rendering_st_updated_from_ssdp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_main_tv_ssdp_location_rendering_st_updated_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1585,7 +1591,7 @@ async def test_update_main_tv_ssdp_location_rendering_st_updated_from_ssdp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_update_missing_mac_added_unique_id_preserved_from_zeroconf(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1676,7 +1682,7 @@ async def test_update_legacy_missing_mac_from_dhcp_no_unique_id(
|
||||
assert entry.unique_id is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_ssdp_location_unique_id_added_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1704,7 +1710,7 @@ async def test_update_ssdp_location_unique_id_added_from_ssdp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_ssdp_location_unique_id_added_from_ssdp_with_rendering_control_st(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1753,10 +1759,10 @@ async def test_form_reauth_legacy(hass: HomeAssistant) -> None:
|
||||
assert result2["reason"] == "reauth_successful"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_form_reauth_websocket(hass: HomeAssistant) -> None:
|
||||
"""Test reauthenticate websocket."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRYDATA_WS)
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=ENTRYDATA_WEBSOCKET)
|
||||
entry.add_to_hass(hass)
|
||||
assert entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
@ -1776,16 +1782,16 @@ async def test_form_reauth_websocket(hass: HomeAssistant) -> None:
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_form_reauth_websocket_cannot_connect(
|
||||
hass: HomeAssistant, remotews: Mock
|
||||
hass: HomeAssistant, remote_websocket: Mock
|
||||
) -> None:
|
||||
"""Test reauthenticate websocket when we cannot connect on the first attempt."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRYDATA_WS)
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=ENTRYDATA_WEBSOCKET)
|
||||
entry.add_to_hass(hass)
|
||||
result = await entry.start_reauth_flow(hass)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
with patch.object(remotews, "open", side_effect=ConnectionFailure):
|
||||
with patch.object(remote_websocket, "open", side_effect=ConnectionFailure):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{},
|
||||
@ -1807,7 +1813,7 @@ async def test_form_reauth_websocket_cannot_connect(
|
||||
|
||||
async def test_form_reauth_websocket_not_supported(hass: HomeAssistant) -> None:
|
||||
"""Test reauthenticate websocket when the device is not supported."""
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRYDATA_WS)
|
||||
entry = MockConfigEntry(domain=DOMAIN, data=ENTRYDATA_WEBSOCKET)
|
||||
entry.add_to_hass(hass)
|
||||
result = await entry.start_reauth_flow(hass)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
@ -1897,7 +1903,7 @@ async def test_form_reauth_encrypted(hass: HomeAssistant) -> None:
|
||||
assert entry.data[CONF_SESSION_ID] == "1"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_incorrect_udn_matching_upnp_udn_unique_id_added_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1923,7 +1929,7 @@ async def test_update_incorrect_udn_matching_upnp_udn_unique_id_added_from_ssdp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_incorrect_udn_matching_mac_unique_id_added_from_ssdp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
@ -1949,14 +1955,14 @@ async def test_update_incorrect_udn_matching_mac_unique_id_added_from_ssdp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_update_incorrect_udn_matching_mac_from_dhcp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
"""Test that DHCP updates the wrong udn from ssdp via mac match."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={**MOCK_ENTRYDATA_WS, CONF_MAC: "aa:bb:aa:aa:aa:aa"},
|
||||
data={**ENTRYDATA_WEBSOCKET, CONF_MAC: "aa:bb:aa:aa:aa:aa"},
|
||||
source=config_entries.SOURCE_SSDP,
|
||||
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
|
||||
)
|
||||
@ -1976,14 +1982,14 @@ async def test_update_incorrect_udn_matching_mac_from_dhcp(
|
||||
assert entry.unique_id == "be9554b9-c9fb-41f4-8920-22da015376a4"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "remoteencws_failing")
|
||||
async def test_no_update_incorrect_udn_not_matching_mac_from_dhcp(
|
||||
hass: HomeAssistant, mock_setup_entry: AsyncMock
|
||||
) -> None:
|
||||
"""Test that DHCP does not update the wrong udn from ssdp via host match."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={**MOCK_ENTRYDATA_WS, CONF_MAC: "aa:bb:ss:ss:dd:pp"},
|
||||
data={**ENTRYDATA_WEBSOCKET, CONF_MAC: "aa:bb:ss:ss:dd:pp"},
|
||||
source=config_entries.SOURCE_SSDP,
|
||||
unique_id="0d1cef00-00dc-1000-9c80-4844f7b172de",
|
||||
)
|
||||
@ -2003,7 +2009,7 @@ async def test_no_update_incorrect_udn_not_matching_mac_from_dhcp(
|
||||
assert entry.unique_id == "0d1cef00-00dc-1000-9c80-4844f7b172de"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "remoteencws_failing")
|
||||
async def test_ssdp_update_mac(hass: HomeAssistant) -> None:
|
||||
"""Ensure that MAC address is correctly updated from SSDP."""
|
||||
with patch(
|
||||
|
@ -18,7 +18,7 @@ from tests.components.diagnostics import get_diagnostics_for_config_entry
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_entry_diagnostics(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
|
@ -40,9 +40,9 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
|
||||
from . import setup_samsungtv_entry
|
||||
from .const import (
|
||||
ENTRYDATA_WEBSOCKET,
|
||||
MOCK_ENTRY_WS_WITH_MAC,
|
||||
MOCK_ENTRYDATA_ENCRYPTED_WS,
|
||||
MOCK_ENTRYDATA_WS,
|
||||
MOCK_SSDP_DATA_MAIN_TV_AGENT_ST,
|
||||
MOCK_SSDP_DATA_RENDERING_CONTROL_ST,
|
||||
)
|
||||
@ -57,7 +57,7 @@ MOCK_CONFIG = {
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "remoteencws_failing", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "remoteencws_failing", "rest_api")
|
||||
async def test_setup(hass: HomeAssistant) -> None:
|
||||
"""Test Samsung TV integration is setup."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIG)
|
||||
@ -101,7 +101,7 @@ async def test_setup_without_port_device_offline(hass: HomeAssistant) -> None:
|
||||
assert config_entries_domain[0].state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "remoteencws_failing", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "remoteencws_failing", "rest_api")
|
||||
async def test_setup_without_port_device_online(hass: HomeAssistant) -> None:
|
||||
"""Test import from yaml when the device is online."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIG)
|
||||
@ -111,7 +111,7 @@ async def test_setup_without_port_device_online(hass: HomeAssistant) -> None:
|
||||
assert config_entries_domain[0].data[CONF_MAC] == "aa:bb:aa:aa:aa:aa"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "remoteencws_failing")
|
||||
@pytest.mark.usefixtures("remote_websocket", "remoteencws_failing")
|
||||
async def test_setup_h_j_model(
|
||||
hass: HomeAssistant, rest_api: Mock, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
@ -126,13 +126,13 @@ async def test_setup_h_j_model(
|
||||
assert "H and J series use an encrypted protocol" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "remoteencws_failing", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_setup_updates_from_ssdp(
|
||||
hass: HomeAssistant, entity_registry: er.EntityRegistry, snapshot: SnapshotAssertion
|
||||
) -> None:
|
||||
"""Test setting up the entry fetches data from ssdp cache."""
|
||||
entry = MockConfigEntry(
|
||||
domain="samsungtv", data=MOCK_ENTRYDATA_WS, entry_id="sample-entry-id"
|
||||
domain="samsungtv", data=ENTRYDATA_WEBSOCKET, entry_id="sample-entry-id"
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
@ -200,7 +200,7 @@ async def test_update_imported_legacy(
|
||||
assert entries[0].data[CONF_PORT] == LEGACY_PORT
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_incorrectly_formatted_mac_fixed(hass: HomeAssistant) -> None:
|
||||
"""Test incorrectly formatted mac is corrected."""
|
||||
with patch(
|
||||
@ -230,7 +230,7 @@ async def test_incorrectly_formatted_mac_fixed(hass: HomeAssistant) -> None:
|
||||
assert config_entries[0].data[CONF_MAC] == "aa:bb:aa:aa:aa:aa"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
@pytest.mark.xfail
|
||||
async def test_cleanup_mac(
|
||||
hass: HomeAssistant, device_registry: dr.DeviceRegistry, snapshot: SnapshotAssertion
|
||||
|
@ -126,7 +126,7 @@ async def test_setup(hass: HomeAssistant) -> None:
|
||||
assert hass.states.get(ENTITY_ID)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_setup_websocket(hass: HomeAssistant) -> None:
|
||||
"""Test setup of platform."""
|
||||
with patch(
|
||||
@ -243,7 +243,10 @@ async def test_update_off(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -
|
||||
|
||||
|
||||
async def test_update_off_ws_no_power_state(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock, rest_api: Mock
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remote_websocket: Mock,
|
||||
rest_api: Mock,
|
||||
) -> None:
|
||||
"""Testing update tv off."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
@ -254,8 +257,8 @@ async def test_update_off_ws_no_power_state(
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
remotews.start_listening = Mock(side_effect=WebSocketException("Boom"))
|
||||
remotews.is_alive.return_value = False
|
||||
remote_websocket.start_listening = Mock(side_effect=WebSocketException("Boom"))
|
||||
remote_websocket.is_alive.return_value = False
|
||||
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
@ -266,9 +269,12 @@ async def test_update_off_ws_no_power_state(
|
||||
rest_api.rest_device_info.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews")
|
||||
@pytest.mark.usefixtures("remote_websocket")
|
||||
async def test_update_off_ws_with_power_state(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock, rest_api: Mock
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remote_websocket: Mock,
|
||||
rest_api: Mock,
|
||||
) -> None:
|
||||
"""Testing update tv off."""
|
||||
with (
|
||||
@ -276,7 +282,7 @@ async def test_update_off_ws_with_power_state(
|
||||
rest_api, "rest_device_info", side_effect=HttpApiError
|
||||
) as mock_device_info,
|
||||
patch.object(
|
||||
remotews, "start_listening", side_effect=WebSocketException("Boom")
|
||||
remote_websocket, "start_listening", side_effect=WebSocketException("Boom")
|
||||
) as mock_start_listening,
|
||||
):
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
@ -296,14 +302,14 @@ async def test_update_off_ws_with_power_state(
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
remotews.start_listening.assert_called_once()
|
||||
remote_websocket.start_listening.assert_called_once()
|
||||
rest_api.rest_device_info.assert_called_once()
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
|
||||
# After initial update, start_listening shouldn't be called
|
||||
remotews.start_listening.reset_mock()
|
||||
remote_websocket.start_listening.reset_mock()
|
||||
|
||||
# Second update uses device_info(ON)
|
||||
rest_api.rest_device_info.reset_mock()
|
||||
@ -330,7 +336,7 @@ async def test_update_off_ws_with_power_state(
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
remotews.start_listening.assert_not_called()
|
||||
remote_websocket.start_listening.assert_not_called()
|
||||
|
||||
|
||||
async def test_update_off_encryptedws(
|
||||
@ -391,7 +397,7 @@ async def test_update_access_denied(
|
||||
async def test_update_ws_connection_failure(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remotews: Mock,
|
||||
remote_websocket: Mock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Testing update tv connection failure exception."""
|
||||
@ -399,11 +405,11 @@ async def test_update_ws_connection_failure(
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
remotews,
|
||||
remote_websocket,
|
||||
"start_listening",
|
||||
side_effect=ConnectionFailure('{"event": "ms.voiceApp.hide"}'),
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
patch.object(remote_websocket, "is_alive", return_value=False),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
@ -421,16 +427,18 @@ async def test_update_ws_connection_failure(
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_update_ws_connection_closed(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remote_websocket: Mock
|
||||
) -> None:
|
||||
"""Testing update tv connection failure exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
remotews, "start_listening", side_effect=ConnectionClosedError(None, None)
|
||||
remote_websocket,
|
||||
"start_listening",
|
||||
side_effect=ConnectionClosedError(None, None),
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
patch.object(remote_websocket, "is_alive", return_value=False),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
@ -442,14 +450,16 @@ async def test_update_ws_connection_closed(
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_update_ws_unauthorized_error(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remote_websocket: Mock
|
||||
) -> None:
|
||||
"""Testing update tv unauthorized failure exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
|
||||
with (
|
||||
patch.object(remotews, "start_listening", side_effect=UnauthorizedError),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
patch.object(
|
||||
remote_websocket, "start_listening", side_effect=UnauthorizedError
|
||||
),
|
||||
patch.object(remote_websocket, "is_alive", return_value=False),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
@ -570,10 +580,12 @@ async def test_send_key_unhandled_response(
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_send_key_websocketexception(hass: HomeAssistant, remotews: Mock) -> None:
|
||||
async def test_send_key_websocketexception(
|
||||
hass: HomeAssistant, remote_websocket: Mock
|
||||
) -> None:
|
||||
"""Testing unhandled response exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
remotews.send_commands = Mock(side_effect=WebSocketException("Boom"))
|
||||
remote_websocket.send_commands = Mock(side_effect=WebSocketException("Boom"))
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
@ -596,10 +608,12 @@ async def test_send_key_websocketexception_encrypted(
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_send_key_os_error_ws(hass: HomeAssistant, remotews: Mock) -> None:
|
||||
async def test_send_key_os_error_ws(
|
||||
hass: HomeAssistant, remote_websocket: Mock
|
||||
) -> None:
|
||||
"""Testing unhandled response exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
remotews.send_commands = Mock(side_effect=OSError("Boom"))
|
||||
remote_websocket.send_commands = Mock(side_effect=OSError("Boom"))
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
@ -687,10 +701,10 @@ async def test_device_class(hass: HomeAssistant) -> None:
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_turn_off_websocket(
|
||||
hass: HomeAssistant, remotews: Mock, caplog: pytest.LogCaptureFixture
|
||||
hass: HomeAssistant, remote_websocket: Mock, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test for turn_off."""
|
||||
remotews.app_list_data = load_json_object_fixture(
|
||||
remote_websocket.app_list_data = load_json_object_fixture(
|
||||
"ws_installed_app_event.json", DOMAIN
|
||||
)
|
||||
with patch(
|
||||
@ -699,20 +713,20 @@ async def test_turn_off_websocket(
|
||||
):
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
|
||||
remotews.send_commands.reset_mock()
|
||||
remote_websocket.send_commands.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key called
|
||||
assert remotews.send_commands.call_count == 1
|
||||
commands = remotews.send_commands.call_args_list[0].args[0]
|
||||
assert remote_websocket.send_commands.call_count == 1
|
||||
commands = remote_websocket.send_commands.call_args_list[0].args[0]
|
||||
assert len(commands) == 1
|
||||
assert isinstance(commands[0], SendRemoteKey)
|
||||
assert commands[0].params["DataOfCmd"] == "KEY_POWER"
|
||||
|
||||
# commands not sent : power off in progress
|
||||
remotews.send_commands.reset_mock()
|
||||
remote_websocket.send_commands.reset_mock()
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
@ -724,11 +738,11 @@ async def test_turn_off_websocket(
|
||||
True,
|
||||
)
|
||||
assert "TV is powering off, not sending launch_app command" in caplog.text
|
||||
remotews.send_commands.assert_not_called()
|
||||
remote_websocket.send_commands.assert_not_called()
|
||||
|
||||
|
||||
async def test_turn_off_websocket_frame(
|
||||
hass: HomeAssistant, remotews: Mock, rest_api: Mock
|
||||
hass: HomeAssistant, remote_websocket: Mock, rest_api: Mock
|
||||
) -> None:
|
||||
"""Test for turn_off."""
|
||||
rest_api.rest_device_info.return_value = load_json_object_fixture(
|
||||
@ -740,14 +754,14 @@ async def test_turn_off_websocket_frame(
|
||||
):
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
|
||||
remotews.send_commands.reset_mock()
|
||||
remote_websocket.send_commands.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
# key called
|
||||
assert remotews.send_commands.call_count == 1
|
||||
commands = remotews.send_commands.call_args_list[0].args[0]
|
||||
assert remote_websocket.send_commands.call_count == 1
|
||||
commands = remote_websocket.send_commands.call_args_list[0].args[0]
|
||||
assert len(commands) == 3
|
||||
assert isinstance(commands[0], SendRemoteKey)
|
||||
assert commands[0].params["Cmd"] == "Press"
|
||||
@ -849,12 +863,12 @@ async def test_turn_off_os_error(
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_turn_off_ws_os_error(
|
||||
hass: HomeAssistant, remotews: Mock, caplog: pytest.LogCaptureFixture
|
||||
hass: HomeAssistant, remote_websocket: Mock, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test for turn_off with OSError."""
|
||||
caplog.set_level(logging.DEBUG)
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
remotews.close = Mock(side_effect=OSError("BOOM"))
|
||||
remote_websocket.close = Mock(side_effect=OSError("BOOM"))
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
@ -969,7 +983,7 @@ async def test_media_previous_track(hass: HomeAssistant, remote_legacy: Mock) ->
|
||||
assert remote_legacy.control.call_args_list == [call("KEY_CHDOWN")]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_turn_on_wol(hass: HomeAssistant) -> None:
|
||||
"""Test turn on."""
|
||||
entry = MockConfigEntry(
|
||||
@ -1126,10 +1140,10 @@ async def test_select_source_invalid_source(hass: HomeAssistant) -> None:
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_play_media_app(hass: HomeAssistant, remotews: Mock) -> None:
|
||||
async def test_play_media_app(hass: HomeAssistant, remote_websocket: Mock) -> None:
|
||||
"""Test for play_media."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
remotews.send_commands.reset_mock()
|
||||
remote_websocket.send_commands.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN,
|
||||
@ -1141,21 +1155,21 @@ async def test_play_media_app(hass: HomeAssistant, remotews: Mock) -> None:
|
||||
},
|
||||
True,
|
||||
)
|
||||
assert remotews.send_commands.call_count == 1
|
||||
commands = remotews.send_commands.call_args_list[0].args[0]
|
||||
assert remote_websocket.send_commands.call_count == 1
|
||||
commands = remote_websocket.send_commands.call_args_list[0].args[0]
|
||||
assert len(commands) == 1
|
||||
assert isinstance(commands[0], ChannelEmitCommand)
|
||||
assert commands[0].params["data"]["appId"] == "3201608010191"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_select_source_app(hass: HomeAssistant, remotews: Mock) -> None:
|
||||
async def test_select_source_app(hass: HomeAssistant, remote_websocket: Mock) -> None:
|
||||
"""Test for select_source."""
|
||||
remotews.app_list_data = load_json_object_fixture(
|
||||
remote_websocket.app_list_data = load_json_object_fixture(
|
||||
"ws_installed_app_event.json", DOMAIN
|
||||
)
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
remotews.send_commands.reset_mock()
|
||||
remote_websocket.send_commands.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN,
|
||||
@ -1163,8 +1177,8 @@ async def test_select_source_app(hass: HomeAssistant, remotews: Mock) -> None:
|
||||
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "Deezer"},
|
||||
True,
|
||||
)
|
||||
assert remotews.send_commands.call_count == 1
|
||||
commands = remotews.send_commands.call_args_list[0].args[0]
|
||||
assert remote_websocket.send_commands.call_count == 1
|
||||
commands = remote_websocket.send_commands.call_args_list[0].args[0]
|
||||
assert len(commands) == 1
|
||||
assert isinstance(commands[0], ChannelEmitCommand)
|
||||
assert commands[0].params["data"]["appId"] == "3201608010191"
|
||||
@ -1173,7 +1187,7 @@ async def test_select_source_app(hass: HomeAssistant, remotews: Mock) -> None:
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_websocket_unsupported_remote_control(
|
||||
hass: HomeAssistant,
|
||||
remotews: Mock,
|
||||
remote_websocket: Mock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
@ -1183,12 +1197,12 @@ async def test_websocket_unsupported_remote_control(
|
||||
assert entry.data[CONF_METHOD] == METHOD_WEBSOCKET
|
||||
assert entry.data[CONF_PORT] == 8001
|
||||
|
||||
remotews.send_commands.reset_mock()
|
||||
remote_websocket.send_commands.reset_mock()
|
||||
|
||||
await hass.services.async_call(
|
||||
MP_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
|
||||
)
|
||||
remotews.raise_mock_ws_event_callback(
|
||||
remote_websocket.raise_mock_ws_event_callback(
|
||||
"ms.error",
|
||||
{
|
||||
"event": "ms.error",
|
||||
@ -1197,8 +1211,8 @@ async def test_websocket_unsupported_remote_control(
|
||||
)
|
||||
|
||||
# key called
|
||||
assert remotews.send_commands.call_count == 1
|
||||
commands = remotews.send_commands.call_args_list[0].args[0]
|
||||
assert remote_websocket.send_commands.call_count == 1
|
||||
commands = remote_websocket.send_commands.call_args_list[0].args[0]
|
||||
assert len(commands) == 1
|
||||
assert isinstance(commands[0], SendRemoteKey)
|
||||
assert commands[0].params["DataOfCmd"] == "KEY_POWER"
|
||||
@ -1227,7 +1241,7 @@ async def test_websocket_unsupported_remote_control(
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "upnp_notify_server")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "upnp_notify_server")
|
||||
async def test_volume_control_upnp(hass: HomeAssistant, dmr_device: Mock) -> None:
|
||||
"""Test for Upnp volume control."""
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
|
||||
@ -1261,7 +1275,7 @@ async def test_volume_control_upnp(hass: HomeAssistant, dmr_device: Mock) -> Non
|
||||
dmr_device.async_set_volume_level.assert_called_once_with(0.6)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_upnp_not_available(
|
||||
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
@ -1279,7 +1293,7 @@ async def test_upnp_not_available(
|
||||
assert "Upnp services are not available" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "upnp_factory")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "upnp_factory")
|
||||
async def test_upnp_missing_service(
|
||||
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
@ -1297,7 +1311,7 @@ async def test_upnp_missing_service(
|
||||
assert "Upnp services are not available" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_upnp_shutdown(
|
||||
hass: HomeAssistant,
|
||||
dmr_device: Mock,
|
||||
@ -1318,7 +1332,7 @@ async def test_upnp_shutdown(
|
||||
upnp_notify_server.async_stop_server.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api", "upnp_notify_server")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api", "upnp_notify_server")
|
||||
async def test_upnp_subscribe_events(hass: HomeAssistant, dmr_device: Mock) -> None:
|
||||
"""Test for Upnp event feedback."""
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
|
||||
@ -1338,7 +1352,7 @@ async def test_upnp_subscribe_events(hass: HomeAssistant, dmr_device: Mock) -> N
|
||||
assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is True
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_upnp_subscribe_events_upnperror(
|
||||
hass: HomeAssistant,
|
||||
dmr_device: Mock,
|
||||
@ -1353,7 +1367,7 @@ async def test_upnp_subscribe_events_upnperror(
|
||||
assert "Error while subscribing during device connect" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_upnp_subscribe_events_upnpresponseerror(
|
||||
hass: HomeAssistant,
|
||||
dmr_device: Mock,
|
||||
@ -1376,7 +1390,7 @@ async def test_upnp_subscribe_events_upnpresponseerror(
|
||||
async def test_upnp_re_subscribe_events(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remotews: Mock,
|
||||
remote_websocket: Mock,
|
||||
dmr_device: Mock,
|
||||
) -> None:
|
||||
"""Test for Upnp event feedback."""
|
||||
@ -1389,9 +1403,9 @@ async def test_upnp_re_subscribe_events(
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
remotews, "start_listening", side_effect=WebSocketException("Boom")
|
||||
remote_websocket, "start_listening", side_effect=WebSocketException("Boom")
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
patch.object(remote_websocket, "is_alive", return_value=False),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
@ -1420,7 +1434,7 @@ async def test_upnp_re_subscribe_events(
|
||||
async def test_upnp_failed_re_subscribe_events(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remotews: Mock,
|
||||
remote_websocket: Mock,
|
||||
dmr_device: Mock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
error: Exception,
|
||||
@ -1435,9 +1449,9 @@ async def test_upnp_failed_re_subscribe_events(
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
remotews, "start_listening", side_effect=WebSocketException("Boom")
|
||||
remote_websocket, "start_listening", side_effect=WebSocketException("Boom")
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
patch.object(remote_websocket, "is_alive", return_value=False),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
|
@ -98,7 +98,7 @@ async def test_send_command_service(hass: HomeAssistant, remoteencws: Mock) -> N
|
||||
assert command.body["param3"] == "dash"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remotews", "rest_api")
|
||||
@pytest.mark.usefixtures("remote_websocket", "rest_api")
|
||||
async def test_turn_on_wol(hass: HomeAssistant) -> None:
|
||||
"""Test turn on."""
|
||||
entry = MockConfigEntry(
|
||||
|
Loading…
x
Reference in New Issue
Block a user