374 lines
13 KiB
Python
374 lines
13 KiB
Python
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()
|