mirror of
https://github.com/home-assistant/core.git
synced 2026-05-18 14:29:57 +01:00
renovate[bot]
b0c2e57649
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> Co-authored-by: abmantis <amfcalt@gmail.com>
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
"""Tests for Broadlink infrared platform."""
|
|
|
|
from unittest.mock import call
|
|
|
|
from broadlink.exceptions import BroadlinkException
|
|
from broadlink.remote import pulses_to_data
|
|
from infrared_protocols.commands.nec import NECCommand
|
|
import pytest
|
|
|
|
from homeassistant.components.broadlink.const import DOMAIN
|
|
from homeassistant.components.infrared import async_send_command
|
|
from homeassistant.const import Platform
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
|
|
|
from . import get_device
|
|
|
|
IR_DEVICES = ["Entrance", "Living Room", "Office", "Garage"]
|
|
NON_IR_DEVICE = "Bedroom"
|
|
|
|
|
|
async def test_infrared_setup_works(
|
|
hass: HomeAssistant,
|
|
device_registry: dr.DeviceRegistry,
|
|
entity_registry: er.EntityRegistry,
|
|
) -> None:
|
|
"""Test the infrared entity is created for all IR-capable devices."""
|
|
for device in map(get_device, IR_DEVICES):
|
|
mock_setup = await device.setup_entry(hass)
|
|
|
|
device_entry = device_registry.async_get_device(
|
|
identifiers={(DOMAIN, mock_setup.entry.unique_id)}
|
|
)
|
|
entries = er.async_entries_for_device(entity_registry, device_entry.id)
|
|
infrared_entities = [
|
|
entry for entry in entries if entry.domain == Platform.INFRARED
|
|
]
|
|
assert len(infrared_entities) == 1
|
|
assert infrared_entities[0].unique_id == f"{device.mac}-emitter"
|
|
|
|
|
|
async def test_infrared_not_created_for_non_ir_device(
|
|
hass: HomeAssistant,
|
|
entity_registry: er.EntityRegistry,
|
|
) -> None:
|
|
"""Test no infrared entity is created for non-IR devices."""
|
|
device = get_device(NON_IR_DEVICE)
|
|
mock_setup = await device.setup_entry(hass)
|
|
|
|
entries = er.async_entries_for_config_entry(
|
|
entity_registry, mock_setup.entry.entry_id
|
|
)
|
|
infrared_entities = [
|
|
entry for entry in entries if entry.domain == Platform.INFRARED
|
|
]
|
|
assert len(infrared_entities) == 0
|
|
|
|
|
|
async def test_infrared_send_command(
|
|
hass: HomeAssistant,
|
|
entity_registry: er.EntityRegistry,
|
|
) -> None:
|
|
"""Test sending an IR command dispatches to the Broadlink API."""
|
|
device = get_device("Entrance")
|
|
mock_setup = await device.setup_entry(hass)
|
|
|
|
entries = er.async_entries_for_config_entry(
|
|
entity_registry, mock_setup.entry.entry_id
|
|
)
|
|
infrared_entity = next(
|
|
entry for entry in entries if entry.domain == Platform.INFRARED
|
|
)
|
|
|
|
command = NECCommand(address=0x20, command=0x10)
|
|
await async_send_command(hass, infrared_entity.entity_id, command)
|
|
|
|
expected_pulses = [abs(t) for t in command.get_raw_timings()]
|
|
expected_packet = pulses_to_data(expected_pulses)
|
|
|
|
assert mock_setup.api.send_data.call_count == 1
|
|
assert mock_setup.api.send_data.call_args == call(expected_packet)
|
|
|
|
|
|
@pytest.mark.parametrize("error", [BroadlinkException("boom"), OSError("boom")])
|
|
async def test_infrared_send_command_error_translates(
|
|
hass: HomeAssistant,
|
|
entity_registry: er.EntityRegistry,
|
|
error: Exception,
|
|
) -> None:
|
|
"""Test that Broadlink API errors translate to HomeAssistantError."""
|
|
device = get_device("Entrance")
|
|
mock_setup = await device.setup_entry(hass)
|
|
mock_setup.api.send_data.side_effect = error
|
|
|
|
entries = er.async_entries_for_config_entry(
|
|
entity_registry, mock_setup.entry.entry_id
|
|
)
|
|
infrared_entity = next(
|
|
entry for entry in entries if entry.domain == Platform.INFRARED
|
|
)
|
|
|
|
command = NECCommand(address=0x20, command=0x10)
|
|
with pytest.raises(HomeAssistantError) as exc_info:
|
|
await async_send_command(hass, infrared_entity.entity_id, command)
|
|
|
|
assert exc_info.value.translation_key == "send_command_failed"
|
|
assert exc_info.value.translation_domain == DOMAIN
|