import json import os import stat import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from io import StringIO from pathlib import Path from types import SimpleNamespace from unittest.mock import ANY, AsyncMock, MagicMock, call, patch import wlc_monitor def sample_state(**changes): state = { "ok": True, "sys_name": "test-wlc", "sys_descr": "Synthetic controller", "sys_uptime": 2_000_000, "run_last_changed": 1_000_000, "run_last_saved": 1_000_000, "aps": {}, "consecutive_failures": 0, "last_poll": "2026-01-01T00:00:00+00:00", } state.update(changes) return state class AlertConfig: fail_threshold = 2 unsaved_ticks = 60 * 60 * 100 host = "192.0.2.10" class CounterTests(unittest.TestCase): def test_increasing_counter_is_not_reset(self): self.assertFalse(wlc_monitor.counter_reset(100, 200, 1)) def test_small_decrease_is_reset(self): self.assertTrue(wlc_monitor.counter_reset(50_000, 100, 60)) def test_32_bit_wrap_is_not_reset(self): self.assertFalse( wlc_monitor.counter_reset(2**32 - 100, 200, elapsed_s=5)) class AlertTests(unittest.TestCase): def test_first_poll_is_silent(self): self.assertEqual( wlc_monitor.build_alerts({}, sample_state(), AlertConfig()), []) def test_unsaved_config_alerts_once_until_saved(self): prev = sample_state( run_last_changed=1_000_000, run_last_saved=900_000) cur = sample_state( sys_uptime=2_000_000, run_last_changed=1_000_000, run_last_saved=900_000, ) first = wlc_monitor.build_alerts(prev, cur, AlertConfig()) self.assertEqual([a[1] for a in first], [ "test-wlc has unsaved config"]) self.assertTrue(cur["_unsaved_alerted"]) later = sample_state( sys_uptime=2_030_000, run_last_changed=1_000_000, run_last_saved=900_000, ) self.assertEqual( wlc_monitor.build_alerts(cur, later, AlertConfig()), []) saved = sample_state( sys_uptime=2_060_000, run_last_changed=1_000_000, run_last_saved=1_000_000, ) wlc_monitor.build_alerts(later, saved, AlertConfig()) self.assertFalse(saved["_unsaved_alerted"]) class StateTests(unittest.TestCase): def test_state_is_written_atomically_with_private_permissions(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "state.json" wlc_monitor.save_state(path, {"ok": True}) self.assertEqual(json.loads(path.read_text()), {"ok": True}) self.assertEqual( stat.S_IMODE(path.stat().st_mode), 0o600) def test_naive_or_invalid_timestamp_is_tolerated(self): self.assertIsNone( wlc_monitor.elapsed_since({"last_poll": "2026-01-01T00:00:00"})) self.assertIsNone( wlc_monitor.elapsed_since({"last_poll": "not-a-timestamp"})) def test_non_object_state_is_ignored(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "state.json" path.write_text("[]") self.assertEqual(wlc_monitor.load_state(path), {}) class ConfigTests(unittest.TestCase): required_env = { "WLC_SNMP_AUTH": "synthetic-auth", "WLC_SNMP_PRIV": "synthetic-privacy", } def write_config(self, directory): path = Path(directory) / "config.ini" path.write_text( "[wlc]\n" "host = 192.0.2.10\n" "snmp_user = test\n" "snmp_auth =\n" "snmp_priv =\n" "[mail]\n" "smtp_host = smtp.example.com\n" "from = monitor@example.com\n" "to = ops@example.com\n" "[monitor]\n" "state_file = state.json\n" ) path.chmod(0o600) return path def test_environment_overrides_file(self): with tempfile.TemporaryDirectory() as directory: path = self.write_config(directory) with patch.dict( os.environ, {**self.required_env, "WLC_HOST": "198.51.100.20"}, clear=True, ): cfg = wlc_monitor.Cfg(path) self.assertEqual(cfg.host, "198.51.100.20") def test_invalid_interval_is_rejected(self): with tempfile.TemporaryDirectory() as directory: path = self.write_config(directory) with patch.dict( os.environ, {**self.required_env, "MONITOR_INTERVAL": "0"}, clear=True, ): with self.assertRaises(SystemExit): wlc_monitor.Cfg(path) def test_negative_unsaved_threshold_is_rejected(self): with tempfile.TemporaryDirectory() as directory: path = self.write_config(directory) with patch.dict( os.environ, { **self.required_env, "MONITOR_UNSAVED_MINUTES": "-1", }, clear=True, ): with self.assertRaises(SystemExit): wlc_monitor.Cfg(path) def test_smtp_authentication_requires_encryption(self): with tempfile.TemporaryDirectory() as directory: path = self.write_config(directory) with patch.dict( os.environ, { **self.required_env, "MAIL_SMTP_SECURITY": "none", "MAIL_SMTP_USER": "test-user", "MAIL_SMTP_PASSWORD": "synthetic-password", }, clear=True, ): with self.assertRaises(SystemExit): wlc_monitor.Cfg(path) def test_secret_can_be_read_from_file(self): with tempfile.TemporaryDirectory() as directory: path = self.write_config(directory) secret = Path(directory) / "auth" secret.write_text("from-file\n") with patch.dict( os.environ, { "WLC_SNMP_PRIV": self.required_env["WLC_SNMP_PRIV"], "WLC_SNMP_AUTH_FILE": str(secret), }, clear=True, ): cfg = wlc_monitor.Cfg(path) self.assertEqual(cfg.auth, "from-file") class DeliveryTests(unittest.TestCase): def test_failed_transition_email_is_retried(self): with tempfile.TemporaryDirectory() as directory: state_path = Path(directory) / "state.json" previous = sample_state( aps={ "ap-one": { "index": "1", "ap_uptime": 1_000, "assoc_uptime": 900, } } ) wlc_monitor.save_state(state_path, previous) current = sample_state(sys_uptime=2_030_000) cfg = SimpleNamespace( state_file=state_path, host="192.0.2.10", port=161, user="test", auth="auth", priv="priv", auth_protocol="sha256", timeout=1, retries=0, fail_threshold=2, unsaved_ticks=360_000, mail_from="monitor@example.com", mail_to=["ops@example.com"], subject_prefix="[TEST]", smtp_host="smtp.example.com", smtp_port=587, smtp_security="starttls", smtp_user=None, smtp_password=None, ) args = SimpleNamespace(show=False, dry_run=False, no_save=False) poller = SimpleNamespace(poll=AsyncMock(return_value=current)) with patch.object(wlc_monitor, "Poller", return_value=poller), \ patch.object( wlc_monitor, "send_email", side_effect=OSError("temporary failure"), ), redirect_stdout(StringIO()), redirect_stderr(StringIO()): self.assertEqual(wlc_monitor.run_once(cfg, args), 1) failed_state = json.loads(state_path.read_text()) self.assertEqual(len(failed_state["_pending_emails"]), 1) poller.poll = AsyncMock(return_value=sample_state( sys_uptime=2_060_000)) with patch.object(wlc_monitor, "Poller", return_value=poller), \ patch.object(wlc_monitor, "send_email") as send, \ redirect_stdout(StringIO()): self.assertEqual(wlc_monitor.run_once(cfg, args), 0) send.assert_called_once() recovered_state = json.loads(state_path.read_text()) self.assertNotIn("_pending_emails", recovered_state) def test_outbox_stops_after_first_failure_to_preserve_order(self): with tempfile.TemporaryDirectory() as directory: state_path = Path(directory) / "state.json" previous = sample_state( ok=False, consecutive_failures=2, _pending_emails=[ {"subject": "OUTAGE", "body": "controller unavailable"} ], ) wlc_monitor.save_state(state_path, previous) cfg = SimpleNamespace( state_file=state_path, host="192.0.2.10", port=161, user="test", auth="auth", priv="priv", auth_protocol="sha256", timeout=1, retries=0, fail_threshold=2, unsaved_ticks=360_000, mail_from="monitor@example.com", mail_to=["ops@example.com"], subject_prefix="[TEST]", smtp_host="smtp.example.com", smtp_port=587, smtp_security="starttls", smtp_user=None, smtp_password=None, ) args = SimpleNamespace(show=False, dry_run=False, no_save=False) poller = SimpleNamespace( poll=AsyncMock(return_value=sample_state())) with patch.object(wlc_monitor, "Poller", return_value=poller), \ patch.object( wlc_monitor, "send_email", side_effect=OSError("still unavailable"), ) as send, redirect_stdout(StringIO()), \ redirect_stderr(StringIO()): self.assertEqual(wlc_monitor.run_once(cfg, args), 1) send.assert_called_once() saved = json.loads(state_path.read_text()) queued = saved["_pending_emails"] self.assertEqual(len(queued), 2) self.assertEqual(queued[0]["subject"], "OUTAGE") self.assertIn("RECOVERY", queued[1]["subject"]) class EmailTests(unittest.TestCase): def config(self, security, user=None, password=None): return SimpleNamespace( mail_from="monitor@example.com", mail_to=["ops@example.com"], smtp_host="smtp.example.com", smtp_port=587, smtp_security=security, smtp_user=user, smtp_password=password, ) def test_starttls_happens_before_authentication(self): server = MagicMock() session = server.__enter__.return_value context = object() with patch.object(wlc_monitor.ssl, "create_default_context", return_value=context), \ patch.object(wlc_monitor.smtplib, "SMTP", return_value=server): wlc_monitor.send_email( self.config("starttls", "user", "password"), "subject", "body", ) self.assertEqual( session.method_calls, [ call.starttls(context=context), call.login("user", "password"), call.send_message(ANY), ], ) def test_implicit_tls_uses_smtp_ssl(self): server = MagicMock() context = object() with patch.object(wlc_monitor.ssl, "create_default_context", return_value=context), \ patch.object( wlc_monitor.smtplib, "SMTP_SSL", return_value=server ) as smtp_ssl: wlc_monitor.send_email( self.config("ssl"), "subject", "body") smtp_ssl.assert_called_once_with( "smtp.example.com", 587, timeout=15, context=context) server.__enter__.return_value.send_message.assert_called_once() def test_plain_relay_does_not_start_tls_or_authenticate(self): server = MagicMock() session = server.__enter__.return_value with patch.object(wlc_monitor.smtplib, "SMTP", return_value=server): wlc_monitor.send_email( self.config("none"), "subject", "body") session.starttls.assert_not_called() session.login.assert_not_called() session.send_message.assert_called_once() if __name__ == "__main__": unittest.main()