gandalf/tests/test_monitor.py

"""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