Add tests for the HDMI-CEC integration (#75094)

* Add basic tests to the HDMI-CEC component

* Add tests for the HDMI-CEC switch component

* Add test for watchdog code

* Start adding tests for the HDMI-CEC media player platform

Also some cleanup and code move.

* Add more tests for media_player

And cleanup some switch tests.

* Improve xfail message for features

* Align test pyCEC dependency with main dependency

* Make fixtures snake_case

* Cleanup call asserts

* Cleanup service tests

* fix issues with media player tests

* Cleanup MockHDMIDevice class

* Cleanup watchdog tests

* Add myself as code owner for the HDMI-CEC integration

* Fix async fire time changed time jump

* Fix event api sync context

* Delint tests

* Parametrize watchdog test

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/76460/head
Pieter Mulder 2022-08-08 13:47:05 +02:00 committed by GitHub
parent a1d5a4bc79
commit 7cd4be1310
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 1329 additions and 3 deletions

View File

@ -473,7 +473,6 @@ omit =
homeassistant/components/harmony/remote.py
homeassistant/components/harmony/util.py
homeassistant/components/haveibeenpwned/sensor.py
homeassistant/components/hdmi_cec/*
homeassistant/components/heatmiser/climate.py
homeassistant/components/hikvision/binary_sensor.py
homeassistant/components/hikvisioncam/switch.py

View File

@ -436,6 +436,8 @@ build.json @home-assistant/supervisor
/tests/components/harmony/ @ehendrix23 @bramkragten @bdraco @mkeesey @Aohzan
/homeassistant/components/hassio/ @home-assistant/supervisor
/tests/components/hassio/ @home-assistant/supervisor
/homeassistant/components/hdmi_cec/ @inytar
/tests/components/hdmi_cec/ @inytar
/homeassistant/components/heatmiser/ @andylockran
/homeassistant/components/heos/ @andrewsayre
/tests/components/heos/ @andrewsayre

View File

@ -219,7 +219,7 @@ def setup(hass: HomeAssistant, base_config: ConfigType) -> bool: # noqa: C901
def _adapter_watchdog(now=None):
_LOGGER.debug("Reached _adapter_watchdog")
event.async_call_later(hass, WATCHDOG_INTERVAL, _adapter_watchdog)
event.call_later(hass, WATCHDOG_INTERVAL, _adapter_watchdog)
if not adapter.initialized:
_LOGGER.info("Adapter not initialized; Trying to restart")
hass.bus.fire(EVENT_HDMI_CEC_UNAVAILABLE)

View File

@ -3,7 +3,7 @@
"name": "HDMI-CEC",
"documentation": "https://www.home-assistant.io/integrations/hdmi_cec",
"requirements": ["pyCEC==0.5.2"],
"codeowners": [],
"codeowners": ["@inytar"],
"iot_class": "local_push",
"loggers": ["pycec"]
}

View File

@ -936,6 +936,9 @@ py-synologydsm-api==1.0.8
# homeassistant.components.seventeentrack
py17track==2021.12.2
# homeassistant.components.hdmi_cec
pyCEC==0.5.2
# homeassistant.components.control4
pyControl4==0.0.6

View File

@ -0,0 +1,49 @@
"""Tests for the HDMI-CEC component."""
from unittest.mock import AsyncMock, Mock
from homeassistant.components.hdmi_cec import KeyPressCommand, KeyReleaseCommand
class MockHDMIDevice:
"""Mock of a HDMIDevice."""
def __init__(self, *, logical_address, **values):
"""Mock of a HDMIDevice."""
self.set_update_callback = Mock(side_effect=self._set_update_callback)
self.logical_address = logical_address
self.name = f"hdmi_{logical_address:x}"
if "power_status" not in values:
# Default to invalid state.
values["power_status"] = -1
self._values = values
self.turn_on = Mock()
self.turn_off = Mock()
self.send_command = Mock()
self.async_send_command = AsyncMock()
def __getattr__(self, name):
"""Get attribute from `_values` if not explicitly set."""
return self._values.get(name)
def __setattr__(self, name, value):
"""Set attributes in `_values` if not one of the known attributes."""
if name in ("power_status", "status"):
self._values[name] = value
self._update()
else:
super().__setattr__(name, value)
def _set_update_callback(self, update):
self._update = update
def assert_key_press_release(fnc, count=0, *, dst, key):
"""Assert that correct KeyPressCommand & KeyReleaseCommand where sent."""
assert fnc.call_count >= count * 2 + 1
press_arg = fnc.call_args_list[count * 2].args[0]
release_arg = fnc.call_args_list[count * 2 + 1].args[0]
assert isinstance(press_arg, KeyPressCommand)
assert press_arg.key == key
assert press_arg.dst == dst
assert isinstance(release_arg, KeyReleaseCommand)
assert release_arg.dst == dst

View File

@ -0,0 +1,59 @@
"""Tests for the HDMI-CEC component."""
from unittest.mock import patch
import pytest
from homeassistant.components.hdmi_cec import DOMAIN
from homeassistant.const import EVENT_HOMEASSISTANT_START
from homeassistant.setup import async_setup_component
@pytest.fixture(name="mock_cec_adapter", autouse=True)
def mock_cec_adapter_fixture():
"""Mock CecAdapter.
Always mocked as it imports the `cec` library which is part of `libcec`.
"""
with patch(
"homeassistant.components.hdmi_cec.CecAdapter", autospec=True
) as mock_cec_adapter:
yield mock_cec_adapter
@pytest.fixture(name="mock_hdmi_network")
def mock_hdmi_network_fixture():
"""Mock HDMINetwork."""
with patch(
"homeassistant.components.hdmi_cec.HDMINetwork", autospec=True
) as mock_hdmi_network:
yield mock_hdmi_network
@pytest.fixture
def create_hdmi_network(hass, mock_hdmi_network):
"""Create an initialized mock hdmi_network."""
async def hdmi_network(config=None):
if not config:
config = {}
await async_setup_component(hass, DOMAIN, {DOMAIN: config})
mock_hdmi_network_instance = mock_hdmi_network.return_value
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
return mock_hdmi_network_instance
return hdmi_network
@pytest.fixture
def create_cec_entity(hass):
"""Create a CecEntity."""
async def cec_entity(hdmi_network, device):
new_device_callback = hdmi_network.set_new_device_callback.call_args.args[0]
new_device_callback(device)
await hass.async_block_till_done()
return cec_entity

View File

@ -0,0 +1,469 @@
"""Tests for the HDMI-CEC component."""
from datetime import timedelta
from unittest.mock import ANY, PropertyMock, call, patch
import pytest
import voluptuous as vol
from homeassistant.components.hdmi_cec import (
DOMAIN,
EVENT_HDMI_CEC_UNAVAILABLE,
SERVICE_POWER_ON,
SERVICE_SELECT_DEVICE,
SERVICE_SEND_COMMAND,
SERVICE_STANDBY,
SERVICE_UPDATE_DEVICES,
SERVICE_VOLUME,
WATCHDOG_INTERVAL,
CecCommand,
KeyPressCommand,
KeyReleaseCommand,
PhysicalAddress,
parse_mapping,
)
from homeassistant.const import EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow
from . import assert_key_press_release
from tests.common import (
MockEntity,
MockEntityPlatform,
async_capture_events,
async_fire_time_changed,
)
@pytest.fixture(name="mock_tcp_adapter")
def mock_tcp_adapter_fixture():
"""Mock TcpAdapter."""
with patch(
"homeassistant.components.hdmi_cec.TcpAdapter", autospec=True
) as mock_tcp_adapter:
yield mock_tcp_adapter
@pytest.mark.parametrize(
"mapping,expected",
[
({}, []),
(
{
"TV": "0.0.0.0",
"Pi Zero": "1.0.0.0",
"Fire TV Stick": "2.1.0.0",
"Chromecast": "2.2.0.0",
"Another Device": "2.3.0.0",
"BlueRay player": "3.0.0.0",
},
[
("TV", "0.0.0.0"),
("Pi Zero", "1.0.0.0"),
("Fire TV Stick", "2.1.0.0"),
("Chromecast", "2.2.0.0"),
("Another Device", "2.3.0.0"),
("BlueRay player", "3.0.0.0"),
],
),
(
{
1: "Pi Zero",
2: {
1: "Fire TV Stick",
2: "Chromecast",
3: "Another Device",
},
3: "BlueRay player",
},
[
("Pi Zero", [1, 0, 0, 0]),
("Fire TV Stick", [2, 1, 0, 0]),
("Chromecast", [2, 2, 0, 0]),
("Another Device", [2, 3, 0, 0]),
("BlueRay player", [3, 0, 0, 0]),
],
),
],
)
def test_parse_mapping_physical_address(mapping, expected):
"""Test the device config mapping function."""
result = parse_mapping(mapping)
result = [
(r[0], str(r[1]) if isinstance(r[1], PhysicalAddress) else r[1]) for r in result
]
assert result == expected
# Test Setup
async def test_setup_cec_adapter(hass, mock_cec_adapter, mock_hdmi_network):
"""Test the general setup of this component."""
await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
mock_cec_adapter.assert_called_once_with(name="HA", activate_source=False)
mock_hdmi_network.assert_called_once()
call_args = mock_hdmi_network.call_args
assert call_args == call(mock_cec_adapter.return_value, loop=ANY)
assert call_args.kwargs["loop"] in (None, hass.loop)
mock_hdmi_network_instance = mock_hdmi_network.return_value
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
mock_hdmi_network_instance.start.assert_called_once_with()
mock_hdmi_network_instance.set_new_device_callback.assert_called_once()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
mock_hdmi_network_instance.stop.assert_called_once_with()
@pytest.mark.parametrize("osd_name", ["test", "test_a_long_name"])
async def test_setup_set_osd_name(hass, osd_name, mock_cec_adapter):
"""Test the setup of this component with the `osd_name` config setting."""
await async_setup_component(hass, DOMAIN, {DOMAIN: {"osd_name": osd_name}})
mock_cec_adapter.assert_called_once_with(name=osd_name[:12], activate_source=False)
async def test_setup_tcp_adapter(hass, mock_tcp_adapter, mock_hdmi_network):
"""Test the setup of this component with the TcpAdapter (`host` config setting)."""
host = "0.0.0.0"
await async_setup_component(hass, DOMAIN, {DOMAIN: {"host": host}})
mock_tcp_adapter.assert_called_once_with(host, name="HA", activate_source=False)
mock_hdmi_network.assert_called_once()
call_args = mock_hdmi_network.call_args
assert call_args == call(mock_tcp_adapter.return_value, loop=ANY)
assert call_args.kwargs["loop"] in (None, hass.loop)
mock_hdmi_network_instance = mock_hdmi_network.return_value
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
mock_hdmi_network_instance.start.assert_called_once_with()
mock_hdmi_network_instance.set_new_device_callback.assert_called_once()
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
mock_hdmi_network_instance.stop.assert_called_once_with()
# Test services
async def test_service_power_on(hass, create_hdmi_network):
"""Test the power on service call."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_POWER_ON,
{},
blocking=True,
)
mock_hdmi_network_instance.power_on.assert_called_once_with()
async def test_service_standby(hass, create_hdmi_network):
"""Test the standby service call."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_STANDBY,
{},
blocking=True,
)
mock_hdmi_network_instance.standby.assert_called_once_with()
async def test_service_select_device_alias(hass, create_hdmi_network):
"""Test the select device service call with a known alias."""
mock_hdmi_network_instance = await create_hdmi_network(
{"devices": {"Chromecast": "1.0.0.0"}}
)
await hass.services.async_call(
DOMAIN,
SERVICE_SELECT_DEVICE,
{"device": "Chromecast"},
blocking=True,
)
mock_hdmi_network_instance.active_source.assert_called_once()
physical_address = mock_hdmi_network_instance.active_source.call_args.args[0]
assert isinstance(physical_address, PhysicalAddress)
assert str(physical_address) == "1.0.0.0"
class MockCecEntity(MockEntity):
"""Mock CEC entity."""
@property
def extra_state_attributes(self):
"""Set the physical address in the attributes."""
return {"physical_address": self._values["physical_address"]}
async def test_service_select_device_entity(hass, create_hdmi_network):
"""Test the select device service call with an existing entity."""
platform = MockEntityPlatform(hass)
await platform.async_add_entities(
[MockCecEntity(name="hdmi_3", physical_address="3.0.0.0")]
)
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_SELECT_DEVICE,
{"device": "test_domain.hdmi_3"},
blocking=True,
)
mock_hdmi_network_instance.active_source.assert_called_once()
physical_address = mock_hdmi_network_instance.active_source.call_args.args[0]
assert isinstance(physical_address, PhysicalAddress)
assert str(physical_address) == "3.0.0.0"
async def test_service_select_device_physical_address(hass, create_hdmi_network):
"""Test the select device service call with a raw physical address."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_SELECT_DEVICE,
{"device": "1.1.0.0"},
blocking=True,
)
mock_hdmi_network_instance.active_source.assert_called_once()
physical_address = mock_hdmi_network_instance.active_source.call_args.args[0]
assert isinstance(physical_address, PhysicalAddress)
assert str(physical_address) == "1.1.0.0"
async def test_service_update_devices(hass, create_hdmi_network):
"""Test the update devices service call."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_UPDATE_DEVICES,
{},
blocking=True,
)
mock_hdmi_network_instance.scan.assert_called_once_with()
@pytest.mark.parametrize(
"count,calls",
[
(3, 3),
(1, 1),
(0, 0),
pytest.param(
"",
1,
marks=pytest.mark.xfail(
reason="While the code allows for an empty string the schema doesn't allow it",
raises=vol.MultipleInvalid,
),
),
],
)
@pytest.mark.parametrize("direction,key", [("up", 65), ("down", 66)])
async def test_service_volume_x_times(
hass, create_hdmi_network, count, calls, direction, key
):
"""Test the volume service call with steps."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_VOLUME,
{direction: count},
blocking=True,
)
assert mock_hdmi_network_instance.send_command.call_count == calls * 2
for i in range(calls):
assert_key_press_release(
mock_hdmi_network_instance.send_command, i, dst=5, key=key
)
@pytest.mark.parametrize("direction,key", [("up", 65), ("down", 66)])
async def test_service_volume_press(hass, create_hdmi_network, direction, key):
"""Test the volume service call with press attribute."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_VOLUME,
{direction: "press"},
blocking=True,
)
mock_hdmi_network_instance.send_command.assert_called_once()
arg = mock_hdmi_network_instance.send_command.call_args.args[0]
assert isinstance(arg, KeyPressCommand)
assert arg.key == key
assert arg.dst == 5
@pytest.mark.parametrize("direction,key", [("up", 65), ("down", 66)])
async def test_service_volume_release(hass, create_hdmi_network, direction, key):
"""Test the volume service call with release attribute."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_VOLUME,
{direction: "release"},
blocking=True,
)
mock_hdmi_network_instance.send_command.assert_called_once()
arg = mock_hdmi_network_instance.send_command.call_args.args[0]
assert isinstance(arg, KeyReleaseCommand)
assert arg.dst == 5
@pytest.mark.parametrize(
"attr,key",
[
("toggle", 67),
("on", 101),
("off", 102),
pytest.param(
"",
101,
marks=pytest.mark.xfail(
reason="The documentation mention it's allowed to pass an empty string, but the schema does not allow this",
raises=vol.MultipleInvalid,
),
),
],
)
async def test_service_volume_mute(hass, create_hdmi_network, attr, key):
"""Test the volume service call with mute."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_VOLUME,
{"mute": attr},
blocking=True,
)
assert mock_hdmi_network_instance.send_command.call_count == 2
assert_key_press_release(mock_hdmi_network_instance.send_command, key=key, dst=5)
@pytest.mark.parametrize(
"data,expected",
[
({"raw": "20:0D"}, "20:0d"),
pytest.param(
{"cmd": "36"},
"ff:36",
marks=pytest.mark.xfail(
reason="String is converted in hex value, the final result looks like 'ff:24', not what you'd expect."
),
),
({"cmd": 54}, "ff:36"),
pytest.param(
{"cmd": "36", "src": "1", "dst": "0"},
"10:36",
marks=pytest.mark.xfail(
reason="String is converted in hex value, the final result looks like 'ff:24', not what you'd expect."
),
),
({"cmd": 54, "src": "1", "dst": "0"}, "10:36"),
pytest.param(
{"cmd": "64", "src": "1", "dst": "0", "att": "4f:44"},
"10:64:4f:44",
marks=pytest.mark.xfail(
reason="`att` only accepts a int or a HEX value, it seems good to allow for raw data here.",
raises=vol.MultipleInvalid,
),
),
pytest.param(
{"cmd": "0A", "src": "1", "dst": "0", "att": "1B"},
"10:0a:1b",
marks=pytest.mark.xfail(
reason="The code tries to run `reduce` on this string and fails.",
raises=TypeError,
),
),
pytest.param(
{"cmd": "0A", "src": "1", "dst": "0", "att": "01"},
"10:0a:1b",
marks=pytest.mark.xfail(
reason="The code tries to run `reduce` on this as an `int` and fails.",
raises=TypeError,
),
),
pytest.param(
{"cmd": "0A", "src": "1", "dst": "0", "att": ["1B", "44"]},
"10:0a:1b:44",
marks=pytest.mark.xfail(
reason="While the code shows that it's possible to passthrough a list, the call schema does not allow it.",
raises=(vol.MultipleInvalid, TypeError),
),
),
],
)
async def test_service_send_command(hass, create_hdmi_network, data, expected):
"""Test the send command service call."""
mock_hdmi_network_instance = await create_hdmi_network()
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_COMMAND,
data,
blocking=True,
)
mock_hdmi_network_instance.send_command.assert_called_once()
command = mock_hdmi_network_instance.send_command.call_args.args[0]
assert isinstance(command, CecCommand)
assert str(command) == expected
@pytest.mark.parametrize(
"adapter_initialized_value, watchdog_actions", [(False, 1), (True, 0)]
)
async def test_watchdog(
hass,
create_hdmi_network,
mock_cec_adapter,
adapter_initialized_value,
watchdog_actions,
):
"""Test the watchdog when adapter is down/up."""
adapter_initialized = PropertyMock(return_value=adapter_initialized_value)
events = async_capture_events(hass, EVENT_HDMI_CEC_UNAVAILABLE)
mock_cec_adapter_instance = mock_cec_adapter.return_value
type(mock_cec_adapter_instance).initialized = adapter_initialized
mock_hdmi_network_instance = await create_hdmi_network()
mock_hdmi_network_instance.set_initialized_callback.assert_called_once()
callback = mock_hdmi_network_instance.set_initialized_callback.call_args.args[0]
callback()
async_fire_time_changed(hass, utcnow() + timedelta(seconds=WATCHDOG_INTERVAL))
await hass.async_block_till_done()
adapter_initialized.assert_called_once_with()
assert len(events) == watchdog_actions
assert mock_cec_adapter_instance.init.call_count == watchdog_actions

View File

@ -0,0 +1,493 @@
"""Tests for the HDMI-CEC media player platform."""
from pycec.const import (
DEVICE_TYPE_NAMES,
KEY_BACKWARD,
KEY_FORWARD,
KEY_MUTE_TOGGLE,
KEY_PAUSE,
KEY_PLAY,
KEY_STOP,
KEY_VOLUME_DOWN,
KEY_VOLUME_UP,
POWER_OFF,
POWER_ON,
STATUS_PLAY,
STATUS_STILL,
STATUS_STOP,
TYPE_AUDIO,
TYPE_PLAYBACK,
TYPE_RECORDER,
TYPE_TUNER,
TYPE_TV,
TYPE_UNKNOWN,
)
import pytest
from homeassistant.components.hdmi_cec import EVENT_HDMI_CEC_UNAVAILABLE
from homeassistant.components.media_player import (
DOMAIN as MEDIA_PLAYER_DOMAIN,
MediaPlayerEntityFeature as MPEF,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
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_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_UP,
STATE_IDLE,
STATE_OFF,
STATE_ON,
STATE_PAUSED,
STATE_PLAYING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from . import MockHDMIDevice, assert_key_press_release
@pytest.fixture(
name="assert_state",
params=[
False,
pytest.param(
True,
marks=pytest.mark.xfail(
reason="""State isn't updated because the function is missing the
`schedule_update_ha_state` for a correct push entity. Would still
update once the data comes back from the device."""
),
),
],
ids=["skip_assert_state", "run_assert_state"],
)
def assert_state_fixture(hass, request):
"""Allow for skipping the assert state changes.
This is broken in this entity, but we still want to test that
the rest of the code works as expected.
"""
def test_state(state, expected):
if request.param:
assert state == expected
else:
assert True
return test_state
async def test_load_platform(hass, create_hdmi_network, create_cec_entity):
"""Test that media_player entity is loaded."""
hdmi_network = await create_hdmi_network(config={"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
mock_hdmi_device.set_update_callback.assert_called_once()
state = hass.states.get("media_player.hdmi_3")
assert state is not None
state = hass.states.get("switch.hdmi_3")
assert state is None
@pytest.mark.parametrize("platform", [{}, {"platform": "switch"}])
async def test_load_types(hass, create_hdmi_network, create_cec_entity, platform):
"""Test that media_player entity is loaded when types is set."""
config = platform | {"types": {"hdmi_cec.hdmi_4": "media_player"}}
hdmi_network = await create_hdmi_network(config=config)
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
mock_hdmi_device.set_update_callback.assert_called_once()
state = hass.states.get("media_player.hdmi_3")
assert state is None
state = hass.states.get("switch.hdmi_3")
assert state is not None
mock_hdmi_device = MockHDMIDevice(logical_address=4)
await create_cec_entity(hdmi_network, mock_hdmi_device)
mock_hdmi_device.set_update_callback.assert_called_once()
state = hass.states.get("media_player.hdmi_4")
assert state is not None
state = hass.states.get("switch.hdmi_4")
assert state is None
async def test_service_on(hass, create_hdmi_network, create_cec_entity, assert_state):
"""Test that media_player triggers on `on` service."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("media_player.hdmi_3")
assert state.state != STATE_ON
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "media_player.hdmi_3"},
blocking=True,
)
await hass.async_block_till_done()
mock_hdmi_device.turn_on.assert_called_once_with()
state = hass.states.get("media_player.hdmi_3")
assert_state(state.state, STATE_ON)
async def test_service_off(hass, create_hdmi_network, create_cec_entity, assert_state):
"""Test that media_player triggers on `off` service."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("media_player.hdmi_3")
assert state.state != STATE_OFF
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "media_player.hdmi_3"},
blocking=True,
)
mock_hdmi_device.turn_off.assert_called_once_with()
state = hass.states.get("media_player.hdmi_3")
assert_state(state.state, STATE_OFF)
@pytest.mark.parametrize(
"type_id,expected_features",
[
(TYPE_TV, (MPEF.TURN_ON, MPEF.TURN_OFF)),
(
TYPE_RECORDER,
(
MPEF.TURN_ON,
MPEF.TURN_OFF,
MPEF.PAUSE,
MPEF.STOP,
MPEF.PREVIOUS_TRACK,
MPEF.NEXT_TRACK,
),
),
pytest.param(
TYPE_RECORDER,
(MPEF.PLAY,),
marks=pytest.mark.xfail(
reason="The feature is wrongly set to PLAY_MEDIA, but should be PLAY."
),
),
(TYPE_UNKNOWN, (MPEF.TURN_ON, MPEF.TURN_OFF)),
pytest.param(
TYPE_TUNER,
(
MPEF.TURN_ON,
MPEF.TURN_OFF,
MPEF.PAUSE,
MPEF.STOP,
),
marks=pytest.mark.xfail(
reason="Checking for the wrong attribute, should be checking `type_id`, is checking `type`."
),
),
pytest.param(
TYPE_TUNER,
(MPEF.PLAY,),
marks=pytest.mark.xfail(
reason="The feature is wrongly set to PLAY_MEDIA, but should be PLAY."
),
),
pytest.param(
TYPE_PLAYBACK,
(
MPEF.TURN_ON,
MPEF.TURN_OFF,
MPEF.PAUSE,
MPEF.STOP,
MPEF.PREVIOUS_TRACK,
MPEF.NEXT_TRACK,
),
marks=pytest.mark.xfail(
reason="Checking for the wrong attribute, should be checking `type_id`, is checking `type`."
),
),
pytest.param(
TYPE_PLAYBACK,
(MPEF.PLAY,),
marks=pytest.mark.xfail(
reason="The feature is wrongly set to PLAY_MEDIA, but should be PLAY."
),
),
(
TYPE_AUDIO,
(
MPEF.TURN_ON,
MPEF.TURN_OFF,
MPEF.VOLUME_STEP,
MPEF.VOLUME_MUTE,
),
),
],
)
async def test_supported_features(
hass, create_hdmi_network, create_cec_entity, type_id, expected_features
):
"""Test that features load as expected."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(
logical_address=3, type=type_id, type_name=DEVICE_TYPE_NAMES[type_id]
)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("media_player.hdmi_3")
supported_features = state.attributes["supported_features"]
for feature in expected_features:
assert supported_features & feature
@pytest.mark.parametrize(
"service,extra_data,key",
[
(SERVICE_VOLUME_DOWN, None, KEY_VOLUME_DOWN),
(SERVICE_VOLUME_UP, None, KEY_VOLUME_UP),
(SERVICE_VOLUME_MUTE, {"is_volume_muted": True}, KEY_MUTE_TOGGLE),
(SERVICE_VOLUME_MUTE, {"is_volume_muted": False}, KEY_MUTE_TOGGLE),
],
)
async def test_volume_services(
hass, create_hdmi_network, create_cec_entity, service, extra_data, key
):
"""Test volume related commands."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3, type=TYPE_AUDIO)
await create_cec_entity(hdmi_network, mock_hdmi_device)
data = {ATTR_ENTITY_ID: "media_player.hdmi_3"}
if extra_data:
data |= extra_data
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
data,
blocking=True,
)
await hass.async_block_till_done()
assert mock_hdmi_device.send_command.call_count == 2
assert_key_press_release(mock_hdmi_device.send_command, dst=3, key=key)
@pytest.mark.parametrize(
"service,key",
[
(SERVICE_MEDIA_NEXT_TRACK, KEY_FORWARD),
(SERVICE_MEDIA_PREVIOUS_TRACK, KEY_BACKWARD),
],
)
async def test_track_change_services(
hass, create_hdmi_network, create_cec_entity, service, key
):
"""Test track change related commands."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3, type=TYPE_RECORDER)
await create_cec_entity(hdmi_network, mock_hdmi_device)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: "media_player.hdmi_3"},
blocking=True,
)
await hass.async_block_till_done()
assert mock_hdmi_device.send_command.call_count == 2
assert_key_press_release(mock_hdmi_device.send_command, dst=3, key=key)
@pytest.mark.parametrize(
"service,key,expected_state",
[
pytest.param(
SERVICE_MEDIA_PLAY,
KEY_PLAY,
STATE_PLAYING,
marks=pytest.mark.xfail(
reason="The wrong feature is defined, should be PLAY, not PLAY_MEDIA"
),
),
(SERVICE_MEDIA_PAUSE, KEY_PAUSE, STATE_PAUSED),
(SERVICE_MEDIA_STOP, KEY_STOP, STATE_IDLE),
],
)
async def test_playback_services(
hass,
create_hdmi_network,
create_cec_entity,
assert_state,
service,
key,
expected_state,
):
"""Test playback related commands."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3, type=TYPE_RECORDER)
await create_cec_entity(hdmi_network, mock_hdmi_device)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: "media_player.hdmi_3"},
blocking=True,
)
await hass.async_block_till_done()
assert mock_hdmi_device.send_command.call_count == 2
assert_key_press_release(mock_hdmi_device.send_command, dst=3, key=key)
state = hass.states.get("media_player.hdmi_3")
assert_state(state.state, expected_state)
@pytest.mark.xfail(reason="PLAY feature isn't enabled")
async def test_play_pause_service(
hass,
create_hdmi_network,
create_cec_entity,
assert_state,
):
"""Test play pause service."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(
logical_address=3, type=TYPE_RECORDER, status=STATUS_PLAY
)
await create_cec_entity(hdmi_network, mock_hdmi_device)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PLAY_PAUSE,
{ATTR_ENTITY_ID: "media_player.hdmi_3"},
blocking=True,
)
await hass.async_block_till_done()
assert mock_hdmi_device.send_command.call_count == 2
assert_key_press_release(mock_hdmi_device.send_command, dst=3, key=KEY_PAUSE)
state = hass.states.get("media_player.hdmi_3")
assert_state(state.state, STATE_PAUSED)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PLAY_PAUSE,
{ATTR_ENTITY_ID: "media_player.hdmi_3"},
blocking=True,
)
await hass.async_block_till_done()
assert mock_hdmi_device.send_command.call_count == 4
assert_key_press_release(mock_hdmi_device.send_command, 1, dst=3, key=KEY_PLAY)
@pytest.mark.parametrize(
"type_id,update_data,expected_state",
[
(TYPE_TV, {"power_status": POWER_OFF}, STATE_OFF),
(TYPE_TV, {"power_status": 3}, STATE_OFF),
(TYPE_TV, {"power_status": POWER_ON}, STATE_ON),
(TYPE_TV, {"power_status": 4}, STATE_ON),
(TYPE_TV, {"power_status": POWER_ON, "status": STATUS_PLAY}, STATE_ON),
(TYPE_RECORDER, {"power_status": POWER_OFF, "status": STATUS_PLAY}, STATE_OFF),
(
TYPE_RECORDER,
{"power_status": POWER_ON, "status": STATUS_PLAY},
STATE_PLAYING,
),
(TYPE_RECORDER, {"power_status": POWER_ON, "status": STATUS_STOP}, STATE_IDLE),
(
TYPE_RECORDER,
{"power_status": POWER_ON, "status": STATUS_STILL},
STATE_PAUSED,
),
(TYPE_RECORDER, {"power_status": POWER_ON, "status": None}, STATE_UNKNOWN),
],
)
async def test_update_state(
hass, create_hdmi_network, create_cec_entity, type_id, update_data, expected_state
):
"""Test state updates work as expected."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3, type=type_id)
await create_cec_entity(hdmi_network, mock_hdmi_device)
for att, val in update_data.items():
setattr(mock_hdmi_device, att, val)
await hass.async_block_till_done()
state = hass.states.get("media_player.hdmi_3")
assert state.state == expected_state
@pytest.mark.parametrize(
"data,expected_state",
[
({"power_status": POWER_OFF}, STATE_OFF),
({"power_status": 3}, STATE_OFF),
({"power_status": POWER_ON, "type": TYPE_TV}, STATE_ON),
({"power_status": 4, "type": TYPE_TV}, STATE_ON),
({"power_status": POWER_ON, "type": TYPE_TV, "status": STATUS_PLAY}, STATE_ON),
(
{"power_status": POWER_OFF, "type": TYPE_RECORDER, "status": STATUS_PLAY},
STATE_OFF,
),
(
{"power_status": POWER_ON, "type": TYPE_RECORDER, "status": STATUS_PLAY},
STATE_PLAYING,
),
(
{"power_status": POWER_ON, "type": TYPE_RECORDER, "status": STATUS_STOP},
STATE_IDLE,
),
(
{"power_status": POWER_ON, "type": TYPE_RECORDER, "status": STATUS_STILL},
STATE_PAUSED,
),
(
{"power_status": POWER_ON, "type": TYPE_RECORDER, "status": None},
STATE_UNKNOWN,
),
],
)
async def test_starting_state(
hass, create_hdmi_network, create_cec_entity, data, expected_state
):
"""Test starting states are set as expected."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3, **data)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("media_player.hdmi_3")
assert state.state == expected_state
@pytest.mark.xfail(
reason="The code only sets the state to unavailable, doesn't set the `_attr_available` to false."
)
async def test_unavailable_status(hass, create_hdmi_network, create_cec_entity):
"""Test entity goes into unavailable status when expected."""
hdmi_network = await create_hdmi_network({"platform": "media_player"})
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
hass.bus.async_fire(EVENT_HDMI_CEC_UNAVAILABLE)
state = hass.states.get("media_player.hdmi_3")
assert state.state == STATE_UNAVAILABLE

View File

@ -0,0 +1,252 @@
"""Tests for the HDMI-CEC switch platform."""
import pytest
from homeassistant.components.hdmi_cec import (
EVENT_HDMI_CEC_UNAVAILABLE,
POWER_OFF,
POWER_ON,
STATUS_PLAY,
STATUS_STILL,
STATUS_STOP,
PhysicalAddress,
)
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from tests.components.hdmi_cec import MockHDMIDevice
@pytest.mark.parametrize("config", [{}, {"platform": "switch"}])
async def test_load_platform(hass, create_hdmi_network, create_cec_entity, config):
"""Test that switch entity is loaded."""
hdmi_network = await create_hdmi_network(config=config)
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
mock_hdmi_device.set_update_callback.assert_called_once()
state = hass.states.get("media_player.hdmi_3")
assert state is None
state = hass.states.get("switch.hdmi_3")
assert state is not None
async def test_load_types(hass, create_hdmi_network, create_cec_entity):
"""Test that switch entity is loaded when types is set."""
config = {"platform": "media_player", "types": {"hdmi_cec.hdmi_3": "switch"}}
hdmi_network = await create_hdmi_network(config=config)
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
mock_hdmi_device.set_update_callback.assert_called_once()
state = hass.states.get("media_player.hdmi_3")
assert state is None
state = hass.states.get("switch.hdmi_3")
assert state is not None
mock_hdmi_device = MockHDMIDevice(logical_address=4)
await create_cec_entity(hdmi_network, mock_hdmi_device)
mock_hdmi_device.set_update_callback.assert_called_once()
state = hass.states.get("media_player.hdmi_4")
assert state is not None
state = hass.states.get("switch.hdmi_4")
assert state is None
async def test_service_on(hass, create_hdmi_network, create_cec_entity):
"""Test that switch triggers on `on` service."""
hdmi_network = await create_hdmi_network()
mock_hdmi_device = MockHDMIDevice(logical_address=3, power_status=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("switch.hdmi_3")
assert state.state != STATE_ON
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: "switch.hdmi_3"}, blocking=True
)
mock_hdmi_device.turn_on.assert_called_once_with()
state = hass.states.get("switch.hdmi_3")
assert state.state == STATE_ON
async def test_service_off(hass, create_hdmi_network, create_cec_entity):
"""Test that switch triggers on `off` service."""
hdmi_network = await create_hdmi_network()
mock_hdmi_device = MockHDMIDevice(logical_address=3, power_status=4)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("switch.hdmi_3")
assert state.state != STATE_OFF
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.hdmi_3"},
blocking=True,
)
mock_hdmi_device.turn_off.assert_called_once_with()
state = hass.states.get("switch.hdmi_3")
assert state.state == STATE_OFF
@pytest.mark.parametrize(
"power_status,expected_state",
[(3, STATE_OFF), (POWER_OFF, STATE_OFF), (4, STATE_ON), (POWER_ON, STATE_ON)],
)
@pytest.mark.parametrize(
"status",
[
None,
STATUS_PLAY,
STATUS_STOP,
STATUS_STILL,
],
)
async def test_device_status_change(
hass, create_hdmi_network, create_cec_entity, power_status, expected_state, status
):
"""Test state change on device status change."""
hdmi_network = await create_hdmi_network()
mock_hdmi_device = MockHDMIDevice(logical_address=3, status=status)
await create_cec_entity(hdmi_network, mock_hdmi_device)
mock_hdmi_device.power_status = power_status
await hass.async_block_till_done()
state = hass.states.get("switch.hdmi_3")
if power_status in (POWER_ON, 4) and status is not None:
pytest.xfail(
reason="`CecSwitchEntity.is_on` returns `False` here instead of `true` as expected."
)
assert state.state == expected_state
@pytest.mark.parametrize(
"device_values, expected",
[
({"osd_name": "Switch", "vendor": "Nintendo"}, "Nintendo Switch"),
({"type_name": "TV"}, "TV 3"),
({"type_name": "Playback", "osd_name": "Switch"}, "Playback 3 (Switch)"),
({"type_name": "TV", "vendor": "Samsung"}, "TV 3"),
(
{"type_name": "Playback", "osd_name": "Super PC", "vendor": "Unknown"},
"Playback 3 (Super PC)",
),
],
)
async def test_friendly_name(
hass, create_hdmi_network, create_cec_entity, device_values, expected
):
"""Test friendly name setup."""
hdmi_network = await create_hdmi_network()
mock_hdmi_device = MockHDMIDevice(logical_address=3, **device_values)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("switch.hdmi_3")
assert state.attributes["friendly_name"] == expected
@pytest.mark.parametrize(
"device_values,expected_attributes",
[
(
{"physical_address": PhysicalAddress("3.0.0.0")},
{"physical_address": "3.0.0.0"},
),
pytest.param(
{},
{},
marks=pytest.mark.xfail(
reason="physical address logic returns a string 'None' instead of not being set."
),
),
(
{"physical_address": PhysicalAddress("3.0.0.0"), "vendor_id": 5},
{"physical_address": "3.0.0.0", "vendor_id": 5, "vendor_name": None},
),
(
{
"physical_address": PhysicalAddress("3.0.0.0"),
"vendor_id": 5,
"vendor": "Samsung",
},
{"physical_address": "3.0.0.0", "vendor_id": 5, "vendor_name": "Samsung"},
),
(
{"physical_address": PhysicalAddress("3.0.0.0"), "type": 1},
{"physical_address": "3.0.0.0", "type_id": 1, "type": None},
),
(
{
"physical_address": PhysicalAddress("3.0.0.0"),
"type": 1,
"type_name": "TV",
},
{"physical_address": "3.0.0.0", "type_id": 1, "type": "TV"},
),
],
)
async def test_extra_state_attributes(
hass, create_hdmi_network, create_cec_entity, device_values, expected_attributes
):
"""Test extra state attributes."""
hdmi_network = await create_hdmi_network()
mock_hdmi_device = MockHDMIDevice(logical_address=3, **device_values)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("switch.hdmi_3")
attributes = state.attributes
# We don't care about these attributes, so just copy them to the expected attributes
for att in ("friendly_name", "icon"):
expected_attributes[att] = attributes[att]
assert attributes == expected_attributes
@pytest.mark.parametrize(
"device_type,expected_icon",
[
(None, "mdi:help"),
(0, "mdi:television"),
(1, "mdi:microphone"),
(2, "mdi:help"),
(3, "mdi:radio"),
(4, "mdi:play"),
(5, "mdi:speaker"),
],
)
async def test_icon(
hass, create_hdmi_network, create_cec_entity, device_type, expected_icon
):
"""Test icon selection."""
hdmi_network = await create_hdmi_network()
mock_hdmi_device = MockHDMIDevice(logical_address=3, type=device_type)
await create_cec_entity(hdmi_network, mock_hdmi_device)
state = hass.states.get("switch.hdmi_3")
assert state.attributes["icon"] == expected_icon
@pytest.mark.xfail(
reason="The code only sets the state to unavailable, doesn't set the `_attr_available` to false."
)
async def test_unavailable_status(hass, create_hdmi_network, create_cec_entity):
"""Test entity goes into unavailable status when expected."""
hdmi_network = await create_hdmi_network()
mock_hdmi_device = MockHDMIDevice(logical_address=3)
await create_cec_entity(hdmi_network, mock_hdmi_device)
hass.bus.async_fire(EVENT_HDMI_CEC_UNAVAILABLE)
await hass.async_block_till_done()
state = hass.states.get("switch.hdmi_3")
assert state.state == STATE_UNAVAILABLE