10b0cf4d0e
Phase 6 — Events + Rule Engine: - EventProcessor subprocess: subscribes to all MQTT events, logs to DB, evaluates rules, fires alert actions - ArmStateFSM: DISARMED/ARMED_HOME/ARMED_AWAY with PIN verification (HMAC-safe), DB persistence, MQTT state publishing - RuleEngine: AND/OR logic, 4 condition types (arm_state, sensor_event, camera_motion, time_window), per-rule cooldown tracking - SSE event stream with subscriber queue pattern and keepalive - Event acknowledge endpoint Phase 7 — Sensor Bridge: - SensorBridge subprocess: subscribes to Zigbee2MQTT, normalizes payloads (contact, occupancy, temperature, humidity, battery, linkquality) - GPIOHandler: conditional gpiozero import, callbacks for reed switches and PIR sensors - SensorRegistry: maps Zigbee addresses and names to config sensor IDs - SensorEvent/SensorState dataclasses - Web UI now shows real sensor states from DB Phase 8 — UPS Monitor: - UPSMonitor subprocess: polls NUT via pynut2 with reconnect backoff - State transition detection: OL→OB (power_loss), charge/runtime thresholds (low_battery, critical), OB→OL (restored) - ShutdownSequence: ordered shutdown with configurable delay and command - All conditionally imported (pynut2, gpiozero) for non-target platforms Fixed test_db fixture to use isolated engines (no global singleton leak). 96 tests passing. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
252 lines
8.1 KiB
Python
252 lines
8.1 KiB
Python
"""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 == []
|