diff --git a/.coveragerc b/.coveragerc index 4b11d681bab..4a46152ec01 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1297,7 +1297,6 @@ omit = homeassistant/components/waze_travel_time/helpers.py homeassistant/components/waze_travel_time/sensor.py homeassistant/components/webostv/__init__.py - homeassistant/components/webostv/media_player.py homeassistant/components/whois/__init__.py homeassistant/components/whois/sensor.py homeassistant/components/wiffi/* diff --git a/tests/components/webostv/__init__.py b/tests/components/webostv/__init__.py index 96c83b33c41..7a4ca9d694d 100644 --- a/tests/components/webostv/__init__.py +++ b/tests/components/webostv/__init__.py @@ -12,6 +12,17 @@ TV_NAME = "fake" ENTITY_ID = f"{MP_DOMAIN}.{TV_NAME}" MOCK_CLIENT_KEYS = {"1.2.3.4": "some-secret"} +CHANNEL_1 = { + "channelNumber": "1", + "channelName": "Channel 1", + "channelId": "ch1id", +} +CHANNEL_2 = { + "channelNumber": "20", + "channelName": "Channel Name 2", + "channelId": "ch2id", +} + async def setup_webostv(hass, unique_id="some-unique-id"): """Initialize webostv and media_player for tests.""" diff --git a/tests/components/webostv/conftest.py b/tests/components/webostv/conftest.py index 43491580419..b99a04f41c3 100644 --- a/tests/components/webostv/conftest.py +++ b/tests/components/webostv/conftest.py @@ -1,8 +1,12 @@ """Common fixtures and objects for the LG webOS integration tests.""" -from unittest.mock import patch +from unittest.mock import AsyncMock, patch import pytest +from homeassistant.components.webostv.const import LIVE_TV_APP_ID + +from . import CHANNEL_1, CHANNEL_2 + from tests.common import async_mock_service @@ -20,10 +24,34 @@ def client_fixture(): ) as mock_client_class: client = mock_client_class.return_value client.hello_info = {"deviceUUID": "some-fake-uuid"} - client.software_info = {"device_id": "00:01:02:03:04:05"} + client.software_info = {"major_ver": "major", "minor_ver": "minor"} client.system_info = {"modelName": "TVFAKE"} client.client_key = "0123456789" - client.apps = {0: {"title": "Applicaiton01"}} - client.inputs = {0: {"label": "Input01"}, 1: {"label": "Input02"}} + client.apps = { + LIVE_TV_APP_ID: { + "title": "Live TV", + "id": LIVE_TV_APP_ID, + "largeIcon": "large-icon", + "icon": "icon", + }, + } + client.inputs = { + "in1": {"label": "Input01", "id": "in1", "appId": "app0"}, + "in2": {"label": "Input02", "id": "in2", "appId": "app1"}, + } + client.current_app_id = LIVE_TV_APP_ID + + client.channels = [CHANNEL_1, CHANNEL_2] + client.current_channel = CHANNEL_1 + + client.volume = 37 + client.sound_output = "speaker" + client.muted = False + client.is_on = True + + async def mock_state_update_callback(): + await client.register_state_update_callback.call_args[0][0](client) + + client.mock_state_update = AsyncMock(side_effect=mock_state_update_callback) yield client diff --git a/tests/components/webostv/test_media_player.py b/tests/components/webostv/test_media_player.py index cbae84ac66d..a4071c741e5 100644 --- a/tests/components/webostv/test_media_player.py +++ b/tests/components/webostv/test_media_player.py @@ -1,50 +1,211 @@ """The tests for the LG webOS media player platform.""" -from homeassistant.components.media_player import DOMAIN as MP_DOMAIN +import asyncio +from datetime import timedelta +from unittest.mock import Mock + +import pytest + +from homeassistant.components import automation +from homeassistant.components.media_player import ( + DOMAIN as MP_DOMAIN, + MediaPlayerDeviceClass, +) from homeassistant.components.media_player.const import ( ATTR_INPUT_SOURCE, + ATTR_INPUT_SOURCE_LIST, + ATTR_MEDIA_CONTENT_ID, + ATTR_MEDIA_CONTENT_TYPE, + ATTR_MEDIA_TITLE, + ATTR_MEDIA_VOLUME_LEVEL, ATTR_MEDIA_VOLUME_MUTED, + MEDIA_TYPE_CHANNEL, + SERVICE_PLAY_MEDIA, SERVICE_SELECT_SOURCE, + SUPPORT_TURN_ON, + SUPPORT_VOLUME_SET, ) from homeassistant.components.webostv.const import ( ATTR_BUTTON, ATTR_PAYLOAD, + ATTR_SOUND_OUTPUT, DOMAIN, + LIVE_TV_APP_ID, SERVICE_BUTTON, SERVICE_COMMAND, + SERVICE_SELECT_SOUND_OUTPUT, + WebOsTvCommandError, ) -from homeassistant.const import ATTR_COMMAND, ATTR_ENTITY_ID, SERVICE_VOLUME_MUTE +from homeassistant.components.webostv.media_player import ( + SUPPORT_WEBOSTV, + SUPPORT_WEBOSTV_VOLUME, +) +from homeassistant.const import ( + ATTR_COMMAND, + ATTR_DEVICE_CLASS, + ATTR_ENTITY_ID, + ATTR_SUPPORTED_FEATURES, + ENTITY_MATCH_NONE, + SERVICE_MEDIA_NEXT_TRACK, + SERVICE_MEDIA_PAUSE, + SERVICE_MEDIA_PLAY, + SERVICE_MEDIA_PLAY_PAUSE, + SERVICE_MEDIA_PREVIOUS_TRACK, + SERVICE_MEDIA_STOP, + SERVICE_TURN_OFF, + SERVICE_VOLUME_DOWN, + SERVICE_VOLUME_MUTE, + SERVICE_VOLUME_SET, + SERVICE_VOLUME_UP, + STATE_OFF, + STATE_ON, +) +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import device_registry +from homeassistant.setup import async_setup_component +from homeassistant.util import dt -from . import ENTITY_ID, setup_webostv +from . import CHANNEL_2, ENTITY_ID, TV_NAME, setup_webostv + +from tests.common import async_fire_time_changed -async def test_mute(hass, client): - """Test simple service call.""" +@pytest.mark.parametrize( + "service, attr_data, client_call", + [ + (SERVICE_VOLUME_MUTE, {ATTR_MEDIA_VOLUME_MUTED: True}, ("set_mute", True)), + (SERVICE_VOLUME_MUTE, {ATTR_MEDIA_VOLUME_MUTED: False}, ("set_mute", False)), + (SERVICE_VOLUME_SET, {ATTR_MEDIA_VOLUME_LEVEL: 1.00}, ("set_volume", 100)), + (SERVICE_VOLUME_SET, {ATTR_MEDIA_VOLUME_LEVEL: 0.54}, ("set_volume", 54)), + (SERVICE_VOLUME_SET, {ATTR_MEDIA_VOLUME_LEVEL: 0.0}, ("set_volume", 0)), + ], +) +async def test_services_with_parameters(hass, client, service, attr_data, client_call): + """Test services that has parameters in calls.""" await setup_webostv(hass) - data = { - ATTR_ENTITY_ID: ENTITY_ID, - ATTR_MEDIA_VOLUME_MUTED: True, - } + data = {ATTR_ENTITY_ID: ENTITY_ID, **attr_data} + assert await hass.services.async_call(MP_DOMAIN, service, data, True) - assert await hass.services.async_call(MP_DOMAIN, SERVICE_VOLUME_MUTE, data, True) - await hass.async_block_till_done() - - client.set_mute.assert_called_once() + getattr(client, client_call[0]).assert_called_once_with(client_call[1]) -async def test_select_source_with_empty_source_list(hass, client): +@pytest.mark.parametrize( + "service, client_call", + [ + (SERVICE_TURN_OFF, "power_off"), + (SERVICE_VOLUME_UP, "volume_up"), + (SERVICE_VOLUME_DOWN, "volume_down"), + (SERVICE_MEDIA_PLAY, "play"), + (SERVICE_MEDIA_PAUSE, "pause"), + (SERVICE_MEDIA_STOP, "stop"), + ], +) +async def test_services(hass, client, service, client_call): + """Test simple services without parameters.""" + await setup_webostv(hass) + + data = {ATTR_ENTITY_ID: ENTITY_ID} + assert await hass.services.async_call(MP_DOMAIN, service, data, True) + + getattr(client, client_call).assert_called_once() + + +async def test_media_play_pause(hass, client): + """Test media play pause service.""" + await setup_webostv(hass) + + data = {ATTR_ENTITY_ID: ENTITY_ID} + + # After init state is playing - check pause call + assert await hass.services.async_call( + MP_DOMAIN, SERVICE_MEDIA_PLAY_PAUSE, data, True + ) + + client.pause.assert_called_once() + client.play.assert_not_called() + + # After pause state is paused - check play call + assert await hass.services.async_call( + MP_DOMAIN, SERVICE_MEDIA_PLAY_PAUSE, data, True + ) + + client.play.assert_called_once() + client.pause.assert_called_once() + + +@pytest.mark.parametrize( + "service, client_call", + [ + (SERVICE_MEDIA_NEXT_TRACK, ("fast_forward", "channel_up")), + (SERVICE_MEDIA_PREVIOUS_TRACK, ("rewind", "channel_down")), + ], +) +async def test_media_next_previous_track( + hass, client, service, client_call, monkeypatch +): + """Test media next/previous track services.""" + await setup_webostv(hass) + + # check channel up/down for live TV channels + data = {ATTR_ENTITY_ID: ENTITY_ID} + assert await hass.services.async_call(MP_DOMAIN, service, data, True) + + getattr(client, client_call[0]).assert_not_called() + getattr(client, client_call[1]).assert_called_once() + + # check next/previous for not Live TV channels + monkeypatch.setattr(client, "current_app_id", "in1") + data = {ATTR_ENTITY_ID: ENTITY_ID} + assert await hass.services.async_call(MP_DOMAIN, service, data, True) + + getattr(client, client_call[0]).assert_called_once() + getattr(client, client_call[1]).assert_called_once() + + +async def test_select_source_with_empty_source_list(hass, client, caplog): """Ensure we don't call client methods when we don't have sources.""" await setup_webostv(hass) + await client.mock_state_update() data = { ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "nonexistent", } - await hass.services.async_call(MP_DOMAIN, SERVICE_SELECT_SOURCE, data) - await hass.async_block_till_done() + assert await hass.services.async_call(MP_DOMAIN, SERVICE_SELECT_SOURCE, data, True) client.launch_app.assert_not_called() client.set_input.assert_not_called() + assert f"Source nonexistent not found for {TV_NAME}" in caplog.text + + +async def test_select_app_source(hass, client): + """Test select app source.""" + await setup_webostv(hass) + await client.mock_state_update() + + data = { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_INPUT_SOURCE: "Live TV", + } + assert await hass.services.async_call(MP_DOMAIN, SERVICE_SELECT_SOURCE, data, True) + + client.launch_app.assert_called_once_with(LIVE_TV_APP_ID) + client.set_input.assert_not_called() + + +async def test_select_input_source(hass, client): + """Test select input source.""" + await setup_webostv(hass) + await client.mock_state_update() + + data = { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_INPUT_SOURCE: "Input01", + } + assert await hass.services.async_call(MP_DOMAIN, SERVICE_SELECT_SOURCE, data, True) + + client.launch_app.assert_not_called() + client.set_input.assert_called_once_with("in1") async def test_button(hass, client): @@ -55,8 +216,7 @@ async def test_button(hass, client): ATTR_ENTITY_ID: ENTITY_ID, ATTR_BUTTON: "test", } - await hass.services.async_call(DOMAIN, SERVICE_BUTTON, data) - await hass.async_block_till_done() + assert await hass.services.async_call(DOMAIN, SERVICE_BUTTON, data, True) client.button.assert_called_once() client.button.assert_called_with("test") @@ -70,8 +230,7 @@ async def test_command(hass, client): ATTR_ENTITY_ID: ENTITY_ID, ATTR_COMMAND: "test", } - await hass.services.async_call(DOMAIN, SERVICE_COMMAND, data) - await hass.async_block_till_done() + assert await hass.services.async_call(DOMAIN, SERVICE_COMMAND, data, True) client.request.assert_called_with("test", payload=None) @@ -85,9 +244,314 @@ async def test_command_with_optional_arg(hass, client): ATTR_COMMAND: "test", ATTR_PAYLOAD: {"target": "https://www.google.com"}, } - await hass.services.async_call(DOMAIN, SERVICE_COMMAND, data) - await hass.async_block_till_done() + assert await hass.services.async_call(DOMAIN, SERVICE_COMMAND, data, True) client.request.assert_called_with( "test", payload={"target": "https://www.google.com"} ) + + +async def test_select_sound_output(hass, client): + """Test select sound output service.""" + await setup_webostv(hass) + + data = { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_SOUND_OUTPUT: "external_speaker", + } + assert await hass.services.async_call( + DOMAIN, SERVICE_SELECT_SOUND_OUTPUT, data, True + ) + + client.change_sound_output.assert_called_once_with("external_speaker") + + +async def test_device_info_startup_off(hass, client, monkeypatch): + """Test device info when device is off at startup.""" + monkeypatch.setattr(client, "system_info", None) + monkeypatch.setattr(client, "is_on", False) + entry = await setup_webostv(hass) + await client.mock_state_update() + + assert hass.states.get(ENTITY_ID).state == STATE_OFF + + device_reg = device_registry.async_get(hass) + device = device_reg.async_get_device({(DOMAIN, entry.unique_id)}) + + assert device + assert device.identifiers == {(DOMAIN, entry.unique_id)} + assert device.manufacturer == "LG" + assert device.name == TV_NAME + assert device.sw_version is None + assert device.model is None + + +async def test_entity_attributes(hass, client, monkeypatch): + """Test entity attributes.""" + entry = await setup_webostv(hass) + await client.mock_state_update() + + # Attributes when device is on + state = hass.states.get(ENTITY_ID) + attrs = state.attributes + + assert state.state == STATE_ON + assert state.name == TV_NAME + assert attrs[ATTR_DEVICE_CLASS] == MediaPlayerDeviceClass.TV + assert attrs[ATTR_MEDIA_VOLUME_MUTED] is False + assert attrs[ATTR_MEDIA_VOLUME_LEVEL] == 0.37 + assert attrs[ATTR_INPUT_SOURCE] == "Live TV" + assert attrs[ATTR_INPUT_SOURCE_LIST] == ["Input01", "Input02", "Live TV"] + assert attrs[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_CHANNEL + assert attrs[ATTR_MEDIA_TITLE] == "Channel 1" + assert attrs[ATTR_SOUND_OUTPUT] == "speaker" + + # Volume level not available + monkeypatch.setattr(client, "volume", None) + await client.mock_state_update() + attrs = hass.states.get(ENTITY_ID).attributes + + assert attrs.get(ATTR_MEDIA_VOLUME_LEVEL) is None + + # Channel change + monkeypatch.setattr(client, "current_channel", CHANNEL_2) + await client.mock_state_update() + attrs = hass.states.get(ENTITY_ID).attributes + + assert attrs[ATTR_MEDIA_TITLE] == "Channel Name 2" + + # Device Info + device_reg = device_registry.async_get(hass) + device = device_reg.async_get_device({(DOMAIN, entry.unique_id)}) + + assert device + assert device.identifiers == {(DOMAIN, entry.unique_id)} + assert device.manufacturer == "LG" + assert device.name == TV_NAME + assert device.sw_version == "major.minor" + assert device.model == "TVFAKE" + + # Sound output when off + monkeypatch.setattr(client, "sound_output", None) + monkeypatch.setattr(client, "is_on", False) + await client.mock_state_update() + state = hass.states.get(ENTITY_ID) + + assert state.state == STATE_OFF + assert state.attributes.get(ATTR_SOUND_OUTPUT) is None + + +async def test_service_entity_id_none(hass, client): + """Test service call with none as entity id.""" + await setup_webostv(hass) + + data = { + ATTR_ENTITY_ID: ENTITY_MATCH_NONE, + ATTR_SOUND_OUTPUT: "external_speaker", + } + assert await hass.services.async_call( + DOMAIN, SERVICE_SELECT_SOUND_OUTPUT, data, True + ) + + client.change_sound_output.assert_not_called() + + +@pytest.mark.parametrize( + "media_id, ch_id", + [ + ("Channel 1", "ch1id"), # Perfect Match by channel name + ("Name 2", "ch2id"), # Partial Match by channel name + ("20", "ch2id"), # Perfect Match by channel number + ], +) +async def test_play_media(hass, client, media_id, ch_id): + """Test play media service.""" + await setup_webostv(hass) + await client.mock_state_update() + + data = { + ATTR_ENTITY_ID: ENTITY_ID, + ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_CHANNEL, + ATTR_MEDIA_CONTENT_ID: media_id, + } + assert await hass.services.async_call(MP_DOMAIN, SERVICE_PLAY_MEDIA, data, True) + + client.set_channel.assert_called_once_with(ch_id) + + +async def test_update_sources_live_tv_find(hass, client, monkeypatch): + """Test finding live TV app id in update sources.""" + await setup_webostv(hass) + await client.mock_state_update() + + # Live TV found in app list + sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST] + + assert "Live TV" in sources + assert len(sources) == 3 + + # Live TV is current app + apps = { + LIVE_TV_APP_ID: { + "title": "Live TV", + "id": "some_id", + }, + } + monkeypatch.setattr(client, "apps", apps) + monkeypatch.setattr(client, "current_app_id", "some_id") + await client.mock_state_update() + sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST] + + assert "Live TV" in sources + assert len(sources) == 3 + + # Live TV is is in inputs + inputs = { + LIVE_TV_APP_ID: { + "label": "Live TV", + "id": "some_id", + "appId": LIVE_TV_APP_ID, + }, + } + monkeypatch.setattr(client, "inputs", inputs) + await client.mock_state_update() + sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST] + + assert "Live TV" in sources + assert len(sources) == 1 + + # Live TV is current input + inputs = { + LIVE_TV_APP_ID: { + "label": "Live TV", + "id": "some_id", + "appId": "some_id", + }, + } + monkeypatch.setattr(client, "inputs", inputs) + await client.mock_state_update() + sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST] + + assert "Live TV" in sources + assert len(sources) == 1 + + # Live TV not found + monkeypatch.setattr(client, "current_app_id", "other_id") + await client.mock_state_update() + sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST] + + assert "Live TV" in sources + assert len(sources) == 1 + + # Live TV not found in sources/apps but is current app + monkeypatch.setattr(client, "apps", {}) + monkeypatch.setattr(client, "current_app_id", LIVE_TV_APP_ID) + await client.mock_state_update() + sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST] + + assert "Live TV" in sources + assert len(sources) == 1 + + # Bad update, keep old update + monkeypatch.setattr(client, "inputs", {}) + await client.mock_state_update() + sources = hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE_LIST] + + assert "Live TV" in sources + assert len(sources) == 1 + + +async def test_client_disconnected(hass, client, monkeypatch): + """Test error not raised when client is disconnected.""" + await setup_webostv(hass) + monkeypatch.setattr(client, "is_connected", Mock(return_value=False)) + monkeypatch.setattr(client, "connect", Mock(side_effect=asyncio.TimeoutError)) + + async_fire_time_changed(hass, dt.utcnow() + timedelta(seconds=20)) + await hass.async_block_till_done() + + +async def test_control_error_handling(hass, client, caplog, monkeypatch): + """Test control errors handling.""" + await setup_webostv(hass) + monkeypatch.setattr(client, "play", Mock(side_effect=WebOsTvCommandError)) + data = {ATTR_ENTITY_ID: ENTITY_ID} + + # Device on, raise HomeAssistantError + with pytest.raises(HomeAssistantError) as exc: + assert await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PLAY, data, True) + + assert ( + str(exc.value) + == f"Error calling async_media_play on entity {ENTITY_ID}, state:on" + ) + assert client.play.call_count == 1 + + # Device off, log a warning + monkeypatch.setattr(client, "is_on", False) + monkeypatch.setattr(client, "play", Mock(side_effect=asyncio.TimeoutError)) + await client.mock_state_update() + assert await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PLAY, data, True) + + assert client.play.call_count == 1 + assert ( + f"Error calling async_media_play on entity {ENTITY_ID}, state:off, error: TimeoutError()" + in caplog.text + ) + + +async def test_supported_features(hass, client, monkeypatch): + """Test test supported features.""" + monkeypatch.setattr(client, "sound_output", "lineout") + await setup_webostv(hass) + await client.mock_state_update() + + # No sound control support + supported = SUPPORT_WEBOSTV + attrs = hass.states.get(ENTITY_ID).attributes + + assert attrs[ATTR_SUPPORTED_FEATURES] == supported + + # Support volume mute, step + monkeypatch.setattr(client, "sound_output", "external_speaker") + await client.mock_state_update() + supported = supported | SUPPORT_WEBOSTV_VOLUME + attrs = hass.states.get(ENTITY_ID).attributes + + assert attrs[ATTR_SUPPORTED_FEATURES] == supported + + # Support volume mute, step, set + monkeypatch.setattr(client, "sound_output", "speaker") + await client.mock_state_update() + supported = supported | SUPPORT_WEBOSTV_VOLUME | SUPPORT_VOLUME_SET + attrs = hass.states.get(ENTITY_ID).attributes + + assert attrs[ATTR_SUPPORTED_FEATURES] == supported + + # Support turn on + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + "platform": "webostv.turn_on", + "entity_id": ENTITY_ID, + }, + "action": { + "service": "test.automation", + "data_template": { + "some": ENTITY_ID, + "id": "{{ trigger.id }}", + }, + }, + }, + ], + }, + ) + supported |= SUPPORT_TURN_ON + await client.mock_state_update() + attrs = hass.states.get(ENTITY_ID).attributes + + assert attrs[ATTR_SUPPORTED_FEATURES] == supported