ha-ninlight/tests/test_sacn.py

"""The sACN/E1.31 settings and status entities."""

from __future__ import annotations

from datetime import timedelta
from typing import Any
from unittest.mock import MagicMock

from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.number import (
    ATTR_MAX,
    ATTR_VALUE,
    SERVICE_SET_VALUE,
)
from homeassistant.components.number import (
    DOMAIN as NUMBER_DOMAIN,
)
from homeassistant.components.select import (
    ATTR_OPTION,
    SERVICE_SELECT_OPTION,
)
from homeassistant.components.select import (
    DOMAIN as SELECT_DOMAIN,
)
from homeassistant.components.switch import (
    DOMAIN as SWITCH_DOMAIN,
)
from homeassistant.components.switch import (
    SERVICE_TURN_OFF,
    SERVICE_TURN_ON,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
import pytest
from pytest_homeassistant_custom_component.common import async_fire_time_changed

from tests.helpers import entity_id_for


async def poll(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
    """Let one poll cycle pass so the entities see the lamp's changed answers."""
    freezer.tick(timedelta(seconds=31))
    async_fire_time_changed(hass)
    await hass.async_block_till_done()


async def test_switch_reads_and_writes(
    hass: HomeAssistant, init_integration, mock_client: MagicMock
) -> None:
    entity_id = entity_id_for(hass, "switch", "sacn_enabled")
    assert hass.states.get(entity_id).state == STATE_ON
    await hass.services.async_call(
        SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: entity_id}, blocking=True
    )
    mock_client.set_sacn.assert_awaited_once_with(enabled=False)
    assert hass.states.get(entity_id).state == STATE_OFF
    await hass.services.async_call(
        SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id}, blocking=True
    )
    assert hass.states.get(entity_id).state == STATE_ON


@pytest.mark.parametrize(
    ("key", "field", "value"),
    [("sacn_universe", "universe", 12), ("sacn_start_slot", "start_slot", 7)],
)
async def test_numbers_read_and_write(
    hass: HomeAssistant,
    init_integration,
    mock_client: MagicMock,
    key: str,
    field: str,
    value: int,
) -> None:
    entity_id = entity_id_for(hass, "number", key)
    assert hass.states.get(entity_id).state == "1.0"
    await hass.services.async_call(
        NUMBER_DOMAIN,
        SERVICE_SET_VALUE,
        {ATTR_ENTITY_ID: entity_id, ATTR_VALUE: value},
        blocking=True,
    )
    mock_client.set_sacn.assert_awaited_once_with(**{field: value})
    assert hass.states.get(entity_id).state == f"{float(value)}"


async def test_start_slot_limit_follows_the_lamp(
    hass: HomeAssistant,
    init_integration,
    lamp_settings: dict[str, Any],
    freezer: FrozenDateTimeFactory,
) -> None:
    """The highest usable start slot depends on how many slots the lamp occupies, and that
    number changes with the slot layout - so it is read from the lamp, not hard-coded."""
    entity_id = entity_id_for(hass, "number", "sacn_start_slot")
    assert hass.states.get(entity_id).attributes[ATTR_MAX] == 507  # 6 slots

    lamp_settings["sacn"]["slots"] = 3
    lamp_settings["sacn"]["layout"] = "rgb"
    await poll(hass, freezer)
    assert hass.states.get(entity_id).attributes[ATTR_MAX] == 510


@pytest.mark.parametrize(
    ("key", "field", "option"),
    [("sacn_on_timeout", "on_timeout", "hold"), ("sacn_layout", "layout", "rgb")],
)
async def test_selects_read_and_write(
    hass: HomeAssistant,
    init_integration,
    mock_client: MagicMock,
    key: str,
    field: str,
    option: str,
) -> None:
    entity_id = entity_id_for(hass, "select", key)
    await hass.services.async_call(
        SELECT_DOMAIN,
        SERVICE_SELECT_OPTION,
        {ATTR_ENTITY_ID: entity_id, ATTR_OPTION: option},
        blocking=True,
    )
    mock_client.set_sacn.assert_awaited_once_with(**{field: option})
    assert hass.states.get(entity_id).state == option


async def test_status_flags_come_from_system_info(
    hass: HomeAssistant,
    init_integration,
    lamp_info: dict[str, Any],
    freezer: FrozenDateTimeFactory,
) -> None:
    entity_id = entity_id_for(hass, "binary_sensor", "sacn_active")
    assert hass.states.get(entity_id).state == STATE_OFF

    lamp_info["sacn"]["active"] = True
    await poll(hass, freezer)
    assert hass.states.get(entity_id).state == STATE_ON


async def test_multicast_flag_is_off_by_default(hass: HomeAssistant, init_integration) -> None:
    """It reads "joined" in normal operation and only earns attention when multicast does
    not arrive - like the other diagnostics, it is there but not in the way."""
    entry = er.async_get(hass).async_get(entity_id_for(hass, "binary_sensor", "sacn_multicast"))
    assert entry is not None and entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION


@pytest.mark.parametrize(
    ("platform", "key"),
    [
        ("switch", "sacn_enabled"),
        ("number", "sacn_universe"),
        ("number", "sacn_start_slot"),
        ("select", "sacn_on_timeout"),
        ("select", "sacn_layout"),
    ],
)
async def test_unavailable_on_firmware_without_sacn(
    hass: HomeAssistant,
    init_integration,
    lamp_settings: dict[str, Any],
    freezer: FrozenDateTimeFactory,
    platform: str,
    key: str,
) -> None:
    """Older firmware answers system/settings without a "sacn" block. The entities exist -
    so they come to life on the next poll after such a lamp is updated - but they say so
    instead of showing a made-up value."""
    entity_id = entity_id_for(hass, platform, key)
    lamp_settings.pop("sacn")
    await poll(hass, freezer)
    assert hass.states.get(entity_id).state == STATE_UNAVAILABLE