ha-ninstrip/tests/test_sacn.py

"""sACN: Schalter, Universe, Startslot, Verhalten nach dem Stromende, Merker."""

from __future__ import annotations

from typing import Any
from unittest.mock import MagicMock

from freezegun.api import FrozenDateTimeFactory
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import entity_registry as er
import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry

from tests.helpers import abfragen, entity_id_for


async def test_schalter_zeigt_den_eingestellten_zustand(
    hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
    assert hass.states.get(entity_id_for(hass, "switch", "sacn_enabled")).state == STATE_ON


async def test_schalter_schreibt_und_liest_zurueck(
    hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
    entity_id = entity_id_for(hass, "switch", "sacn_enabled")
    await hass.services.async_call("switch", "turn_off", {ATTR_ENTITY_ID: entity_id}, blocking=True)
    mock_client.set_sacn.assert_awaited_once_with(enabled=False)
    # Nicht geraten, sondern zurueckgelesen: das Geraet antwortet nur mit "ok".
    assert mock_client.get_settings.await_count >= 2
    assert hass.states.get(entity_id).state == STATE_OFF


async def test_universe_und_startslot(
    hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
    universe = entity_id_for(hass, "number", "sacn_universe")
    slot = entity_id_for(hass, "number", "sacn_start_slot")
    assert hass.states.get(universe).state == "1.0"
    assert hass.states.get(slot).state == "1.0"
    await hass.services.async_call(
        "number", "set_value", {ATTR_ENTITY_ID: universe, "value": 7}, blocking=True
    )
    mock_client.set_sacn.assert_awaited_with(universe=7)
    await hass.services.async_call(
        "number", "set_value", {ATTR_ENTITY_ID: slot, "value": 4}, blocking=True
    )
    mock_client.set_sacn.assert_awaited_with(start_slot=4)


async def test_startslot_endet_bei_510(
    hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
    """Hinter 510 passt bei drei Kanaeln kein ganzes Pixel mehr ins Universe."""
    slot = entity_id_for(hass, "number", "sacn_start_slot")
    assert hass.states.get(slot).attributes["max"] == 510
    with pytest.raises(ServiceValidationError):
        await hass.services.async_call(
            "number", "set_value", {ATTR_ENTITY_ID: slot, "value": 511}, blocking=True
        )
    mock_client.set_sacn.assert_not_awaited()


@pytest.mark.parametrize(("kanaele", "grenze"), [(3, 510), (4, 509), (5, 508)])
async def test_die_grenze_folgt_den_kanaelen_je_pixel(
    hass: HomeAssistant,
    init_integration: MockConfigEntry,
    strip_settings: dict[str, Any],
    freezer: FrozenDateTimeFactory,
    kanaele: int,
    grenze: int,
) -> None:
    strip_settings["sacn"]["channels_per_pixel"] = kanaele
    await abfragen(hass, freezer)
    slot = entity_id_for(hass, "number", "sacn_start_slot")
    assert hass.states.get(slot).attributes["max"] == grenze


async def test_unsinnige_kanalzahl_faellt_auf_drei_zurueck(
    hass: HomeAssistant,
    init_integration: MockConfigEntry,
    strip_settings: dict[str, Any],
    freezer: FrozenDateTimeFactory,
) -> None:
    strip_settings["sacn"]["channels_per_pixel"] = "drei"
    await abfragen(hass, freezer)
    slot = entity_id_for(hass, "number", "sacn_start_slot")
    assert hass.states.get(slot).attributes["max"] == 510


async def test_verhalten_nach_dem_stromende(
    hass: HomeAssistant, init_integration: MockConfigEntry, mock_client: MagicMock
) -> None:
    entity_id = entity_id_for(hass, "select", "sacn_on_timeout")
    state = hass.states.get(entity_id)
    assert state.state == "release"
    assert state.attributes["options"] == ["release", "hold"]
    await hass.services.async_call(
        "select", "select_option", {ATTR_ENTITY_ID: entity_id, "option": "hold"}, blocking=True
    )
    mock_client.set_sacn.assert_awaited_once_with(on_timeout="hold")
    assert hass.states.get(entity_id).state == "hold"


async def test_unbekanntes_verhalten_zeigt_keinen_wert(
    hass: HomeAssistant,
    init_integration: MockConfigEntry,
    strip_settings: dict[str, Any],
    freezer: FrozenDateTimeFactory,
) -> None:
    strip_settings["sacn"]["on_timeout"] = "irgendwas"
    await abfragen(hass, freezer)
    assert hass.states.get(entity_id_for(hass, "select", "sacn_on_timeout")).state == "unknown"


async def test_aktiv_merker_folgt_dem_strom(
    hass: HomeAssistant,
    init_integration: MockConfigEntry,
    strip_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
    strip_info["sacn"]["active"] = True
    await abfragen(hass, freezer)
    assert hass.states.get(entity_id).state == STATE_ON


async def test_multicast_merker_ist_von_haus_aus_aus(
    hass: HomeAssistant, init_integration: MockConfigEntry
) -> None:
    eintrag = er.async_get(hass).async_get(entity_id_for(hass, "binary_sensor", "sacn_multicast"))
    assert eintrag.disabled_by is er.RegistryEntryDisabler.INTEGRATION


async def test_firmware_ohne_sacn_laesst_die_entitaeten_unverfuegbar(
    hass: HomeAssistant,
    mock_config_entry: MockConfigEntry,
    mock_client: MagicMock,
    strip_settings: dict[str, Any],
    strip_info: dict[str, Any],
) -> None:
    """Die Entitaeten entstehen trotzdem - nach einem Firmware-Update werden
    sie beim naechsten Abfragen lebendig, ohne die Integration neu zu laden."""
    strip_settings.pop("sacn")
    strip_info.pop("sacn")
    mock_config_entry.add_to_hass(hass)
    assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
    await hass.async_block_till_done()
    for platform, key in (
        ("switch", "sacn_enabled"),
        ("number", "sacn_universe"),
        ("number", "sacn_start_slot"),
        ("select", "sacn_on_timeout"),
        ("binary_sensor", "sacn_active"),
    ):
        entity_id = entity_id_for(hass, platform, key)
        assert hass.states.get(entity_id).state == STATE_UNAVAILABLE, entity_id