jaredandClaude Opus 5.5
e32499b7f6
Lint / Python (flake8) (pull_request) Successful in 2m6s
Lint / JS (eslint) (pull_request) Successful in 18s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 2m11s
Test / Python Tests (pytest) (pull_request) Successful in 2m20s
Lint / JS (eslint) (push) Successful in 10s
Lint / Python (flake8) (push) Successful in 1m58s
Lint / Notify on failure (push) Skipped
Security / Python Security (bandit) (push) Successful in 1m58s
Lint / Deploy (push) Successful in 4s
Test / Python Tests (pytest) (push) Successful in 58s
Modem and WAN: - Poll the BGW320 status pages read-only for light levels, PON and WAN state, with a WAN / Fiber panel, 7-day Rx trend, and alerts on WAN down, optical limits, and Rx drift - Read WAN outages and Smart Power restarts from the UniFi system log, since outages can be over between 5 minute polls Device pages: - Add device search to the dashboard and a page per host, switch, router, AP, PDU, or modem - Move all Link Debug data into expandable rows there; /links redirects - Collect UDM ports and UniFi device details (firmware, load, WAN, outlets) - Reorder the dashboard Monitoring fixes: - Alert when a configured host stops reporting: its series disappear rather than reporting down, so a dead host never raised an alert - Alert on Prometheus outages instead of blanking the dashboard - Treat a failed Pulse ping as unknown rather than host unreachable - Stop new switch ports reporting lifetime bytes as one interval's rate - Run database sessions in UTC: other server time zones shifted displayed ages and timed suppression expiries Also log every avatar 404 reason, add a favicon, and correct the WAN link labels (1G on Port 9, not 10G SFP+). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
376 lines
16 KiB
Python
376 lines
16 KiB
Python
"""Tests for monitor.py alert logic: database and external services are mocked."""
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
import modem as modem_api # noqa: E402
|
|
import monitor # noqa: E402
|
|
|
|
HOST_A, HOST_B = '10.0.0.1:9100', '10.0.0.2:9100'
|
|
|
|
|
|
@pytest.fixture
|
|
def db():
|
|
with patch.object(monitor, 'db') as mock_db:
|
|
mock_db.check_suppressed.return_value = False
|
|
mock_db.get_baseline.return_value = {}
|
|
# Second consecutive failure and no ticket yet, unless a test overrides it
|
|
mock_db.upsert_event.return_value = (1, False, 2, None)
|
|
mock_db.get_modem_rx_values.return_value = []
|
|
mock_db.get_state.return_value = None
|
|
yield mock_db
|
|
|
|
|
|
@pytest.fixture
|
|
def mon(db):
|
|
m = monitor.NetworkMonitor.__new__(monitor.NetworkMonitor)
|
|
m.cfg = {}
|
|
m.fail_thresh = 2
|
|
m.cluster_thresh = 2
|
|
m.cluster_name = 'test-cluster'
|
|
m.tickets = MagicMock()
|
|
m.tickets.create.return_value = '123'
|
|
m.prom = MagicMock(url='http://prom.test')
|
|
m._instance_map = {HOST_A: 'alpha', HOST_B: 'beta'}
|
|
m._last_iface_states = {}
|
|
m.modem = MagicMock(base_url='http://modem.test')
|
|
m.modem_name = 'att-modem'
|
|
m.modem_lan_ip = '192.168.1.254'
|
|
m.modem_fail_thresh = 2
|
|
m.rx_drift_db = 3.0
|
|
m.rx_drift_min_samples = 288
|
|
m._prev_modem_counters = None
|
|
m._prev_modem_time = 0.0
|
|
return m
|
|
|
|
|
|
def upserted(db):
|
|
return [(c.args[0], c.args[3], c.args[4]) for c in db.upsert_event.call_args_list]
|
|
|
|
|
|
def healthy_modem(rx=-16.9, rx_status=None):
|
|
def metric(value, alarm_low, alarm_high, warn_low, warn_high):
|
|
return {'value': value, 'alarm_low': alarm_low, 'alarm_high': alarm_high,
|
|
'warn_low': warn_low, 'warn_high': warn_high,
|
|
'alarm_flag': False, 'warn_flag': False}
|
|
return {
|
|
'broadband_state': 'Up', 'optical_state': 'Up', 'pon_state': 'OPERATION (O5)',
|
|
'counters': {'rx_bytes': 1},
|
|
'optics': {
|
|
'rx_power': metric(rx, -29.2, -9.0, -27.9, -10.0),
|
|
'tx_power': metric(3.3, -1.0, 5.9, 0.0, 4.9),
|
|
'temperature': metric(34.0, -50.0, 95.0, -45.0, 90.0),
|
|
},
|
|
}
|
|
|
|
|
|
class TestDeadHostDetection:
|
|
def test_host_missing_from_prometheus_alerts_and_tickets(self, mon, db):
|
|
mon._process_interfaces({HOST_A: {'eth0': True}}, [])
|
|
assert ('host_metrics_missing', 'beta', HOST_B) in upserted(db)
|
|
assert 'No metrics' in mon.tickets.create.call_args.args[0]
|
|
|
|
def test_host_reporting_resolves_the_event(self, mon, db):
|
|
mon._process_interfaces({HOST_A: {'eth0': True}, HOST_B: {'eth0': True}}, [])
|
|
db.resolve_event.assert_any_call('host_metrics_missing', 'beta', HOST_B)
|
|
assert not db.upsert_event.called
|
|
|
|
def test_no_ticket_before_threshold(self, mon, db):
|
|
db.upsert_event.return_value = (1, True, 1, None)
|
|
mon._process_interfaces({HOST_A: {'eth0': True}}, [])
|
|
assert not mon.tickets.create.called
|
|
|
|
def test_suppressed_host_gets_no_ticket(self, mon, db):
|
|
db.check_suppressed.return_value = True
|
|
mon._process_interfaces({HOST_A: {'eth0': True}}, [])
|
|
assert not mon.tickets.create.called
|
|
|
|
def test_missing_hosts_count_toward_cluster_alert(self, mon, db):
|
|
mon._instance_map['10.0.0.3:9100'] = 'gamma'
|
|
mon._process_interfaces({HOST_A: {'eth0': True}}, [])
|
|
assert ('cluster_network_issue', 'test-cluster', '') in upserted(db)
|
|
|
|
def test_dead_host_shows_as_down_on_dashboard(self, mon):
|
|
snap = mon._collect_snapshot({HOST_A: {'eth0': True}})
|
|
assert snap['hosts']['alpha']['status'] == 'up'
|
|
assert snap['hosts']['beta']['status'] == 'down'
|
|
|
|
|
|
class TestPrometheusOutage:
|
|
def test_empty_result_means_unavailable(self):
|
|
client = monitor.PrometheusClient('http://prom.test')
|
|
with patch.object(client, '_fetch', return_value=[]):
|
|
assert client.get_interface_states() is None
|
|
|
|
def test_failed_query_is_unavailable_not_empty(self):
|
|
client = monitor.PrometheusClient('http://prom.test')
|
|
with patch.object(monitor.requests, 'get', side_effect=OSError('boom')):
|
|
assert client.get_interface_states() is None
|
|
assert client.query('up') == []
|
|
|
|
def test_outage_alerts_about_prometheus_not_each_host(self, mon, db):
|
|
mon._process_prometheus_outage([])
|
|
assert upserted(db) == [('monitor_source_unavailable', 'prometheus', '')]
|
|
|
|
def test_dashboard_keeps_hosts_as_unknown(self, mon):
|
|
mon._last_iface_states = {HOST_A: {'eth0': True}}
|
|
snap = mon._collect_snapshot(None)
|
|
assert snap['hosts']['alpha']['status'] == 'unknown'
|
|
assert snap['hosts']['alpha']['interfaces'] == {'eth0': 'up'}
|
|
assert snap['hosts']['beta']['status'] == 'unknown'
|
|
|
|
|
|
class TestPingUnknown:
|
|
def _pulse(self, output):
|
|
pulse = monitor.PulseClient({})
|
|
pulse.run_command = MagicMock(return_value=output)
|
|
return pulse
|
|
|
|
def test_pulse_failure_is_unknown(self):
|
|
assert self._pulse(None).ping('10.0.0.1') is None
|
|
|
|
def test_reachability(self):
|
|
assert self._pulse('REACHABLE\n').ping('10.0.0.1') is True
|
|
assert self._pulse('UNREACHABLE\n').ping('10.0.0.1') is False
|
|
|
|
def test_unknown_neither_alerts_nor_resolves(self, mon, db):
|
|
mon.cfg = {'monitor': {'ping_hosts': [{'name': 'pbs', 'ip': '10.0.0.3'}]}}
|
|
mon._process_ping_hosts([], {'pbs': None})
|
|
assert not db.upsert_event.called
|
|
assert not db.resolve_event.called
|
|
|
|
def test_unreachable_still_alerts(self, mon, db):
|
|
mon.cfg = {'monitor': {'ping_hosts': [{'name': 'pbs', 'ip': '10.0.0.3'}]}}
|
|
mon._process_ping_hosts([], {'pbs': False})
|
|
assert ('host_unreachable', 'pbs', '10.0.0.3') in upserted(db)
|
|
|
|
|
|
class TestUnifiRates:
|
|
def _collector(self):
|
|
c = monitor.LinkStatsCollector.__new__(monitor.LinkStatsCollector)
|
|
c._prev_unifi_time = 100.0
|
|
c._prev_unifi = {'sw': {'ports': {'Port 1': {
|
|
'tx_bytes': 100, 'rx_bytes': 100, 'tx_errors': 0, 'rx_errors': 0, 'tx_dropped': 0, 'rx_dropped': 0,
|
|
}}}}
|
|
return c
|
|
|
|
def test_new_port_has_no_rate_spike(self):
|
|
port = {'tx_bytes': 10**12, 'rx_bytes': 10**12, 'tx_errors': 5, 'rx_errors': 5,
|
|
'tx_dropped': 5, 'rx_dropped': 5}
|
|
raw = {'sw': {'ip': '', 'model': '', 'ports': {'Port 1': dict(port, tx_bytes=300), 'Port 2': port}}}
|
|
result = self._collector()._compute_unifi_rates(raw, 200.0)['sw']['ports']
|
|
assert result['Port 2']['tx_bytes_rate'] is None
|
|
assert result['Port 1']['tx_bytes_rate'] == 2.0
|
|
|
|
|
|
class TestModemAlerts:
|
|
def test_healthy_link_raises_nothing(self, mon, db):
|
|
mon._process_modem(healthy_modem(), [])
|
|
assert not db.upsert_event.called
|
|
assert not mon.tickets.create.called
|
|
|
|
def test_wan_down_is_p1(self, mon, db):
|
|
data = healthy_modem()
|
|
data['broadband_state'] = 'Down'
|
|
mon._process_modem(data, [])
|
|
assert ('wan_down', 'att-modem', '') in upserted(db)
|
|
assert mon.tickets.create.call_args.kwargs['priority'] == '1'
|
|
|
|
def test_cut_fiber_opens_only_one_ticket(self, mon, db):
|
|
data = healthy_modem(rx=-40.0)
|
|
data['optical_state'] = 'Down'
|
|
mon._process_modem(data, [])
|
|
assert [e[0] for e in upserted(db)] == ['wan_down']
|
|
|
|
def test_low_rx_warning(self, mon, db):
|
|
mon._process_modem(healthy_modem(rx=-28.0), [])
|
|
assert ('fiber_signal_warning', 'att-modem', 'rx_power') in upserted(db)
|
|
assert mon.tickets.create.call_args.kwargs['priority'] == '3'
|
|
db.resolve_event.assert_any_call('fiber_signal_alarm', 'att-modem', 'rx_power')
|
|
|
|
def test_low_rx_alarm_with_link_up(self, mon, db):
|
|
mon._process_modem(healthy_modem(rx=-30.0), [])
|
|
assert ('fiber_signal_alarm', 'att-modem', 'rx_power') in upserted(db)
|
|
db.resolve_event.assert_any_call('fiber_signal_warning', 'att-modem', 'rx_power')
|
|
|
|
def test_recovery_resolves_signal_events(self, mon, db):
|
|
mon._process_modem(healthy_modem(), [])
|
|
db.resolve_event.assert_any_call('fiber_signal_alarm', 'att-modem', 'rx_power')
|
|
db.resolve_event.assert_any_call('fiber_signal_warning', 'att-modem', 'rx_power')
|
|
db.resolve_event.assert_any_call('wan_down', 'att-modem', '')
|
|
|
|
def test_suppression_blocks_ticket(self, mon, db):
|
|
db.check_suppressed.return_value = True
|
|
mon._process_modem(healthy_modem(rx=-30.0), [])
|
|
assert upserted(db)
|
|
assert not mon.tickets.create.called
|
|
|
|
def test_drift_needs_enough_history(self, mon, db):
|
|
db.get_modem_rx_values.return_value = [-17.0] * 10
|
|
mon._process_modem(healthy_modem(rx=-25.0), [])
|
|
assert not any(e[0] == 'fiber_rx_drift' for e in upserted(db))
|
|
|
|
def test_drift_alerts_inside_the_modem_limits(self, mon, db):
|
|
db.get_modem_rx_values.return_value = [-17.0] * 300
|
|
mon._process_modem(healthy_modem(rx=-21.0), [])
|
|
assert ('fiber_rx_drift', 'att-modem', 'rx_power') in upserted(db)
|
|
assert 'fiber_signal_warning' not in [e[0] for e in upserted(db)]
|
|
|
|
def test_small_change_is_not_drift(self, mon, db):
|
|
db.get_modem_rx_values.return_value = [-17.0] * 300
|
|
mon._process_modem(healthy_modem(rx=-18.0), [])
|
|
assert not db.upsert_event.called
|
|
db.resolve_event.assert_any_call('fiber_rx_drift', 'att-modem', 'rx_power')
|
|
|
|
def test_stronger_signal_is_not_drift(self, mon, db):
|
|
db.get_modem_rx_values.return_value = [-17.0] * 300
|
|
mon._process_modem(healthy_modem(rx=-12.0), [])
|
|
assert not db.upsert_event.called
|
|
|
|
|
|
class TestModemCollection:
|
|
def test_unreachable_keeps_link_events_open(self, mon, db):
|
|
mon.modem.poll.side_effect = modem_api.ModemUnreachable('timeout')
|
|
mon._collect_modem([])
|
|
assert upserted(db) == [('modem_unreachable', 'att-modem', '')]
|
|
assert db.set_state.call_args.args[1]['reachable'] is False
|
|
assert not db.record_modem_sample.called
|
|
resolved = [c.args[0] for c in db.resolve_event.call_args_list]
|
|
assert 'wan_down' not in resolved and 'fiber_signal_alarm' not in resolved
|
|
|
|
def test_unreadable_page_is_a_warning_not_an_outage(self, mon, db):
|
|
mon.modem.poll.side_effect = modem_api.ModemDataInvalid('login required')
|
|
mon._collect_modem([])
|
|
assert upserted(db) == [('modem_data_invalid', 'att-modem', '')]
|
|
assert mon.tickets.create.call_args.kwargs['priority'] == '3'
|
|
|
|
def test_success_stores_status_and_sample(self, mon, db):
|
|
mon.modem.poll.return_value = healthy_modem()
|
|
mon._collect_modem([])
|
|
assert db.set_state.call_args.args[0] == 'modem_stats'
|
|
assert db.set_state.call_args.args[1]['reachable'] is True
|
|
db.record_modem_sample.assert_called_once_with(-16.9, 3.3, 34.0)
|
|
|
|
def test_drift_baseline_excludes_the_current_reading(self, mon, db):
|
|
order = []
|
|
db.get_modem_rx_values.side_effect = lambda days: order.append('read') or []
|
|
db.record_modem_sample.side_effect = lambda *a: order.append('write')
|
|
mon.modem.poll.return_value = healthy_modem()
|
|
mon._collect_modem([])
|
|
assert order == ['read', 'write']
|
|
|
|
|
|
class TestWanLog:
|
|
def _mon(self, mon, events):
|
|
mon.unifi = MagicMock()
|
|
mon.unifi.get_system_events.return_value = events
|
|
return mon
|
|
|
|
def _state(self, db, **values):
|
|
db.get_state.side_effect = lambda key, default=None: values.get(key, default)
|
|
|
|
def test_first_run_starts_from_now_without_tickets(self, mon, db):
|
|
self._state(db)
|
|
self._mon(mon, [])._process_wan_log([])
|
|
assert db.set_state.call_args.args[0] == 'unifi_log_cursor'
|
|
assert not mon.unifi.get_system_events.called
|
|
|
|
def test_log_failure_keeps_the_cursor(self, mon, db):
|
|
self._state(db, unifi_log_cursor='1000')
|
|
self._mon(mon, None)._process_wan_log([])
|
|
assert not db.set_state.called
|
|
|
|
def test_rpdu_restart_opens_a_ticket_and_resolves(self, mon, db):
|
|
self._state(db, unifi_log_cursor=str(int(time.time() * 1000)))
|
|
db.upsert_event.return_value = (9, True, 1, None)
|
|
self._mon(mon, [{'key': modem_api.LOG_MODEM_RESTARTED, 'ts': time.time(), 'device': 'RPDU', 'wan': ''}])
|
|
mon._process_wan_log([])
|
|
assert 'power-cycled' in mon.tickets.create.call_args.args[0]
|
|
db.resolve_event.assert_called_with('modem_power_cycled', 'att-modem', '')
|
|
|
|
def test_short_outage_only_records_history(self, mon, db):
|
|
now = time.time()
|
|
self._state(db, unifi_log_cursor=str(int(now * 1000)))
|
|
self._mon(mon, [
|
|
{'key': modem_api.LOG_WAN_FAILED, 'ts': now, 'device': 'UDM', 'wan': 'WAN1'},
|
|
{'key': modem_api.LOG_WAN_RESTORED, 'ts': now + 90, 'device': 'UDM', 'wan': 'WAN1'},
|
|
])
|
|
mon._process_wan_log([])
|
|
assert not mon.tickets.create.called
|
|
saved = dict((c.args[0], c.args[1]) for c in db.set_state.call_args_list)
|
|
assert saved['wan_history']['outages'][0]['restored_at'] == now + 90
|
|
|
|
def test_unstable_wan_tickets_once_a_day(self, mon, db):
|
|
now = time.time()
|
|
history = json.dumps({'outages': [], 'restarts': [], 'unstable_ticketed_at': now - 3600})
|
|
self._state(db, unifi_log_cursor=str(int(now * 1000)), wan_history=history)
|
|
self._mon(mon, [{'key': modem_api.LOG_WAN_UNSTABLE, 'ts': now, 'device': 'UDM', 'wan': 'WAN1'}])
|
|
mon._process_wan_log([])
|
|
assert not mon.tickets.create.called
|
|
|
|
def test_unstable_wan_tickets_after_a_day(self, mon, db):
|
|
now = time.time()
|
|
self._state(db, unifi_log_cursor=str(int(now * 1000)))
|
|
self._mon(mon, [{'key': modem_api.LOG_WAN_UNSTABLE, 'ts': now, 'device': 'UDM', 'wan': 'WAN1'}])
|
|
mon._process_wan_log([])
|
|
assert 'unstable' in mon.tickets.create.call_args.args[0]
|
|
|
|
def test_suppression_blocks_one_shot_ticket(self, mon, db):
|
|
now = time.time()
|
|
db.check_suppressed.return_value = True
|
|
self._state(db, unifi_log_cursor=str(int(now * 1000)))
|
|
self._mon(mon, [{'key': modem_api.LOG_MODEM_RESTARTED, 'ts': now, 'device': 'RPDU', 'wan': ''}])
|
|
mon._process_wan_log([])
|
|
assert not mon.tickets.create.called
|
|
|
|
def test_uptime_going_backwards_records_a_restart(self, mon, db):
|
|
self._state(db, modem_stats=json.dumps({'uptime_s': 5000}))
|
|
mon._note_modem_restart({'uptime_s': 60})
|
|
saved = dict((c.args[0], c.args[1]) for c in db.set_state.call_args_list)
|
|
assert len(saved['wan_history']['restarts']) == 1
|
|
|
|
def test_uptime_going_forward_records_nothing(self, mon, db):
|
|
self._state(db, modem_stats=json.dumps({'uptime_s': 5000}))
|
|
mon._note_modem_restart({'uptime_s': 5300})
|
|
assert not db.set_state.called
|
|
|
|
|
|
class TestUnifiDeviceDetail:
|
|
def test_parses_string_stats_and_outlets(self):
|
|
d = monitor.UnifiClient._device_detail({
|
|
'version': '7.0', 'uptime': 10, 'system-stats': {'cpu': ' 2.4 ', 'mem': 'bad'},
|
|
'outlet_table': [{'index': 13, 'name': 'AT&T Modem', 'relay_state': True,
|
|
'cycle_enabled': True, 'outlet_power': '0.000'}],
|
|
'wan1': {'up': True, 'speed': 1000, 'latency': 15, 'availability': 100},
|
|
})
|
|
assert d['cpu_pct'] == 2.4 and d['mem_pct'] is None
|
|
assert d['outlets'][0]['auto_power_cycle'] is True and d['outlets'][0]['power_w'] == 0.0
|
|
assert d['wan']['wan1']['speed_mbps'] == 1000
|
|
assert 'wan2' not in d['wan']
|
|
|
|
def test_system_log_filters_and_paginates(self):
|
|
c = monitor.UnifiClient({'controller': 'https://u.test', 'api_key': 'k'})
|
|
page = MagicMock()
|
|
page.json.side_effect = [
|
|
{'data': [
|
|
{'key': modem_api.LOG_MODEM_RESTARTED, 'timestamp': 2000, 'parameters': {'DEVICE': {'name': 'RPDU'}}},
|
|
{'key': 'CLIENT_CONNECTED_WIRED_2', 'timestamp': 2000},
|
|
{'key': modem_api.LOG_WAN_FAILED, 'timestamp': 500},
|
|
]},
|
|
{'data': []},
|
|
]
|
|
c.session.post = MagicMock(return_value=page)
|
|
got = c.get_system_events(1000, modem_api.WAN_LOG_KEYS)
|
|
assert got == [{'key': modem_api.LOG_MODEM_RESTARTED, 'ts': 2.0, 'device': 'RPDU', 'wan': ''}]
|
|
|
|
def test_system_log_failure_is_none_not_empty(self):
|
|
c = monitor.UnifiClient({'controller': 'https://u.test', 'api_key': 'k'})
|
|
c.session.post = MagicMock(side_effect=OSError('boom'))
|
|
assert c.get_system_events(0, modem_api.WAN_LOG_KEYS) is None
|