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