innocuous-symmetry/core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use DmrDevice to communicate with SamsungTV (#68777)

Browse Source 
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet
2022-03-29 00:23:38 +02:00
committed by GitHub
parent d0e5e51863
commit 8eb2e131e5
4 changed files with 222 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,7 +4,11 @@ from datetime import datetime, timedelta
import logging
from unittest.mock import DEFAULT as DEFAULT_MOCK, AsyncMock, Mock, call, patch
from async_upnp_client.exceptions import UpnpActionResponseError
from async_upnp_client.exceptions import (
UpnpActionResponseError,
UpnpError,
UpnpResponseError,
)
import pytest
from samsungctl import exceptions
from samsungtvws.async_remote import SamsungTVWSAsyncRemote
@@ -42,10 +46,7 @@ from homeassistant.components.samsungtv.const import (
METHOD_WEBSOCKET,
TIMEOUT_WEBSOCKET,
)
from homeassistant.components.samsungtv.media_player import (
SUPPORT_SAMSUNGTV,
UPNP_SVC_RENDERING_CONTROL,
)
from homeassistant.components.samsungtv.media_player import SUPPORT_SAMSUNGTV
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
@@ -80,11 +81,7 @@ from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from . import (
async_wait_config_entry_reload,
setup_samsungtv_entry,
upnp_get_action_mock,
)
from . import async_wait_config_entry_reload, setup_samsungtv_entry
from .const import (
MOCK_ENTRYDATA_ENCRYPTED_WS,
SAMPLE_DEVICE_INFO_FRAME,
@@ -1329,39 +1326,30 @@ async def test_websocket_unsupported_remote_control(
assert state.state == STATE_UNAVAILABLE
@pytest.mark.usefixtures("remotews", "rest_api")
@pytest.mark.usefixtures("remotews", "rest_api", "upnp_notify_server")
async def test_volume_control_upnp(
hass: HomeAssistant, upnp_device: Mock, caplog: pytest.LogCaptureFixture
hass: HomeAssistant, dmr_device: Mock, caplog: pytest.LogCaptureFixture
) -> None:
"""Test for Upnp volume control."""
upnp_get_volume = upnp_get_action_mock(
upnp_device, UPNP_SVC_RENDERING_CONTROL, "GetVolume"
)
upnp_get_volume.async_call.return_value = {"CurrentVolume": 44}
upnp_get_mute = upnp_get_action_mock(
upnp_device, UPNP_SVC_RENDERING_CONTROL, "GetMute"
)
upnp_get_mute.async_call.return_value = {"CurrentMute": False}
await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
upnp_get_volume.async_call.assert_called_once()
upnp_get_mute.async_call.assert_called_once()
state = hass.states.get(ENTITY_ID)
assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.44
assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is False
# Upnp action succeeds
upnp_set_volume = upnp_get_action_mock(
upnp_device, UPNP_SVC_RENDERING_CONTROL, "SetVolume"
)
assert await hass.services.async_call(
DOMAIN,
SERVICE_VOLUME_SET,
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_LEVEL: 0.5},
True,
)
dmr_device.async_set_volume_level.assert_called_once_with(0.5)
assert "Unable to set volume level on" not in caplog.text
# Upnp action failed
upnp_set_volume.async_call.side_effect = UpnpActionResponseError(
dmr_device.async_set_volume_level.reset_mock()
dmr_device.async_set_volume_level.side_effect = UpnpActionResponseError(
status=500, error_code=501, error_desc="Action Failed"
)
assert await hass.services.async_call(
@@ -1370,6 +1358,7 @@ async def test_volume_control_upnp(
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_LEVEL: 0.6},
True,
)
dmr_device.async_set_volume_level.assert_called_once_with(0.6)
assert "Unable to set volume level on" in caplog.text
@@ -1390,7 +1379,7 @@ async def test_upnp_not_available(
assert "Upnp services are not available" in caplog.text
@pytest.mark.usefixtures("remotews", "upnp_device", "rest_api")
@pytest.mark.usefixtures("remotews", "rest_api", "upnp_factory")
async def test_upnp_missing_service(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
@@ -1404,4 +1393,79 @@ async def test_upnp_missing_service(
{ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_LEVEL: 0.6},
True,
)
assert f"Upnp service {UPNP_SVC_RENDERING_CONTROL} is not available" in caplog.text
assert "Upnp services are not available" in caplog.text
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_upnp_shutdown(
hass: HomeAssistant,
dmr_device: Mock,
upnp_notify_server: Mock,
) -> None:
"""Ensure that Upnp cleanup takes effect."""
entry = await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
assert await entry.async_unload(hass)
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_UNAVAILABLE
dmr_device.async_unsubscribe_services.assert_called_once()
upnp_notify_server.async_stop_server.assert_called_once()
@pytest.mark.usefixtures("remotews", "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)
state = hass.states.get(ENTITY_ID)
assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.44
assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is False
# DMR Devices gets updated, and raise event
dmr_device.volume_level = 0
dmr_device.is_volume_muted = True
dmr_device.raise_event(None, None)
# State gets updated without the need to wait for next update
state = hass.states.get(ENTITY_ID)
assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0
assert state.attributes[ATTR_MEDIA_VOLUME_MUTED] is True
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_upnp_subscribe_events_upnperror(
hass: HomeAssistant,
dmr_device: Mock,
upnp_notify_server: Mock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test for failure to subscribe Upnp services."""
with patch.object(dmr_device, "async_subscribe_services", side_effect=UpnpError):
await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
upnp_notify_server.async_stop_server.assert_called_once()
assert "Error while subscribing during device connect" in caplog.text
@pytest.mark.usefixtures("remotews", "rest_api")
async def test_upnp_subscribe_events_upnpresponseerror(
hass: HomeAssistant,
dmr_device: Mock,
upnp_notify_server: Mock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test for failure to subscribe Upnp services."""
with patch.object(
dmr_device,
"async_subscribe_services",
side_effect=UpnpResponseError(status=501),
):
await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
upnp_notify_server.async_stop_server.assert_not_called()
assert "Device rejected subscription" in caplog.text