Mock: add Thermostat + Button RequestProperties handlers
The HA coordinator walks ObjectType.THERMOSTAT (6) and ObjectType.BUTTON (3) via raw RequestProperties to discover them — the high-level get_object_properties() path only knows zones/units/areas in v1.0. The mock was returning Nak for both, which made HA discover zero thermostats and zero buttons no matter how MockState was seeded. src/omni_pca/mock_panel.py: - New MockButtonState dataclass (just a name) - MockState gains buttons: dict[int, MockButtonState] (with the same bare-string -> dataclass __post_init__ promotion as the others) - _OBJ_BUTTON = 3, _BUTTON_NAME_LEN = 12, _THERMOSTAT_NAME_LEN = 12 constants - thermostat_name_bytes() / button_name_bytes() helpers - _build_thermostat_properties() emits the 23-byte Properties body matching ThermostatProperties.parse offsets (object number BE u16, communicating flag, current temp, heat/cool setpoints, system/fan/ hold modes, thermostat type, 12-byte NUL-padded name) - _build_button_properties() emits the 15-byte body (object number BE u16 + 12-byte name) - _reply_properties / _object_store dispatch both new types tests/test_e2e_client_mock.py — two new e2e tests drive raw RequestProperties walks for thermostats and buttons against a seeded mock and assert ThermostatProperties / ButtonProperties parse cleanly, mirroring what the HA coordinator's _walk_properties() does. 333 tests pass (was 331); ruff clean. Mock surface now matches every opcode the HA coordinator and entity platforms actually call.
This commit is contained in:
parent
83d85a9885
commit
93b7e1f604
@ -59,6 +59,7 @@ _log = logging.getLogger(__name__)
|
|||||||
# enuObjectType (clsOmniLink2.cs / enuObjectType.cs)
|
# enuObjectType (clsOmniLink2.cs / enuObjectType.cs)
|
||||||
_OBJ_ZONE = 1
|
_OBJ_ZONE = 1
|
||||||
_OBJ_UNIT = 2
|
_OBJ_UNIT = 2
|
||||||
|
_OBJ_BUTTON = 3
|
||||||
_OBJ_AREA = 5
|
_OBJ_AREA = 5
|
||||||
_OBJ_THERMOSTAT = 6
|
_OBJ_THERMOSTAT = 6
|
||||||
|
|
||||||
@ -66,6 +67,8 @@ _OBJ_THERMOSTAT = 6
|
|||||||
_ZONE_NAME_LEN = 15
|
_ZONE_NAME_LEN = 15
|
||||||
_UNIT_NAME_LEN = 12
|
_UNIT_NAME_LEN = 12
|
||||||
_AREA_NAME_LEN = 12
|
_AREA_NAME_LEN = 12
|
||||||
|
_BUTTON_NAME_LEN = 12
|
||||||
|
_THERMOSTAT_NAME_LEN = 12
|
||||||
_PHONE_LEN = 24
|
_PHONE_LEN = 24
|
||||||
|
|
||||||
# Per-object-type record sizes for the basic Status (opcode 35) reply.
|
# Per-object-type record sizes for the basic Status (opcode 35) reply.
|
||||||
@ -156,6 +159,13 @@ class MockZoneState:
|
|||||||
return val & 0xFF
|
return val & 0xFF
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class MockButtonState:
|
||||||
|
"""One programmable button macro (no live state — buttons just fire programs)."""
|
||||||
|
|
||||||
|
name: str = ""
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class MockThermostatState:
|
class MockThermostatState:
|
||||||
"""One programmable thermostat. Defaults are sane Omni Pro II values."""
|
"""One programmable thermostat. Defaults are sane Omni Pro II values."""
|
||||||
@ -197,6 +207,7 @@ class MockState:
|
|||||||
units: dict[int, MockUnitState] = field(default_factory=dict)
|
units: dict[int, MockUnitState] = field(default_factory=dict)
|
||||||
areas: dict[int, MockAreaState] = field(default_factory=dict)
|
areas: dict[int, MockAreaState] = field(default_factory=dict)
|
||||||
thermostats: dict[int, MockThermostatState] = field(default_factory=dict)
|
thermostats: dict[int, MockThermostatState] = field(default_factory=dict)
|
||||||
|
buttons: dict[int, MockButtonState] = field(default_factory=dict)
|
||||||
|
|
||||||
# User-code table for ExecuteSecurityCommand validation.
|
# User-code table for ExecuteSecurityCommand validation.
|
||||||
# Mapping is ``{code_index: 4-digit pin}``; the panel returns the
|
# Mapping is ``{code_index: 4-digit pin}``; the panel returns the
|
||||||
@ -230,6 +241,7 @@ class MockState:
|
|||||||
self.units = _promote_dict(self.units, MockUnitState)
|
self.units = _promote_dict(self.units, MockUnitState)
|
||||||
self.areas = _promote_dict(self.areas, MockAreaState)
|
self.areas = _promote_dict(self.areas, MockAreaState)
|
||||||
self.thermostats = _promote_dict(self.thermostats, MockThermostatState)
|
self.thermostats = _promote_dict(self.thermostats, MockThermostatState)
|
||||||
|
self.buttons = _promote_dict(self.buttons, MockButtonState)
|
||||||
|
|
||||||
# ---- name-bytes helpers (kept for back-compat with old callers) -----
|
# ---- name-bytes helpers (kept for back-compat with old callers) -----
|
||||||
|
|
||||||
@ -245,6 +257,14 @@ class MockState:
|
|||||||
a = self.areas.get(idx)
|
a = self.areas.get(idx)
|
||||||
return _name_bytes(a.name if a else "", _AREA_NAME_LEN)
|
return _name_bytes(a.name if a else "", _AREA_NAME_LEN)
|
||||||
|
|
||||||
|
def thermostat_name_bytes(self, idx: int) -> bytes:
|
||||||
|
t = self.thermostats.get(idx)
|
||||||
|
return _name_bytes(t.name if t else "", _THERMOSTAT_NAME_LEN)
|
||||||
|
|
||||||
|
def button_name_bytes(self, idx: int) -> bytes:
|
||||||
|
b = self.buttons.get(idx)
|
||||||
|
return _name_bytes(b.name if b else "", _BUTTON_NAME_LEN)
|
||||||
|
|
||||||
|
|
||||||
def _promote_dict(
|
def _promote_dict(
|
||||||
raw: dict[int, object],
|
raw: dict[int, object],
|
||||||
@ -618,6 +638,10 @@ class MockPanel:
|
|||||||
return self._build_unit_properties(target)
|
return self._build_unit_properties(target)
|
||||||
if obj_type == _OBJ_AREA:
|
if obj_type == _OBJ_AREA:
|
||||||
return self._build_area_properties(target)
|
return self._build_area_properties(target)
|
||||||
|
if obj_type == _OBJ_THERMOSTAT:
|
||||||
|
return self._build_thermostat_properties(target)
|
||||||
|
if obj_type == _OBJ_BUTTON:
|
||||||
|
return self._build_button_properties(target)
|
||||||
return _build_nak(OmniLink2MessageType.RequestProperties)
|
return _build_nak(OmniLink2MessageType.RequestProperties)
|
||||||
|
|
||||||
def _object_store(self, obj_type: int) -> dict[int, object] | None:
|
def _object_store(self, obj_type: int) -> dict[int, object] | None:
|
||||||
@ -629,6 +653,8 @@ class MockPanel:
|
|||||||
return self.state.areas # type: ignore[return-value]
|
return self.state.areas # type: ignore[return-value]
|
||||||
if obj_type == _OBJ_THERMOSTAT:
|
if obj_type == _OBJ_THERMOSTAT:
|
||||||
return self.state.thermostats # type: ignore[return-value]
|
return self.state.thermostats # type: ignore[return-value]
|
||||||
|
if obj_type == _OBJ_BUTTON:
|
||||||
|
return self.state.buttons # type: ignore[return-value]
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def _build_zone_properties(self, index: int) -> Message:
|
def _build_zone_properties(self, index: int) -> Message:
|
||||||
@ -678,6 +704,58 @@ class MockPanel:
|
|||||||
)
|
)
|
||||||
return encode_v2(OmniLink2MessageType.Properties, body)
|
return encode_v2(OmniLink2MessageType.Properties, body)
|
||||||
|
|
||||||
|
def _build_thermostat_properties(self, index: int) -> Message:
|
||||||
|
# Properties.Data layout for Thermostat (Data[0]=opcode, body starts
|
||||||
|
# at Data[1]. ``payload`` here strips the opcode; payload[i]==Data[i+1]):
|
||||||
|
# payload[0] object type (Thermostat = 6)
|
||||||
|
# payload[1..2] object number (BE u16)
|
||||||
|
# payload[3] communicating flag
|
||||||
|
# payload[4] temperature raw
|
||||||
|
# payload[5] heat setpoint raw
|
||||||
|
# payload[6] cool setpoint raw
|
||||||
|
# payload[7] mode
|
||||||
|
# payload[8] fan mode
|
||||||
|
# payload[9] hold mode
|
||||||
|
# payload[10] thermostat type
|
||||||
|
# payload[11..22] 12-byte name
|
||||||
|
t = self.state.thermostats.get(index)
|
||||||
|
body = (
|
||||||
|
bytes(
|
||||||
|
[
|
||||||
|
_OBJ_THERMOSTAT,
|
||||||
|
(index >> 8) & 0xFF,
|
||||||
|
index & 0xFF,
|
||||||
|
t.status if t else 0, # communicating flag
|
||||||
|
t.temperature_raw if t else 0,
|
||||||
|
t.heat_setpoint_raw if t else 0,
|
||||||
|
t.cool_setpoint_raw if t else 0,
|
||||||
|
t.system_mode if t else 0,
|
||||||
|
t.fan_mode if t else 0,
|
||||||
|
t.hold_mode if t else 0,
|
||||||
|
1, # thermostat type: AUTO_HEAT_COOL
|
||||||
|
]
|
||||||
|
)
|
||||||
|
+ self.state.thermostat_name_bytes(index)
|
||||||
|
)
|
||||||
|
return encode_v2(OmniLink2MessageType.Properties, body)
|
||||||
|
|
||||||
|
def _build_button_properties(self, index: int) -> Message:
|
||||||
|
# Properties.Data layout for Button:
|
||||||
|
# payload[0] object type (Button = 3)
|
||||||
|
# payload[1..2] object number (BE u16)
|
||||||
|
# payload[3..14] 12-byte name (NUL-padded)
|
||||||
|
body = (
|
||||||
|
bytes(
|
||||||
|
[
|
||||||
|
_OBJ_BUTTON,
|
||||||
|
(index >> 8) & 0xFF,
|
||||||
|
index & 0xFF,
|
||||||
|
]
|
||||||
|
)
|
||||||
|
+ self.state.button_name_bytes(index)
|
||||||
|
)
|
||||||
|
return encode_v2(OmniLink2MessageType.Properties, body)
|
||||||
|
|
||||||
def _build_area_properties(self, index: int) -> Message:
|
def _build_area_properties(self, index: int) -> Message:
|
||||||
# Properties.Data for Area:
|
# Properties.Data for Area:
|
||||||
# [0]=opcode, [1]=ObjectType, [2..3]=ObjectNumber,
|
# [0]=opcode, [1]=ObjectType, [2..3]=ObjectNumber,
|
||||||
|
|||||||
@ -18,6 +18,7 @@ from omni_pca.connection import HandshakeError
|
|||||||
from omni_pca.events import ArmingChanged, UnitStateChanged
|
from omni_pca.events import ArmingChanged, UnitStateChanged
|
||||||
from omni_pca.mock_panel import (
|
from omni_pca.mock_panel import (
|
||||||
MockAreaState,
|
MockAreaState,
|
||||||
|
MockButtonState,
|
||||||
MockPanel,
|
MockPanel,
|
||||||
MockState,
|
MockState,
|
||||||
MockThermostatState,
|
MockThermostatState,
|
||||||
@ -272,6 +273,76 @@ async def test_e2e_unit_command_pushes_unit_state_changed_event() -> None:
|
|||||||
assert ev.is_on is True
|
assert ev.is_on is True
|
||||||
|
|
||||||
|
|
||||||
|
async def test_e2e_thermostat_properties_discovery() -> None:
|
||||||
|
"""The HA coordinator walks thermostats via raw RequestProperties; ensure
|
||||||
|
the mock answers and the response parses cleanly into ThermostatProperties.
|
||||||
|
"""
|
||||||
|
from omni_pca.models import ObjectType as ObjType
|
||||||
|
from omni_pca.models import ThermostatProperties
|
||||||
|
from omni_pca.opcodes import OmniLink2MessageType
|
||||||
|
|
||||||
|
state = MockState(
|
||||||
|
thermostats={
|
||||||
|
1: MockThermostatState(name="LIVING_ROOM"),
|
||||||
|
3: MockThermostatState(name="MASTER"),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
panel = MockPanel(controller_key=CONTROLLER_KEY, state=state)
|
||||||
|
async with (
|
||||||
|
panel.serve() as (host, port),
|
||||||
|
OmniClient(host=host, port=port, controller_key=CONTROLLER_KEY) as cli,
|
||||||
|
):
|
||||||
|
found: dict[int, str] = {}
|
||||||
|
cursor = 0
|
||||||
|
for _ in range(10):
|
||||||
|
payload = bytes(
|
||||||
|
[int(ObjType.THERMOSTAT), (cursor >> 8) & 0xFF, cursor & 0xFF, 1, 0, 0, 0]
|
||||||
|
)
|
||||||
|
reply = await cli.connection.request(
|
||||||
|
OmniLink2MessageType.RequestProperties, payload
|
||||||
|
)
|
||||||
|
if reply.opcode == int(OmniLink2MessageType.EOD):
|
||||||
|
break
|
||||||
|
t = ThermostatProperties.parse(reply.payload)
|
||||||
|
found[t.index] = t.name
|
||||||
|
cursor = t.index
|
||||||
|
assert found == {1: "LIVING_ROOM", 3: "MASTER"}
|
||||||
|
|
||||||
|
|
||||||
|
async def test_e2e_button_properties_discovery() -> None:
|
||||||
|
"""Same idea for button discovery (the HA coordinator drives this too)."""
|
||||||
|
from omni_pca.models import ButtonProperties
|
||||||
|
from omni_pca.models import ObjectType as ObjType
|
||||||
|
from omni_pca.opcodes import OmniLink2MessageType
|
||||||
|
|
||||||
|
state = MockState(
|
||||||
|
buttons={
|
||||||
|
1: MockButtonState(name="GOOD_MORNING"),
|
||||||
|
5: MockButtonState(name="MOVIE_MODE"),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
panel = MockPanel(controller_key=CONTROLLER_KEY, state=state)
|
||||||
|
async with (
|
||||||
|
panel.serve() as (host, port),
|
||||||
|
OmniClient(host=host, port=port, controller_key=CONTROLLER_KEY) as cli,
|
||||||
|
):
|
||||||
|
found: dict[int, str] = {}
|
||||||
|
cursor = 0
|
||||||
|
for _ in range(10):
|
||||||
|
payload = bytes(
|
||||||
|
[int(ObjType.BUTTON), (cursor >> 8) & 0xFF, cursor & 0xFF, 1, 0, 0, 0]
|
||||||
|
)
|
||||||
|
reply = await cli.connection.request(
|
||||||
|
OmniLink2MessageType.RequestProperties, payload
|
||||||
|
)
|
||||||
|
if reply.opcode == int(OmniLink2MessageType.EOD):
|
||||||
|
break
|
||||||
|
b = ButtonProperties.parse(reply.payload)
|
||||||
|
found[b.index] = b.name
|
||||||
|
cursor = b.index
|
||||||
|
assert found == {1: "GOOD_MORNING", 5: "MOVIE_MODE"}
|
||||||
|
|
||||||
|
|
||||||
async def test_e2e_acknowledge_alerts() -> None:
|
async def test_e2e_acknowledge_alerts() -> None:
|
||||||
panel = MockPanel(controller_key=CONTROLLER_KEY)
|
panel = MockPanel(controller_key=CONTROLLER_KEY)
|
||||||
async with (
|
async with (
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user