"""Tests for sensor bridge subsystem — models, registry, payload normalization.""" import time import pytest from vigilar.config import SensorConfig, VigilarConfig from vigilar.constants import EventType from vigilar.sensors.bridge import normalize_zigbee_payload from vigilar.sensors.models import SensorEvent, SensorState from vigilar.sensors.registry import SensorRegistry # --- Fixtures --- @pytest.fixture def zigbee_sensor_config(): return VigilarConfig( sensors=[ SensorConfig( id="front_door", display_name="Front Door", type="CONTACT", protocol="ZIGBEE", device_address="front_door_sensor", location="Entrance", ), SensorConfig( id="hallway_motion", display_name="Hallway Motion", type="MOTION", protocol="ZIGBEE", device_address="hallway_pir", location="Hallway", ), SensorConfig( id="living_room_temp", display_name="Living Room Temp", type="TEMPERATURE", protocol="ZIGBEE", device_address="lr_climate", location="Living Room", ), SensorConfig( id="garage_reed", display_name="Garage Reed", type="CONTACT", protocol="GPIO", device_address="17", location="Garage", ), SensorConfig( id="disabled_sensor", display_name="Disabled", type="CONTACT", protocol="ZIGBEE", device_address="disabled_one", location="Attic", enabled=False, ), ], ) # --- Dataclass Tests --- class TestSensorEvent: def test_create_with_state(self): event = SensorEvent( sensor_id="front_door", event_type=EventType.CONTACT_OPEN, state="OPEN", source_protocol="ZIGBEE", ) assert event.sensor_id == "front_door" assert event.event_type == EventType.CONTACT_OPEN assert event.state == "OPEN" assert event.value is None assert event.timestamp > 0 def test_create_with_value(self): event = SensorEvent( sensor_id="living_room_temp", event_type="temperature", value=22.5, source_protocol="ZIGBEE", ) assert event.value == 22.5 assert event.state is None def test_to_dict(self): event = SensorEvent( sensor_id="s1", event_type="temperature", value=18.0, timestamp=1000, source_protocol="ZIGBEE", ) d = event.to_dict() assert d["sensor_id"] == "s1" assert d["value"] == 18.0 assert d["ts"] == 1000 assert "state" not in d class TestSensorState: def test_create(self): state = SensorState( sensor_id="front_door", states={"CONTACT_OPEN": "OPEN"}, last_seen=12345, battery_pct=87, ) assert state.sensor_id == "front_door" assert state.battery_pct == 87 assert state.signal_strength is None def test_to_dict(self): state = SensorState(sensor_id="s1", states={"temp": 22.5}, last_seen=100) d = state.to_dict() assert d["sensor_id"] == "s1" assert d["states"] == {"temp": 22.5} assert d["battery_pct"] is None # --- Registry Tests --- class TestSensorRegistry: def test_lookup_by_zigbee_name(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) sensor = reg.get_sensor_by_zigbee_name("front_door_sensor") assert sensor is not None assert sensor.id == "front_door" def test_lookup_by_address(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) sensor = reg.get_sensor_by_address("hallway_pir") assert sensor is not None assert sensor.id == "hallway_motion" def test_lookup_unknown_returns_none(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) assert reg.get_sensor_by_zigbee_name("nonexistent") is None assert reg.get_sensor_by_address("nonexistent") is None def test_disabled_sensors_excluded(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) assert reg.get_sensor_by_zigbee_name("disabled_one") is None assert reg.get_sensor("disabled_sensor") is None def test_gpio_not_in_zigbee_lookup(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) assert reg.get_sensor_by_zigbee_name("17") is None def test_gpio_sensors_list(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) gpio = reg.gpio_sensors() assert len(gpio) == 1 assert gpio[0].id == "garage_reed" def test_zigbee_sensors_list(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) zigbee = reg.zigbee_sensors() assert len(zigbee) == 3 def test_all_sensors_excludes_disabled(self, zigbee_sensor_config): reg = SensorRegistry(zigbee_sensor_config) assert len(reg.all_sensors()) == 4 # --- Zigbee Payload Normalization Tests --- class TestNormalizeZigbeePayload: def test_contact_open(self): results = normalize_zigbee_payload({"contact": False}) assert len(results) == 1 r = results[0] assert r["event_type"] == EventType.CONTACT_OPEN assert r["state"] == "OPEN" assert r["source"] == "zigbee" assert "ts" in r def test_contact_closed(self): results = normalize_zigbee_payload({"contact": True}) assert len(results) == 1 assert results[0]["event_type"] == EventType.CONTACT_CLOSED assert results[0]["state"] == "CLOSED" def test_motion_active(self): results = normalize_zigbee_payload({"occupancy": True}) assert len(results) == 1 assert results[0]["event_type"] == EventType.MOTION_START assert results[0]["state"] == "ACTIVE" def test_motion_idle(self): results = normalize_zigbee_payload({"occupancy": False}) assert len(results) == 1 assert results[0]["event_type"] == EventType.MOTION_END assert results[0]["state"] == "IDLE" def test_temperature(self): results = normalize_zigbee_payload({"temperature": 22.5}) assert len(results) == 1 r = results[0] assert r["event_type"] == "temperature" assert r["value"] == 22.5 assert r["unit"] == "C" def test_humidity(self): results = normalize_zigbee_payload({"humidity": 45}) assert len(results) == 1 assert results[0]["value"] == 45.0 def test_battery(self): results = normalize_zigbee_payload({"battery": 87}) assert len(results) == 1 assert results[0]["event_type"] == "battery" assert results[0]["pct"] == 87 def test_linkquality(self): results = normalize_zigbee_payload({"linkquality": 120}) assert len(results) == 1 assert results[0]["event_type"] == "linkquality" assert results[0]["value"] == 120 def test_multi_field_payload(self): """A real Zigbee sensor often sends multiple fields at once.""" results = normalize_zigbee_payload({ "contact": False, "battery": 92, "linkquality": 85, "temperature": 21.3, }) event_types = {r["event_type"] for r in results} assert EventType.CONTACT_OPEN in event_types assert "battery" in event_types assert "temperature" in event_types assert "linkquality" in event_types assert len(results) == 4 def test_empty_payload(self): results = normalize_zigbee_payload({}) assert results == [] def test_unknown_fields_ignored(self): results = normalize_zigbee_payload({"unknown_field": 42, "foo": "bar"}) assert results == []