diff --git a/.coveragerc b/.coveragerc index 9d839cdb5f6..0a1430cce91 100644 --- a/.coveragerc +++ b/.coveragerc @@ -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 diff --git a/CODEOWNERS b/CODEOWNERS index 906119e9006..292879aabea 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -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 diff --git a/homeassistant/components/hdmi_cec/__init__.py b/homeassistant/components/hdmi_cec/__init__.py index 056eacb6a5b..fa64b3dac41 100644 --- a/homeassistant/components/hdmi_cec/__init__.py +++ b/homeassistant/components/hdmi_cec/__init__.py @@ -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) diff --git a/homeassistant/components/hdmi_cec/manifest.json b/homeassistant/components/hdmi_cec/manifest.json index 8ea56a51fa9..46e7f61719f 100644 --- a/homeassistant/components/hdmi_cec/manifest.json +++ b/homeassistant/components/hdmi_cec/manifest.json @@ -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"] } diff --git a/requirements_test_all.txt b/requirements_test_all.txt index a9c55603282..0326765cf74 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -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 diff --git a/tests/components/hdmi_cec/__init__.py b/tests/components/hdmi_cec/__init__.py new file mode 100644 index 00000000000..c131bf96b41 --- /dev/null +++ b/tests/components/hdmi_cec/__init__.py @@ -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 diff --git a/tests/components/hdmi_cec/conftest.py b/tests/components/hdmi_cec/conftest.py new file mode 100644 index 00000000000..c5f82c04e19 --- /dev/null +++ b/tests/components/hdmi_cec/conftest.py @@ -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 diff --git a/tests/components/hdmi_cec/test_init.py b/tests/components/hdmi_cec/test_init.py new file mode 100644 index 00000000000..751c9b051f0 --- /dev/null +++ b/tests/components/hdmi_cec/test_init.py @@ -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 diff --git a/tests/components/hdmi_cec/test_media_player.py b/tests/components/hdmi_cec/test_media_player.py new file mode 100644 index 00000000000..861134e2715 --- /dev/null +++ b/tests/components/hdmi_cec/test_media_player.py @@ -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 diff --git a/tests/components/hdmi_cec/test_switch.py b/tests/components/hdmi_cec/test_switch.py new file mode 100644 index 00000000000..61e1e03e4a6 --- /dev/null +++ b/tests/components/hdmi_cec/test_switch.py @@ -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