Ryan Malloy
b412dc0f37
Coordinator's _discover_programs is no longer a placeholder. It now drives client.iter_programs() (v2 path) or the v1 adapter's forward to OmniClientV1.iter_programs (v1 path), populating OmniData.programs with decoded Program records keyed by slot. Errors are logged and swallowed so partial enumeration doesn't break entry setup — programs are non-critical telemetry. OmniData.programs is now dict[int, Program] rather than the ProgramProperties dict that was an empty placeholder. The ProgramProperties dataclass remains in models.py for the Properties opcode reply path; only the coordinator's value type changed. New OmniProgramsSensor on the sensor platform: a single diagnostic entity per panel whose state is the count of defined programs and whose 'programs' attribute lists each program's slot, type name, and schedule fields. Easy to consume from automations and the developer-states UI. Mock fixture seeds three programs (TIMED+TIMED+EVENT at slots 12 / 42 / 99). New integration test verifies the sensor enumerates them in slot-ascending order with the expected per-record fields. Full suite: 494 passed, 1 skipped (fixture-gated).
181 lines
6.6 KiB
Python
181 lines
6.6 KiB
Python
"""HA-side integration: integration loads, entities materialize."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from custom_components.omni_pca.const import DOMAIN
|
|
from homeassistant.config_entries import ConfigEntryState
|
|
from homeassistant.const import STATE_OFF, STATE_UNAVAILABLE
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
|
|
async def test_integration_loads_against_mock_panel(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
"""End-to-end: HA discovers our integration, completes the secure
|
|
session against the mock, populates the coordinator, and lands in
|
|
LOADED state with no errors."""
|
|
assert configured_panel.state is ConfigEntryState.LOADED
|
|
coordinator = hass.data[DOMAIN][configured_panel.entry_id]
|
|
assert coordinator.data is not None
|
|
assert coordinator.data.system_info is not None
|
|
assert coordinator.data.system_info.model_name == "Omni Pro II"
|
|
|
|
|
|
async def test_zone_entities_created(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
"""Every named zone in MockState lands as a binary_sensor entity."""
|
|
states = hass.states.async_all("binary_sensor")
|
|
zone_entity_ids = [s.entity_id for s in states if "front_door" in s.entity_id.lower()
|
|
or "garage_entry" in s.entity_id.lower()
|
|
or "living_motion" in s.entity_id.lower()]
|
|
# Each zone gets a primary + bypassed entity, so at least 3 names x 2 = 6
|
|
# plus the system-level AC / battery / trouble entities.
|
|
assert len(zone_entity_ids) >= 3, (
|
|
f"expected zone entities, got {[s.entity_id for s in states]}"
|
|
)
|
|
|
|
|
|
async def test_alarm_panel_entity_created(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
"""One alarm_control_panel per discovered area."""
|
|
states = hass.states.async_all("alarm_control_panel")
|
|
assert len(states) == 1
|
|
assert states[0].state != STATE_UNAVAILABLE
|
|
|
|
|
|
async def test_light_entities_for_units(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
"""One light entity per discovered unit."""
|
|
states = hass.states.async_all("light")
|
|
assert len(states) == 2
|
|
# Both units default to off in the mock.
|
|
for s in states:
|
|
assert s.state in (STATE_OFF, STATE_UNAVAILABLE)
|
|
|
|
|
|
async def test_switch_entities_for_zone_bypass(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
"""One bypass switch per binary zone."""
|
|
states = hass.states.async_all("switch")
|
|
assert len(states) == 3 # one per binary zone
|
|
|
|
|
|
async def test_climate_entity_for_thermostat(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
states = hass.states.async_all("climate")
|
|
assert len(states) == 1
|
|
|
|
|
|
async def test_button_entity_for_panel_button(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
states = hass.states.async_all("button")
|
|
assert len(states) == 1
|
|
|
|
|
|
async def test_programs_sensor_reflects_seeded_panel(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
"""The diagnostic Panel Programs sensor enumerates discovered programs.
|
|
|
|
Mock seed has 3 programs at slots 12 / 42 / 99 (see
|
|
:func:`populated_state`); after discovery the coordinator stashes them
|
|
on ``OmniData.programs`` and the sensor surfaces a count + per-slot
|
|
summary attribute.
|
|
"""
|
|
programs_states = [
|
|
s for s in hass.states.async_all("sensor")
|
|
if s.entity_id.endswith("_panel_programs") or "panel_programs" in s.entity_id
|
|
]
|
|
assert len(programs_states) == 1, (
|
|
f"expected one panel_programs sensor, got {[s.entity_id for s in programs_states]}"
|
|
)
|
|
s = programs_states[0]
|
|
assert int(s.state) == 3
|
|
summaries = s.attributes["programs"]
|
|
assert [p["slot"] for p in summaries] == [12, 42, 99]
|
|
# Spot-check the per-record fields landed in summary form.
|
|
by_slot = {p["slot"]: p for p in summaries}
|
|
assert by_slot[12]["type"] == "TIMED"
|
|
assert by_slot[12]["hour"] == 6 and by_slot[12]["minute"] == 0
|
|
assert by_slot[99]["type"] == "EVENT"
|
|
assert by_slot[99]["month"] == 5 and by_slot[99]["day"] == 12
|
|
|
|
|
|
async def test_event_entity_per_panel(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
states = hass.states.async_all("event")
|
|
assert len(states) == 1
|
|
|
|
|
|
async def test_unload_entry(
|
|
hass: HomeAssistant, configured_panel
|
|
) -> None:
|
|
"""Unloading the config entry tears everything down cleanly."""
|
|
assert await hass.config_entries.async_unload(configured_panel.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert configured_panel.state is ConfigEntryState.NOT_LOADED
|
|
# Coordinator removed from hass.data
|
|
assert configured_panel.entry_id not in hass.data.get(DOMAIN, {})
|
|
|
|
|
|
async def test_turn_unit_on_via_light_service(
|
|
hass: HomeAssistant, configured_panel, panel
|
|
) -> None:
|
|
"""Drive a HA service call; verify it reaches the mock and updates state."""
|
|
mock, _, _ = panel
|
|
light_states = hass.states.async_all("light")
|
|
assert light_states, "expected at least one light entity"
|
|
target = light_states[0].entity_id
|
|
await hass.services.async_call(
|
|
"light", "turn_on", {"entity_id": target}, blocking=True
|
|
)
|
|
await hass.async_block_till_done()
|
|
# The mock's state updated for whichever unit was first in sorted order.
|
|
on_units = [u for u in mock.state.units.values() if u.state == 1]
|
|
assert on_units, "expected the mock to record the unit as ON"
|
|
|
|
|
|
async def test_arm_panel_via_alarm_service(
|
|
hass: HomeAssistant, configured_panel, panel
|
|
) -> None:
|
|
"""Arm the panel from HA; verify the mock area transitions."""
|
|
mock, _, _ = panel
|
|
alarm_states = hass.states.async_all("alarm_control_panel")
|
|
assert alarm_states, "expected one alarm_control_panel entity"
|
|
target = alarm_states[0].entity_id
|
|
await hass.services.async_call(
|
|
"alarm_control_panel",
|
|
"alarm_arm_away",
|
|
{"entity_id": target, "code": "1234"},
|
|
blocking=True,
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert mock.state.areas[1].mode == 3 # SecurityMode.AWAY
|
|
|
|
|
|
async def test_arm_panel_with_wrong_code_keeps_disarmed(
|
|
hass: HomeAssistant, configured_panel, panel
|
|
) -> None:
|
|
"""Wrong code: panel stays disarmed and HA surfaces the error."""
|
|
mock, _, _ = panel
|
|
alarm_states = hass.states.async_all("alarm_control_panel")
|
|
target = alarm_states[0].entity_id
|
|
# The service should raise; we don't assert the exception class because
|
|
# HA wraps it. We just assert the panel mode didn't change.
|
|
import contextlib
|
|
with contextlib.suppress(Exception):
|
|
await hass.services.async_call(
|
|
"alarm_control_panel",
|
|
"alarm_arm_away",
|
|
{"entity_id": target, "code": "9999"},
|
|
blocking=True,
|
|
)
|
|
assert mock.state.areas[1].mode == 0 # still disarmed
|