From d74634331cf1b4b693f327f575c586edebd22bc0 Mon Sep 17 00:00:00 2001 From: Ignace Date: Thu, 10 Sep 2026 20:46:31 +0200 Subject: [PATCH] status alerts grace period 10 mins --- README.md | 8 +++- test_status_alerts.py | 99 +++++++++++++++++++++++++++++++++++++++++++ ups2rrd.py | 42 ++++++++++++++---- 3 files changed, 139 insertions(+), 10 deletions(-) create mode 100644 test_status_alerts.py diff --git a/README.md b/README.md index 9e67323..c0f41e2 100644 --- a/README.md +++ b/README.md @@ -825,7 +825,7 @@ full-size PNG when a graph is clicked. The collector also sends an ntfy notification when: -- The UPS status changes +- The UPS status changes and the new status persists for 10 minutes - Mean battery charge over the last 10 measurements crosses below 50% - Mean battery charge over the last 10 measurements crosses below 5% - Estimated battery runtime crosses below 5 minutes @@ -835,6 +835,12 @@ Alert state is stored in `/var/lib/rrd/bx950mi-alert-state.json`, allowing changes to be detected across cron runs. The first run establishes a baseline without sending notifications. Threshold notifications can fire again after a value recovers above the threshold and later drops below it. +Status alerts wait until the new status has persisted for 10 minutes. Returning +to the last confirmed status cancels the pending alert, so brief flip-flops +produce no status notifications. A different pending status restarts the timer. +Pending changes are saved across cron runs; alerts are sent on the first run +at or after the 10-minute mark. This delay applies only to status notifications: +battery/runtime alerts and automatic shutdowns still use the current readings. Battery-charge alerts start after 10 valid measurements have been collected; using their rolling mean prevents a single erroneous `0%` report from causing a false alarm. Raw measurements are still stored in the RRD and graphed. diff --git a/test_status_alerts.py b/test_status_alerts.py new file mode 100644 index 0000000..238fe6f --- /dev/null +++ b/test_status_alerts.py @@ -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() diff --git a/ups2rrd.py b/ups2rrd.py index 55f70c4..6562078 100644 --- a/ups2rrd.py +++ b/ups2rrd.py @@ -71,6 +71,7 @@ SHUTDOWN_COMMAND = "sudo /usr/sbin/shutdown -h now" REMOTE_SHUTDOWN_RUNTIME_SECONDS = 5 * 60 LOCAL_SHUTDOWN_RUNTIME_SECONDS = 200 BATTERY_CHARGE_MEAN_SAMPLES = 10 +STATUS_ALERT_DELAY_SECONDS = 10 * 60 GRAPH_COLOR_ARGS = [ "--color", "BACK#F7F5EFFF", @@ -486,6 +487,36 @@ def crossed_below(previous, current, threshold): ) +def process_status_alert(previous, current_status): + """Notify only after a new status persists for ten minutes across runs.""" + + # Older state files only have the latest observed status. + confirmed = previous.get("confirmed_status", previous.get("status")) + pending = previous.get("pending_status") + pending_since = previous.get("pending_status_since") + + if not confirmed: + confirmed = current_status + pending = pending_since = None + elif current_status == confirmed: + pending = pending_since = None + elif current_status: + now = time.time() + if current_status != pending or pending_since is None: + pending = current_status + pending_since = now + elif now - pending_since >= STATUS_ALERT_DELAY_SECONDS: + notify(f"UPS status changed: {confirmed} -> {current_status}") + confirmed = current_status + pending = pending_since = None + + return { + "confirmed_status": confirmed, + "pending_status": pending, + "pending_status_since": pending_since, + } + + def battery_charge_history(previous, current_charge): """Return up to the last 10 valid battery-charge measurements.""" @@ -585,18 +616,10 @@ def process_alerts(current): current_runtime = current.get("battery_runtime") shutdown_sent = previous.get("remote_shutdown_sent", False) local_shutdown_sent = previous.get("local_shutdown_sent", False) + status_alert_state = process_status_alert(previous, current_status) # An empty state is the initial baseline, not an alert condition. if previous: - if ( - previous_status - and current_status - and current_status != previous_status - ): - notify( - f"UPS status changed: {previous_status} -> {current_status}" - ) - if current_charge_mean is not None and ( previous_charge_mean is None or crossed_below(previous_charge_mean, current_charge_mean, 50) @@ -673,6 +696,7 @@ def process_alerts(current): # Keep the last valid reading when NUT temporarily omits a value. state = { + **status_alert_state, "status": current_status or previous_status, "battery_charge": ( current_charge if current_charge is not None else previous_charge