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>
325 lines
12 KiB
Python
325 lines
12 KiB
Python
"""Tests for UPS monitor state transitions and shutdown sequence."""
|
|
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
import pytest
|
|
|
|
from vigilar.config import MQTTConfig, SystemConfig, UPSConfig, VigilarConfig
|
|
from vigilar.constants import EventType, Severity, Topics, UPSStatus
|
|
from vigilar.ups.monitor import UPSMonitor, _parse_status, _safe_float
|
|
|
|
|
|
# --- Helpers ---
|
|
|
|
def _make_cfg(**ups_overrides) -> VigilarConfig:
|
|
return VigilarConfig(
|
|
system=SystemConfig(data_dir="/tmp/vigilar-test"),
|
|
mqtt=MQTTConfig(),
|
|
ups=UPSConfig(**ups_overrides),
|
|
)
|
|
|
|
|
|
def _make_nut_vars(
|
|
status: str = "OL",
|
|
charge: str = "100",
|
|
runtime: str = "3600",
|
|
voltage: str = "230.0",
|
|
load: str = "25",
|
|
) -> dict[str, str]:
|
|
return {
|
|
"ups.status": status,
|
|
"battery.charge": charge,
|
|
"battery.runtime": runtime,
|
|
"input.voltage": voltage,
|
|
"ups.load": load,
|
|
}
|
|
|
|
|
|
# --- _parse_status ---
|
|
|
|
class TestParseStatus:
|
|
def test_online(self):
|
|
assert _parse_status("OL") == UPSStatus.ONLINE
|
|
|
|
def test_on_battery(self):
|
|
assert _parse_status("OB") == UPSStatus.ON_BATTERY
|
|
|
|
def test_low_battery(self):
|
|
assert _parse_status("OB LB") == UPSStatus.LOW_BATTERY
|
|
|
|
def test_online_charging(self):
|
|
assert _parse_status("OL CHRG") == UPSStatus.ONLINE
|
|
|
|
def test_unknown(self):
|
|
assert _parse_status("BYPASS") == UPSStatus.UNKNOWN
|
|
|
|
def test_empty(self):
|
|
assert _parse_status("") == UPSStatus.UNKNOWN
|
|
|
|
|
|
class TestSafeFloat:
|
|
def test_valid(self):
|
|
assert _safe_float("42.5") == 42.5
|
|
|
|
def test_none(self):
|
|
assert _safe_float(None) == 0.0
|
|
|
|
def test_garbage(self):
|
|
assert _safe_float("n/a", -1.0) == -1.0
|
|
|
|
|
|
# --- State Transition Tests ---
|
|
|
|
class TestStateTransitions:
|
|
"""Test UPSMonitor._poll_once for state transition detection."""
|
|
|
|
def _poll(self, monitor: UPSMonitor, bus: MagicMock, engine: MagicMock, nut_vars: dict):
|
|
client = MagicMock()
|
|
client.GetUPSVars.return_value = nut_vars
|
|
monitor._poll_once(client, bus, engine)
|
|
|
|
def test_online_to_on_battery_publishes_power_loss(self):
|
|
cfg = _make_cfg()
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ONLINE
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event") as mock_ie, \
|
|
patch("vigilar.ups.monitor.insert_system_event") as mock_ise:
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OB", charge="80", runtime="1800"))
|
|
|
|
assert monitor._prev_status == UPSStatus.ON_BATTERY
|
|
|
|
# Check power loss was published
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_STATUS in topics_published
|
|
assert Topics.UPS_POWER_LOSS in topics_published
|
|
|
|
def test_on_battery_to_online_publishes_restored(self):
|
|
cfg = _make_cfg()
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ON_BATTERY
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OL"))
|
|
|
|
assert monitor._prev_status == UPSStatus.ONLINE
|
|
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_RESTORED in topics_published
|
|
|
|
def test_low_battery_to_online_publishes_restored(self):
|
|
cfg = _make_cfg()
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.LOW_BATTERY
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OL"))
|
|
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_RESTORED in topics_published
|
|
|
|
def test_charge_below_threshold_publishes_low_battery(self):
|
|
cfg = _make_cfg(low_battery_threshold_pct=30)
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ON_BATTERY
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OB", charge="15", runtime="1800"))
|
|
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_LOW_BATTERY in topics_published
|
|
|
|
def test_charge_above_threshold_no_low_battery(self):
|
|
cfg = _make_cfg(low_battery_threshold_pct=20)
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ON_BATTERY
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OB", charge="50", runtime="3600"))
|
|
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_LOW_BATTERY not in topics_published
|
|
|
|
def test_critical_runtime_triggers_shutdown(self):
|
|
cfg = _make_cfg(critical_runtime_threshold_s=300, shutdown_delay_s=0)
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ON_BATTERY
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"), \
|
|
patch("vigilar.ups.shutdown.insert_event"), \
|
|
patch("vigilar.ups.shutdown.insert_system_event"), \
|
|
patch("vigilar.ups.shutdown.os.system") as mock_system, \
|
|
patch("vigilar.ups.shutdown.time.sleep"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OB", charge="10", runtime="60"))
|
|
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_CRITICAL in topics_published
|
|
assert Topics.SYSTEM_SHUTDOWN in topics_published
|
|
assert monitor._shutdown_triggered is True
|
|
mock_system.assert_called_once_with("shutdown -h now")
|
|
|
|
def test_critical_runtime_only_triggers_once(self):
|
|
cfg = _make_cfg(critical_runtime_threshold_s=300, shutdown_delay_s=0)
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ON_BATTERY
|
|
monitor._shutdown_triggered = True # already triggered
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OB", charge="5", runtime="30"))
|
|
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_CRITICAL not in topics_published
|
|
|
|
def test_online_no_low_battery_even_if_low_charge(self):
|
|
"""When on mains, low charge should not trigger low_battery event."""
|
|
cfg = _make_cfg(low_battery_threshold_pct=50)
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ONLINE
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars(status="OL", charge="10"))
|
|
|
|
topics_published = [c.args[0] for c in bus.publish_event.call_args_list]
|
|
assert Topics.UPS_LOW_BATTERY not in topics_published
|
|
|
|
def test_status_always_published(self):
|
|
cfg = _make_cfg()
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ONLINE
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
self._poll(monitor, bus, engine, _make_nut_vars())
|
|
|
|
assert bus.publish_event.call_args_list[0].args[0] == Topics.UPS_STATUS
|
|
|
|
def test_bytes_vars_decoded(self):
|
|
"""NUT client may return bytes keys/values."""
|
|
cfg = _make_cfg()
|
|
monitor = UPSMonitor(cfg)
|
|
monitor._prev_status = UPSStatus.ONLINE
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
byte_vars = {
|
|
b"ups.status": b"OL",
|
|
b"battery.charge": b"95",
|
|
b"battery.runtime": b"7200",
|
|
b"input.voltage": b"231.2",
|
|
b"ups.load": b"18",
|
|
}
|
|
|
|
client = MagicMock()
|
|
client.GetUPSVars.return_value = byte_vars
|
|
|
|
with patch("vigilar.ups.monitor.insert_event"), \
|
|
patch("vigilar.ups.monitor.insert_system_event"):
|
|
monitor._poll_once(client, bus, engine)
|
|
|
|
assert monitor._prev_status == UPSStatus.ONLINE
|
|
bus.publish_event.assert_called_once()
|
|
|
|
|
|
# --- Shutdown Sequence Tests ---
|
|
|
|
class TestShutdownSequence:
|
|
def test_execute_order(self):
|
|
"""Verify: publish -> log -> sleep -> os.system."""
|
|
from vigilar.ups.shutdown import ShutdownSequence
|
|
|
|
ups_cfg = UPSConfig(shutdown_delay_s=10)
|
|
seq = ShutdownSequence(ups_cfg)
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
call_order = []
|
|
|
|
bus.publish_event.side_effect = lambda *a, **kw: call_order.append("publish")
|
|
|
|
with patch("vigilar.ups.shutdown.insert_system_event", side_effect=lambda *a, **kw: call_order.append("insert_system_event")), \
|
|
patch("vigilar.ups.shutdown.insert_event", side_effect=lambda *a, **kw: call_order.append("insert_event")), \
|
|
patch("vigilar.ups.shutdown.time.sleep", side_effect=lambda s: call_order.append(f"sleep_{s}")) as mock_sleep, \
|
|
patch("vigilar.ups.shutdown.os.system", side_effect=lambda cmd: call_order.append(f"system_{cmd}")) as mock_sys:
|
|
seq.execute(bus, engine, reason="test_critical")
|
|
|
|
assert call_order[0] == "publish"
|
|
assert "insert_system_event" in call_order
|
|
assert "insert_event" in call_order
|
|
assert "sleep_10" in call_order
|
|
assert "system_shutdown -h now" in call_order
|
|
# Sleep and system call should come after publish/log
|
|
sleep_idx = call_order.index("sleep_10")
|
|
system_idx = call_order.index("system_shutdown -h now")
|
|
assert sleep_idx < system_idx
|
|
|
|
def test_custom_shutdown_command(self):
|
|
from vigilar.ups.shutdown import ShutdownSequence
|
|
|
|
ups_cfg = UPSConfig(shutdown_delay_s=0)
|
|
seq = ShutdownSequence(ups_cfg, shutdown_command="poweroff")
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.shutdown.insert_system_event"), \
|
|
patch("vigilar.ups.shutdown.insert_event"), \
|
|
patch("vigilar.ups.shutdown.time.sleep"), \
|
|
patch("vigilar.ups.shutdown.os.system") as mock_sys:
|
|
seq.execute(bus, engine, reason="test")
|
|
|
|
mock_sys.assert_called_once_with("poweroff")
|
|
|
|
def test_publishes_system_shutdown_topic(self):
|
|
from vigilar.ups.shutdown import ShutdownSequence
|
|
|
|
ups_cfg = UPSConfig(shutdown_delay_s=0)
|
|
seq = ShutdownSequence(ups_cfg)
|
|
|
|
bus = MagicMock()
|
|
engine = MagicMock()
|
|
|
|
with patch("vigilar.ups.shutdown.insert_system_event"), \
|
|
patch("vigilar.ups.shutdown.insert_event"), \
|
|
patch("vigilar.ups.shutdown.time.sleep"), \
|
|
patch("vigilar.ups.shutdown.os.system"):
|
|
seq.execute(bus, engine, reason="battery_dead")
|
|
|
|
bus.publish_event.assert_called_once()
|
|
assert bus.publish_event.call_args.args[0] == Topics.SYSTEM_SHUTDOWN
|
|
assert bus.publish_event.call_args.kwargs["reason"] == "battery_dead"
|