"""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"