SetupData side (clsHAC.cs:3020-3038): five contiguous bool[8] arrays immediately after ExitDelay carry per-area config flags. Offsets: 1787..1794: EntryChime 1795..1802: QuickArm 1803..1810: AutoBypass 1811..1818: AllOnForAlarm 1819..1826: TroubleBeep Verified against live fixture: area 1 shows real homeowner choices (QuickArm + AllOnForAlarm enabled, others off), unused areas 2-8 carry the panel defaults (EntryChime/AutoBypass/TroubleBeep on by default). PerimeterChime and AudibleExitDelay aren't in this contiguous block — they live past FlashLightNum, HouseCodes flags, and 6 TimeClock When-structs. Deferred. New PcaAccount fields: area_entry_chime, area_quick_arm, area_auto_bypass, area_all_on_for_alarm, area_trouble_beep — all dict[int, bool]. MockAreaState gains the same five fields. They aren't carried in the Properties reply on the wire (the OL2 message format doesn't have them), so they live on MockState for snapshots and any future SetupData-aware code, but don't surface through HA discovery yet. v2 client list_area_names fallback: when the Properties walk turns up no named areas (common — most homes don't name them), synthesize "Area 1".."Area 8" so HA's _discover_areas has slots to walk. Mirrors the v1 adapter behaviour exactly. Knock-on win in the live-fixture HA test: area 1 now reaches coordinator.data.areas with its configured 60s/90s delays from SetupData, end-to-end through .pca → MockState → wire Properties → HA's AreaProperties parser. Full suite: 499 passed, 1 skipped.
208 lines
8.2 KiB
Python
208 lines
8.2 KiB
Python
"""HA-side integration: optional .pca file source for panel programs.
|
|
|
|
When ``CONF_PCA_PATH`` is set in the entry data, the coordinator should
|
|
parse the .pca file at that path (with ``CONF_PCA_KEY`` as the per-install
|
|
key) and use those programs *instead* of streaming them over the wire.
|
|
The wire-based discovery for everything else (zones, units, etc.) is
|
|
unaffected.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from custom_components.omni_pca.const import (
|
|
CONF_CONTROLLER_KEY,
|
|
CONF_PCA_KEY,
|
|
CONF_PCA_PATH,
|
|
DOMAIN,
|
|
)
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.const import CONF_HOST, CONF_PORT
|
|
from homeassistant.core import HomeAssistant
|
|
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
|
|
|
from .conftest import CONTROLLER_KEY_HEX
|
|
|
|
LIVE_FIXTURE_PLAIN = Path(
|
|
"/home/kdm/home-auto/HAI/pca-re/extracted/Our_House.pca.plain"
|
|
)
|
|
|
|
|
|
def _materialize_encrypted_fixture(tmp_path: Path) -> tuple[Path, int]:
|
|
"""Re-encrypt the plain fixture so parse_pca_file can decrypt it.
|
|
|
|
parse_pca_file always runs the XOR keystream. The plain dump bypasses
|
|
that, so we re-apply the keystream with KEY_EXPORT and write the
|
|
result to tmp_path. Returns (file_path, key) the coordinator should use.
|
|
"""
|
|
from omni_pca.pca_file import KEY_EXPORT, decrypt_pca_bytes
|
|
|
|
plain = LIVE_FIXTURE_PLAIN.read_bytes()
|
|
# XOR is symmetric — "decrypt" of plain bytes with the export key
|
|
# produces a valid encrypted .pca that parse_pca_file can read back.
|
|
encrypted = decrypt_pca_bytes(plain, KEY_EXPORT)
|
|
fixture = tmp_path / "Test_House.pca"
|
|
fixture.write_bytes(encrypted)
|
|
return fixture, KEY_EXPORT
|
|
|
|
|
|
@pytest.fixture
|
|
async def configured_with_pca(
|
|
hass: HomeAssistant, panel: tuple[Any, str, int], tmp_path: Path
|
|
) -> AsyncIterator[ConfigEntry]:
|
|
"""Config entry pointing at a .pca file fixture for programs."""
|
|
if not LIVE_FIXTURE_PLAIN.is_file():
|
|
pytest.skip(f"live .pca fixture missing: {LIVE_FIXTURE_PLAIN}")
|
|
|
|
fixture_path, pca_key = _materialize_encrypted_fixture(tmp_path)
|
|
|
|
_, host, port = panel
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={
|
|
CONF_HOST: host,
|
|
CONF_PORT: port,
|
|
CONF_CONTROLLER_KEY: CONTROLLER_KEY_HEX,
|
|
CONF_PCA_PATH: str(fixture_path),
|
|
CONF_PCA_KEY: pca_key,
|
|
},
|
|
title=f"Mock Omni @ {host}:{port} (with .pca)",
|
|
unique_id=f"{host}:{port}",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
try:
|
|
yield entry
|
|
finally:
|
|
if entry.entry_id in hass.data.get(DOMAIN, {}):
|
|
await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
|
|
async def test_pca_source_overrides_wire_programs(
|
|
hass: HomeAssistant, configured_with_pca: ConfigEntry
|
|
) -> None:
|
|
"""The fixture .pca has 330 defined programs (Phase 1 recon). The mock
|
|
panel only seeded 3 in conftest. When pca_path is set, the .pca count
|
|
wins — proving the coordinator routed through _discover_programs_from_pca,
|
|
not iter_programs."""
|
|
coordinator = hass.data[DOMAIN][configured_with_pca.entry_id]
|
|
assert len(coordinator.data.programs) == 330
|
|
|
|
# Sanity: the diagnostic sensor reflects the .pca count, not the mock seed.
|
|
sensors = [
|
|
s for s in hass.states.async_all("sensor")
|
|
if "panel_programs" in s.entity_id
|
|
]
|
|
assert len(sensors) == 1
|
|
assert int(sensors[0].state) == 330
|
|
|
|
|
|
async def test_mockpanel_from_pca_drives_full_ha_discovery(
|
|
hass: HomeAssistant, tmp_path: Path
|
|
) -> None:
|
|
"""End-to-end: build a MockPanel state straight from the live .pca,
|
|
then point HA at that mock with no other configuration. The
|
|
integration should discover *every* named zone / unit / button /
|
|
thermostat from the .pca via the normal wire path — no .pca config
|
|
needed, because the mock is now serving real data.
|
|
"""
|
|
if not LIVE_FIXTURE_PLAIN.is_file():
|
|
pytest.skip(f"live .pca fixture missing: {LIVE_FIXTURE_PLAIN}")
|
|
|
|
from custom_components.omni_pca.const import CONF_CONTROLLER_KEY
|
|
from omni_pca.mock_panel import MockPanel, MockState
|
|
from omni_pca.pca_file import KEY_EXPORT, decrypt_pca_bytes
|
|
|
|
encrypted = decrypt_pca_bytes(LIVE_FIXTURE_PLAIN.read_bytes(), KEY_EXPORT)
|
|
state = MockState.from_pca(encrypted, key=KEY_EXPORT)
|
|
# Sanity — the from_pca seeding matches the live fixture's names.
|
|
assert len(state.zones) == 16
|
|
assert len(state.units) == 44
|
|
|
|
panel = MockPanel(
|
|
controller_key=bytes(range(16)), # matches CONTROLLER_KEY_HEX
|
|
state=state,
|
|
)
|
|
async with panel.serve(host="127.0.0.1") as (host, port):
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={
|
|
CONF_HOST: host,
|
|
CONF_PORT: port,
|
|
CONF_CONTROLLER_KEY: CONTROLLER_KEY_HEX,
|
|
},
|
|
title=f"Mock Omni @ {host}:{port} (from .pca)",
|
|
unique_id=f"{host}:{port}",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
try:
|
|
coordinator = hass.data[DOMAIN][entry.entry_id]
|
|
# All 16 zones surfaced through normal wire discovery.
|
|
assert len(coordinator.data.zones) == 16
|
|
# Units, buttons, thermostats too.
|
|
assert len(coordinator.data.units) == 44
|
|
assert len(coordinator.data.buttons) == 16
|
|
assert len(coordinator.data.thermostats) == 2
|
|
# And the programs sensor reflects 330 from wire iter_programs.
|
|
assert len(coordinator.data.programs) == 330
|
|
# Zone types flowed from SetupData → mock → wire Properties
|
|
# reply → HA's ZoneProperties parser.
|
|
zone_types_by_slot = {
|
|
idx: z.zone_type for idx, z in coordinator.data.zones.items()
|
|
}
|
|
assert zone_types_by_slot[1] == 0x00 # GARAGE ENTRY → EntryExit
|
|
assert zone_types_by_slot[3] == 0x01 # BACK DOOR → Perimeter
|
|
assert zone_types_by_slot[7] == 0x03 # LIVINGROOM MOT → AwayInt
|
|
assert zone_types_by_slot[11] == 0x55 # OUTSIDE TEMP → outdoor temp
|
|
# Per-zone area assignments — single-area install, every
|
|
# zone surfaces as area=1 through the wire Properties reply.
|
|
for idx, z in coordinator.data.zones.items():
|
|
assert z.area == 1, f"zone {idx} expected area=1 got {z.area}"
|
|
# Areas: the live fixture has no user-assigned area names
|
|
# but the v2 client's list_area_names now falls back to
|
|
# "Area 1".."Area 8". HA's _discover_areas then enumerates
|
|
# each, walks the Properties reply, and lands the configured
|
|
# entry/exit delays from SetupData.
|
|
assert 1 in coordinator.data.areas
|
|
assert coordinator.data.areas[1].entry_delay == 60
|
|
assert coordinator.data.areas[1].exit_delay == 90
|
|
finally:
|
|
await hass.config_entries.async_unload(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
|
|
async def test_pca_path_validation_rejects_missing_file(
|
|
hass: HomeAssistant, tmp_path: Path
|
|
) -> None:
|
|
"""The config-flow validator returns ``pca_not_found`` for an absent
|
|
file. We exercise the helper directly to avoid spinning a full mock
|
|
panel just for the validation branch."""
|
|
from custom_components.omni_pca.config_flow import OmniConfigFlow
|
|
|
|
flow = OmniConfigFlow()
|
|
flow.hass = hass
|
|
err = await flow._validate_pca(str(tmp_path / "does-not-exist.pca"), 0)
|
|
assert err == "pca_not_found"
|
|
|
|
|
|
async def test_pca_path_validation_rejects_garbage(
|
|
hass: HomeAssistant, tmp_path: Path
|
|
) -> None:
|
|
"""A file that doesn't decode as a .pca returns ``pca_decode_failed``."""
|
|
from custom_components.omni_pca.config_flow import OmniConfigFlow
|
|
|
|
garbage = tmp_path / "garbage.pca"
|
|
garbage.write_bytes(b"not a real pca file" * 1000)
|
|
flow = OmniConfigFlow()
|
|
flow.hass = hass
|
|
err = await flow._validate_pca(str(garbage), 0)
|
|
assert err == "pca_decode_failed"
|