#!/usr/bin/env python3 """Unit tests for the voice intake watchdog. Run with: python3 -m unittest discover -s tests -v (No third-party packages needed; paho is imported lazily by watchdog.py.) """ import os import sys import time import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from watchdog import ( # noqa: E402 AlertState, Config, Layer, Snapshot, advance, build_message, classify, ) NOW = 1_700_000_000.0 def snap(**kw): base = dict( now=NOW, broker_reachable=True, bridge_running=True, bridge_subscribed=True, last_audio_at=NOW - 10, started_at=NOW - 3600, presence=None, ) base.update(kw) return Snapshot(**base) class ClassifyTests(unittest.TestCase): def test_healthy(self): self.assertEqual(classify(snap(), 900), Layer.HEALTHY) def test_broker_down_wins_over_everything(self): s = snap(broker_reachable=False, bridge_running=False, last_audio_at=NOW - 9999) self.assertEqual(classify(s, 900), Layer.BROKER) def test_bridge_down(self): s = snap(bridge_running=False) self.assertEqual(classify(s, 900), Layer.BRIDGE) def test_bridge_running_but_not_subscribed(self): s = snap(bridge_running=True, bridge_subscribed=False) self.assertEqual(classify(s, 900), Layer.BRIDGE) def test_publisher_silent_after_timeout(self): s = snap(last_audio_at=NOW - 901) self.assertEqual(classify(s, 900), Layer.PUBLISHER) def test_just_inside_timeout_is_healthy(self): s = snap(last_audio_at=NOW - 899) self.assertEqual(classify(s, 900), Layer.HEALTHY) def test_presence_offline_is_immediate_publisher_failure(self): s = snap(presence="offline", last_audio_at=NOW - 1) self.assertEqual(classify(s, 900), Layer.PUBLISHER) def test_no_audio_since_start_uses_start_time(self): s = snap(last_audio_at=None, started_at=NOW - 1000) self.assertEqual(classify(s, 900, require_evidence=False), Layer.PUBLISHER) def test_unarmed_silence_is_not_a_failure(self): # No heartbeat and no retained presence: the heartbeat firmware is not # flashed yet, so silence must not be reported as a dead publisher. s = snap(last_audio_at=None, presence=None, started_at=NOW - 100000) self.assertEqual(classify(s, 900), Layer.HEALTHY) def test_presence_online_arms_the_silence_check(self): s = snap(last_audio_at=None, presence="online", started_at=NOW - 1000) self.assertEqual(classify(s, 900), Layer.PUBLISHER) class AdvanceTests(unittest.TestCase): def setUp(self): self.cfg = Config(silence_timeout=900, repeat_interval=86400) def test_first_failure_alerts_once(self): state = AlertState() alerts = advance(state, snap(bridge_running=False), self.cfg) self.assertEqual([a.kind for a in alerts], ["alert"]) self.assertEqual(alerts[0].layer, "bridge") self.assertEqual(state.layer, "bridge") def test_no_repeat_before_interval(self): state = AlertState(layer="bridge", first_seen_at=NOW, last_alert_at=NOW) alerts = advance( state, snap(now=NOW + 60, bridge_running=False), self.cfg ) self.assertEqual(alerts, []) def test_repeat_after_interval(self): state = AlertState(layer="bridge", first_seen_at=NOW, last_alert_at=NOW) alerts = advance( state, snap(now=NOW + 86400, bridge_running=False), self.cfg, ) self.assertEqual([a.kind for a in alerts], ["repeat"]) self.assertEqual(state.last_alert_at, NOW + 86400) def test_daily_repeats_keep_firing(self): state = AlertState(layer="publisher", first_seen_at=NOW, last_alert_at=NOW) fires = [] for day in (1, 2, 3): alerts = advance( state, snap(now=NOW + day * 86400, last_audio_at=NOW - 5000), self.cfg, ) fires.extend(a.kind for a in alerts) self.assertEqual(fires, ["repeat", "repeat", "repeat"]) def test_recovery_clears_state_and_notifies_once(self): state = AlertState(layer="bridge", first_seen_at=NOW, last_alert_at=NOW) alerts = advance(state, snap(now=NOW + 120), self.cfg) self.assertEqual([a.kind for a in alerts], ["recovery"]) self.assertIsNone(state.layer) self.assertIsNone(state.last_alert_at) # A second healthy cycle is silent. self.assertEqual(advance(state, snap(now=NOW + 150), self.cfg), []) def test_layer_change_is_a_new_alert(self): state = AlertState(layer="publisher", first_seen_at=NOW, last_alert_at=NOW) alerts = advance( state, snap(now=NOW + 60, bridge_running=False), self.cfg, ) self.assertEqual([a.kind for a in alerts], ["alert"]) self.assertEqual(alerts[0].layer, "bridge") class MessageTests(unittest.TestCase): def setUp(self): self.cfg = Config( apprise_tag="sam_ntfy", apprise_tag_repeat="sam", silence_timeout=900, ) def test_broker_names_broker_and_not_esp32(self): title, body, tag = build_message( __import__("watchdog").Alert("alert", "broker", NOW), self.cfg ) self.assertIn("broker", title.lower()) self.assertIn("mosquitto", body.lower()) self.assertIn("Do NOT power-cycle the ESP32", body) self.assertEqual(tag, "sam_ntfy") def test_bridge_names_bridge_and_not_esp32(self): title, body, _ = build_message( __import__("watchdog").Alert("alert", "bridge", NOW), self.cfg ) self.assertIn("voice_bridge", title) self.assertIn("docker compose up -d", body) self.assertIn("Do NOT power-cycle the ESP32", body) def test_publisher_says_power_cycle(self): title, body, _ = build_message( __import__("watchdog").Alert("alert", "publisher", NOW), self.cfg ) self.assertIn("ESP32", title) self.assertIn("try a power-cycle", body.lower()) def test_repeat_escalates_tag(self): _, _, tag = build_message( __import__("watchdog").Alert("repeat", "publisher", NOW), self.cfg ) self.assertEqual(tag, "sam") def test_recovery_uses_primary_tag(self): _, _, tag = build_message( __import__("watchdog").Alert("recovery", "publisher", NOW), self.cfg ) self.assertEqual(tag, "sam_ntfy") class DockerTimeTests(unittest.TestCase): def test_parses_rfc3339_nano(self): import datetime as dt from watchdog import _parse_docker_time expected = dt.datetime( 2026, 10, 14, 9, 3, 11, 123456, tzinfo=dt.timezone.utc ).timestamp() ts = _parse_docker_time("2026-10-14T09:03:11.123456789Z") self.assertAlmostEqual(ts, expected, places=3) def test_bad_value_falls_back_to_now(self): from watchdog import _parse_docker_time self.assertAlmostEqual(_parse_docker_time("not-a-time"), time.time(), delta=5) if __name__ == "__main__": unittest.main()