status alerts grace period 10 mins

2026-09-10 20:46:31 +02:00
parent c23350400c
commit d74634331c
3 changed files with 139 additions and 10 deletions
@@ -0,0 +1,99 @@
+"""Status alert regression tests with hardware integrations stubbed out."""
+
+import importlib.util
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+
+with patch.dict(sys.modules, {
+    name: MagicMock()
+    for name in ("paramiko", "rrdtool", "dbus_next", "dbus_next.aio",
+                 "dbus_next.constants", "PyNUTClient")
+}):
+    spec = importlib.util.spec_from_file_location(
+        "collector_under_test", Path(__file__).with_name("ups2rrd.py")
+    )
+    collector = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(collector)
+
+
+class StatusAlertTests(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.state_file = str(Path(self.directory.name) / "state.json")
+        for target, value in (
+            ("ALERT_STATE_FILE", self.state_file),
+            ("notify", MagicMock()),
+            ("shutdown_remote_server", MagicMock()),
+            ("shutdown_local_server", MagicMock()),
+        ):
+            patcher = patch.object(collector, target, value)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+
+    def sample(self, seconds, status, runtime=1000):
+        with patch.object(collector.time, "time", return_value=seconds):
+            collector.process_alerts({
+                "status": status, "battery_charge": 100,
+                "battery_runtime": runtime,
+            })
+        with open(self.state_file) as state_file:
+            return json.load(state_file)
+
+    def test_initial_baseline_and_short_flip_flops(self):
+        self.sample(0, "OL")
+        self.sample(60, "OB")
+        self.sample(659, "OL")
+        self.sample(700, "OB")
+        state = self.sample(1300, "OL")
+        collector.notify.assert_not_called()
+        self.assertIsNone(state["pending_status"])
+
+    def test_persistent_change_and_recovery_each_alert_once(self):
+        self.sample(0, "OL")
+        self.sample(60, "OB")
+        self.sample(659, "OB")
+        collector.notify.assert_not_called()
+        self.sample(660, "OB")
+        self.sample(720, "OB")
+        collector.notify.assert_called_once_with("UPS status changed: OL -> OB")
+        self.sample(780, "OL")
+        self.sample(1380, "OL")
+        self.assertEqual(collector.notify.call_count, 2)
+        collector.notify.assert_called_with("UPS status changed: OB -> OL")
+
+    def test_different_pending_status_restarts_timer(self):
+        self.sample(0, "OL")
+        self.sample(60, "OB")
+        self.sample(600, "LB")
+        self.sample(660, "LB")
+        collector.notify.assert_not_called()
+        self.sample(1200, "LB")
+        collector.notify.assert_called_once_with("UPS status changed: OL -> LB")
+
+    def test_existing_state_migrates_and_alerts_after_late_poll(self):
+        with open(self.state_file, "w") as state_file:
+            json.dump({"status": "OL"}, state_file)
+        self.sample(100, "OB")
+        collector.notify.assert_not_called()
+        self.sample(750, "OB")
+        collector.notify.assert_called_once_with("UPS status changed: OL -> OB")
+
+    def test_runtime_alert_and_shutdowns_are_immediate(self):
+        self.sample(0, "OL")
+        state = self.sample(60, "OB", runtime=100)
+        self.assertEqual(state["confirmed_status"], "OL")
+        collector.shutdown_remote_server.assert_called_once()
+        collector.shutdown_local_server.assert_called_once()
+        messages = [call.args[0] for call in collector.notify.call_args_list]
+        self.assertTrue(any("runtime is below 5 minutes" in m for m in messages))
+        self.assertFalse(any("UPS status changed" in m for m in messages))
+
+
+if __name__ == "__main__":
+    unittest.main()