jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 01:08:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add webostv 100% tests coverage for media player (#64723)

Browse Source
This commit is contained in:
Shay Levy 2022-01-23 02:06:48 +02:00 committed by GitHub
parent 5654490aa8
commit 12780a3173
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 529 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -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/*

11
tests/components/webostv/__init__.py
View File

@ -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."""

36
tests/components/webostv/conftest.py
View File

@ -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

508
tests/components/webostv/test_media_player.py
View File

@ -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