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 datetime import timedelta
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
        mock_db.event_stats.return_value = {'last_7d': 0, 'last_30d': 0,
                                            'last_first_seen': None, 'last_resolved_at': None}
        mock_db.get_events_for_target.return_value = []
        mock_db.cutsheet_load.return_value = {'items': [], 'cables': [], 'port_labels': []}
        mock_db.get_pending_recovery_notes.return_value = []
        mock_db.get_device_link.return_value = ''
        yield mock_db


@pytest.fixture
def mon(db):
    m = monitor.NetworkMonitor.__new__(monitor.NetworkMonitor)
    m.cfg = {}
    m.fail_thresh = 2
    m.host_fail_thresh = 5
    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):
        db.upsert_event.return_value = (1, False, 5, None)
        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_reboot_length_outage_is_recorded_without_a_ticket(self, mon, db):
        db.upsert_event.return_value = (1, False, 4, None)
        mon._process_interfaces({HOST_A: {'eth0': True}}, [])
        assert ('host_metrics_missing', 'beta', HOST_B) in upserted(db)
        assert not mon.tickets.create.called

    def test_interface_down_keeps_the_shorter_threshold(self, mon, db):
        db.get_baseline.return_value = {'alpha': {'eth0': 'up'}, 'beta': {'eth0': 'up'}}
        mon._process_interfaces({HOST_A: {'eth0': False}, HOST_B: {'eth0': True}}, [])
        assert 'Interface eth0 link-down' in mon.tickets.create.call_args.args[0]

    def test_suppressed_host_gets_no_ticket(self, mon, db):
        db.upsert_event.return_value = (1, False, 5, None)
        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_unused_interface_never_alerts(self, mon, db):
        mon._unused = {'alpha': {'eth1': {'reason': 'spare'}}}
        db.get_baseline.return_value = {'alpha': {'eth0': 'up', 'eth1': 'up'}, 'beta': {'eth0': 'up'}}
        mon._process_interfaces({HOST_A: {'eth0': True, 'eth1': False}, HOST_B: {'eth0': True}}, [])
        assert ('interface_down', 'alpha', 'eth1') not in upserted(db)
        assert not mon.tickets.create.called

    def test_unused_interface_does_not_degrade_the_host(self, mon):
        mon._unused = {'alpha': {'eth1': {}}}
        snap = mon._collect_snapshot({HOST_A: {'eth0': True, 'eth1': False}, HOST_B: {'eth0': True}})
        assert snap['hosts']['alpha']['status'] == 'up'
        assert snap['hosts']['alpha']['interfaces'] == {'eth0': 'up', 'eth1': 'unused'}

    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)

    def test_ping_host_ticket_waits_out_a_reboot(self, mon, db):
        mon.cfg = {'monitor': {'ping_hosts': [{'name': 'pbs', 'ip': '10.0.0.3'}]}}
        db.upsert_event.return_value = (1, False, 4, None)
        mon._process_ping_hosts([], {'pbs': False})
        assert not mon.tickets.create.called
        db.upsert_event.return_value = (1, False, 5, None)
        mon._process_ping_hosts([], {'pbs': False})
        assert 'Host unreachable' in mon.tickets.create.call_args.args[0]


class TestUnifiRates:
    def _collector(self):
        c = monitor.LinkStatsCollector.__new__(monitor.LinkStatsCollector)
        c._prev_unifi_time = 100.0
        c._prev_unifi = {'sw': {'ports': {'Port 1': {
            'port_idx': 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, port_idx=1, tx_bytes=300), 'Port 2': dict(port, port_idx=2)}}}
        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

    def test_port_renamed_in_unifi_keeps_its_rates(self):
        port = {'port_idx': 1, 'tx_bytes': 300, 'rx_bytes': 100, 'tx_errors': 0, 'rx_errors': 0,
                'tx_dropped': 0, 'rx_dropped': 0}
        raw = {'sw': {'ip': '', 'model': '', 'ports': {'Office TV': port}}}
        result = self._collector()._compute_unifi_rates(raw, 200.0)['sw']['ports']
        assert result['Office TV']['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


class TestSwitchPortCollection:
    def test_ports_sharing_a_name_are_both_kept(self):
        c = monitor.UnifiClient({'controller': 'https://u.test', 'api_key': 'k'})
        resp = MagicMock()
        resp.json.return_value = {'data': [{'type': 'usw', 'name': 'sw', 'port_table': [
            {'port_idx': 13, 'name': 'Office TV', 'up': False},
            {'port_idx': 14, 'name': 'Office TV', 'up': True},
            {'port_idx': 15},
        ]}]}
        c.session.get = MagicMock(return_value=resp)
        ports = c.get_switch_ports()['sw']['ports']
        assert {n: p['port_idx'] for n, p in ports.items()} == {
            'Office TV': 13, 'Office TV (Port 14)': 14, 'Port 15': 15,
        }


class TestSfpModules:
    OPTIC = {'sfp_found': True, 'sfp_compliance': '10G BASE-SR', 'sfp_part': 'SFP-10G-SR', 'sfp_vendor': 'OEM',
             'sfp_serial': 'YC1', 'sfp_rxpower': '-2.45', 'sfp_txpower': '-2.56', 'sfp_temperature': '40.6',
             'sfp_voltage': '3.318', 'sfp_current': '7.564', 'sfp_rxfault': False, 'sfp_txfault': False}

    def test_port_without_a_cage_has_no_module_info(self):
        assert monitor.UnifiClient._sfp_detail({'port_idx': 1}) is None

    def test_empty_cage(self):
        assert monitor.UnifiClient._sfp_detail({'sfp_found': False}) == {'found': False}

    def test_optic_with_readings(self):
        d = monitor.UnifiClient._sfp_detail(self.OPTIC)
        assert d['kind'] == 'optic' and d['rx_power_dbm'] == -2.45 and d['current_ma'] == 7.564
        assert d['rx_fault'] is False

    def test_dac_by_compliance_or_part_number(self):
        assert monitor.UnifiClient._sfp_detail({'sfp_found': True, 'sfp_compliance': 'DAC'})['kind'] == 'dac'
        d = monitor.UnifiClient._sfp_detail({'sfp_found': True, 'sfp_compliance': 'Unknown',
                                             'sfp_part': 'SFP-H10GB-CU3M'})
        assert d['kind'] == 'dac' and d['compliance'] == '' and d['rx_power_dbm'] is None

    def test_unidentified_module_and_bad_readings(self):
        d = monitor.UnifiClient._sfp_detail({'sfp_found': True, 'sfp_part': 'L56SF017-SD-R',
                                             'sfp_rxpower': 'n/a'})
        assert d['kind'] == '' and d['rx_power_dbm'] is None

    def test_udm_fault_flag_spelling(self):
        d = monitor.UnifiClient._sfp_detail({'sfp_found': True, 'sfp_rx_los': True, 'sfp_tx_fault': True})
        assert d['rx_fault'] is True and d['tx_fault'] is True

    def test_dac_ends_are_paired_by_serial(self):
        c = monitor.UnifiClient({'controller': 'https://u.test', 'api_key': 'k'})
        dac = {'sfp_found': True, 'sfp_part': 'SFP-H10GB-CU1M', 'sfp_serial': 'CSC1'}
        resp = MagicMock()
        resp.json.return_value = {'data': [
            {'type': 'udm', 'name': 'gw', 'port_table': [dict(dac, port_idx=11)]},
            {'type': 'usw', 'name': 'agg', 'port_table': [
                dict(dac, port_idx=7),
                dict(dac, port_idx=1, sfp_serial='LONE'),
                dict(self.OPTIC, port_idx=8),
            ]},
        ]}
        c.session.get = MagicMock(return_value=resp)
        sw = c.get_switch_ports()
        assert sw['gw']['ports']['Port 11']['sfp']['cable_peer'] == {'switch': 'agg', 'port_idx': 7}
        assert sw['agg']['ports']['Port 7']['sfp']['cable_peer'] == {'switch': 'gw', 'port_idx': 11}
        assert 'cable_peer' not in sw['agg']['ports']['Port 1']['sfp']
        assert 'cable_peer' not in sw['agg']['ports']['Port 8']['sfp']


class TestLldpCollection:
    def _client_with(self, entry):
        c = monitor.UnifiClient({'controller': 'https://u.test', 'api_key': 'k'})
        resp = MagicMock()
        resp.json.return_value = {'data': [{
            'type': 'usw', 'name': 'sw', 'lldp_table': [entry],
            'port_table': [{'port_idx': 3, 'name': 'Port 3'}],
        }]}
        c.session.get = MagicMock(return_value=resp)
        return c

    def test_current_unifi_field_name(self):
        ports = self._client_with({'local_port_idx': 3, 'chassis_id': 'aa'}).get_switch_ports()['sw']['ports']
        assert ports['Port 3']['lldp']['chassis_id'] == 'aa'

    def test_older_unifi_field_name(self):
        ports = self._client_with({'lldp_port_idx': 3, 'chassis_id': 'aa'}).get_switch_ports()['sw']['ports']
        assert ports['Port 3']['lldp']['chassis_id'] == 'aa'

    def test_bond_does_not_hide_its_member_nic(self):
        collector = monitor.LinkStatsCollector.__new__(monitor.LinkStatsCollector)
        collector.prom = MagicMock()
        collector.unifi = MagicMock()
        collector.unifi.get_wired_clients.return_value = []
        mac = 'aa:bb:cc:dd:ee:ff'
        collector.prom.query.return_value = [
            {'metric': {'device': 'bond0', 'address': mac, 'instance': '10.0.0.1:9100'}},
            {'metric': {'device': 'eth0', 'address': mac, 'instance': '10.0.0.1:9100'}},
            {'metric': {'device': 'vmbr0', 'address': mac, 'instance': '10.0.0.1:9100'}},
        ]
        switches = {'sw': {'ports': {'p': {'lldp': {'chassis_id': 'x', 'port_id': mac.upper()}}}}}
        collector._resolve_neighbors(switches, {}, {'10.0.0.1:9100': 'alpha'})
        lldp = switches['sw']['ports']['p']['lldp']
        assert (lldp['system_name'], lldp['port_id']) == ('alpha', 'eth0')

    def test_client_list_failure_is_empty(self):
        c = monitor.UnifiClient({'controller': 'https://u.test', 'api_key': 'k'})
        c.session.get = MagicMock(side_effect=OSError('boom'))
        assert c.get_wired_clients() == []


class TestCutsheetMismatch:
    CABLE = {'id': 7, 'label': 'C07', 'a_device': 'sw', 'a_port': '13', 'a_side': '', 'b_device': 'tv',
             'b_port': '', 'b_side': '', 'status': 'installed'}

    def _run(self, mon, db, seen='printer', consec=1, ticket=None, suppressions=(), open_targets=()):
        mon.poll_interval = 120
        mon.cutsheet_ticket_hours = getattr(mon, 'cutsheet_ticket_hours', 24)
        db.cutsheet_load.return_value = {'items': [], 'cables': [dict(self.CABLE)], 'port_labels': []}
        db.upsert_event.return_value = (5, consec == 1, consec, ticket)
        db.get_open_event_targets.return_value = list(open_targets)
        db.check_suppressed.side_effect = lambda sups, kind, name, detail='': (kind, name) in sups
        link = {'unifi_switches': {'sw': {'ports': {'Port 13': {
            'port_idx': 13, 'up': True, 'lldp': {'system_name': seen}}}}}}
        mon._process_cutsheet(link, list(suppressions))

    def test_mismatch_opens_an_event_but_no_ticket_yet(self, mon, db):
        self._run(mon, db)
        args = db.upsert_event.call_args.args
        assert args[:5] == ('cable_mismatch', 'warning', 'cutsheet', 'sw', 'cable 7')
        assert 'printer' in args[5]
        assert not mon.tickets.create.called

    def test_ticket_after_a_day_of_polls(self, mon, db):
        self._run(mon, db, consec=719)
        assert not mon.tickets.create.called
        self._run(mon, db, consec=720)
        title = mon.tickets.create.call_args.args[0]
        assert 'Cable C07 does not match the cutsheet' in title
        assert mon.tickets.create.call_args.kwargs['priority'] == '4'

    def test_no_second_ticket_and_suppression_respected(self, mon, db):
        self._run(mon, db, consec=900, ticket='123')
        self._run(mon, db, consec=900, suppressions=[('unifi_device', 'sw')])
        assert not mon.tickets.create.called

    def test_zero_hours_never_tickets(self, mon, db):
        mon.cutsheet_ticket_hours = 0
        self._run(mon, db, consec=10_000)
        assert db.upsert_event.called and not mon.tickets.create.called

    def test_fixed_cable_resolves_its_event(self, mon, db):
        self._run(mon, db, seen='tv', open_targets=[('sw', 'cable 7'), ('other', 'cable 9')])
        assert not db.upsert_event.called
        resolved = {c.args for c in db.resolve_event.call_args_list}
        assert ('cable_mismatch', 'sw', 'cable 7') in resolved and ('cable_mismatch', 'other', 'cable 9') in resolved


class TestTicketContext:
    SWITCHES = {'agg': {'ports': {
        'Large1': {'port_idx': 7, 'up': True, 'speed_mbps': 10000,
                   'lldp': {'system_name': 'large1', 'port_id': 'enp1s0f0'},
                   'sfp': {'found': True, 'kind': 'dac', 'part': 'DAC-10G'}},
        'Port 3': {'port_idx': 3, 'up': False, 'speed_mbps': 0},
    }}}

    def _mon(self, mon, switches=None):
        mon.public_url = 'https://gandalf.test'
        mon.poll_interval = 120
        mon._link_data = {'unifi_switches': switches if switches is not None else self.SWITCHES}
        return mon

    def _desc(self, mon):
        return mon.tickets.create.call_args.args[1]

    def test_interface_ticket_names_the_switch_port_and_links(self, mon, db):
        self._mon(mon)._ticket_interface(1, 'large1', 'enp1s0f0', 2)
        desc = self._desc(mon)
        assert '### Where it is connected' in desc
        assert '`enp1s0f0` → [agg port 7 "Large1": up at 10 Gbps, module: DAC DAC-10G]' in desc
        assert '(https://gandalf.test/device/agg#port-7)' in desc
        assert 'https://gandalf.test/device/large1' in desc
        assert 'Opened automatically by Gandalf' in desc
        assert 'inventory' not in desc

    def test_inventory_link_when_set(self, mon, db):
        db.get_device_link.return_value = 'https://inv.test/item/abc'
        self._mon(mon)._ticket_interface(1, 'large1', 'enp1s0f0', 2)
        assert '- [large1 in the inventory](https://inv.test/item/abc)' in self._desc(mon)
        db.get_device_link.assert_called_with('large1')

    def test_dead_host_uses_the_last_known_port(self, mon, db):
        mon._known_uplinks = {'large1': {'seen': '2026-11-10 10:00:00 UTC', 'uplinks': [
            {'switch': 'agg', 'port': 'Port 3', 'port_idx': 3, 'host_interface': 'eno1'}]}}
        self._mon(mon)._ticket_metrics_missing(1, 'micro1', 'x:9100', 5)
        assert 'not seen this device on a switch port' in self._desc(mon)
        mon.tickets.create.reset_mock()
        unplugged = {'agg': {'ports': {'Port 3': {'port_idx': 3, 'up': False}}}}
        self._mon(mon, unplugged)._ticket_unreachable(1, 'large1', '10.0.0.9', 5)
        desc = self._desc(mon)
        assert 'agg port 3: down' in desc
        assert 'last seen linked there 2026-11-10 10:00:00 UTC' in desc
        assert 'Down for at least' in desc and '10 min' in desc

    def test_cutsheet_runs_and_history(self, mon, db):
        db.cutsheet_load.return_value = {'items': [], 'port_labels': [], 'cables': [
            {'id': 4, 'label': 'C04', 'a_device': 'agg', 'a_port': '7', 'a_side': '',
             'b_device': 'large1', 'b_port': 'enp1s0f0', 'b_side': '', 'status': 'installed'}]}
        db.event_stats.return_value = {'last_7d': 3, 'last_30d': 5,
                                       'last_first_seen': monitor._utcnow(), 'last_resolved_at': monitor._utcnow()}
        self._mon(mon)._ticket_interface(1, 'large1', 'enp1s0f0', 2)
        desc = self._desc(mon)
        assert '[**C04**](https://gandalf.test/cutsheet#cable-4): agg port 7 → large1 enp1s0f0' in desc
        assert 'Happened 3 times in the last 7 days and 5 in the last 30' in desc
        assert 'This keeps coming back' in desc

    def test_other_open_alerts_are_listed(self, mon, db):
        db.get_events_for_target.return_value = [
            {'id': 1, 'event_type': 'interface_down', 'target_name': 'large1', 'target_detail': 'eno1'},
            {'id': 2, 'event_type': 'interface_down', 'target_name': 'large1', 'target_detail': 'eno2',
             'first_seen': 'T', 'ticket_id': '55'},
            {'id': 3, 'event_type': 'host_unreachable', 'target_name': 'large1', 'resolved_at': 'R'},
        ]
        self._mon(mon)._ticket_interface(1, 'large1', 'eno1', 2)
        desc = self._desc(mon)
        assert '- interface down on large1 eno2 since T (ticket #55)' in desc
        assert 'eno1 since' not in desc and 'host unreachable' not in desc

    def test_a_failing_section_still_opens_the_ticket(self, mon, db):
        db.event_stats.side_effect = RuntimeError('db gone')
        self._mon(mon)._ticket_interface(1, 'large1', 'enp1s0f0', 2)
        desc = self._desc(mon)
        assert '### History' not in desc and '### Where it is connected' in desc
        db.set_ticket_id.assert_called_with(1, '123')

    def test_cable_mismatch_shows_the_switch_port(self, mon, db):
        self._mon(mon)._ticket_cable_mismatch(9, {
            'switch': 'agg', 'port': '7', 'cable_id': 4, 'label': 'C04', 'key': 'cable 4',
            'detail': 'A different device is connected', 'path': 'agg port 7 → tv'}, 24)
        desc = self._desc(mon)
        assert '[C04](https://gandalf.test/cutsheet#cable-4)' in desc
        assert '[agg port 7 "Large1": up at 10 Gbps' in desc

    def test_titles_keep_the_tag_format(self, mon, db):
        mon.unifi_fail_thresh = 3
        self._mon(mon)._ticket_unifi(1, {'name': 'ap1', 'type': 'uap', 'model': 'U6', 'ip': '1.2.3.4'})
        assert mon.tickets.create.call_args.args[0] == \
            '[ap1][auto][production][issue][network][single-node] UniFi device offline'

    def test_remember_uplinks_saves_only_on_change(self, mon, db):
        self._mon(mon)
        mon._known_uplinks = None
        mon._remember_uplinks({'unifi_switches': self.SWITCHES})
        saved = db.set_state.call_args.args
        assert saved[0] == 'known_uplinks'
        assert saved[1]['large1']['uplinks'] == [
            {'switch': 'agg', 'port': 'Large1', 'port_idx': 7, 'host_interface': 'enp1s0f0'}]
        db.set_state.reset_mock()
        mon._remember_uplinks({'unifi_switches': self.SWITCHES})
        assert not db.set_state.called


class TestRecoveryNotes:
    def _event(self, **kw):
        now = monitor._utcnow()
        e = {'id': 8, 'event_type': 'host_metrics_missing', 'target_name': 'micro1', 'target_detail': 'x:9100',
             'ticket_id': '993', 'first_seen': now - timedelta(minutes=12), 'resolved_at': now,
             'consecutive_failures': 6}
        e.update(kw)
        return e

    def test_cleared_alert_comments_and_marks(self, mon, db):
        mon.public_url = 'https://gandalf.test'
        mon.tickets.comments_denied = False
        mon.tickets.comment.return_value = True
        db.get_pending_recovery_notes.return_value = [self._event()]
        mon._post_recovery_notes()
        ticket_id, text = mon.tickets.comment.call_args.args
        assert ticket_id == '993'
        assert 'Cleared' in text and '12 min' in text and 'Failed checks:** 6' in text
        assert 'https://gandalf.test/device/micro1' in text
        db.mark_recovery_notified.assert_called_once_with(8)
        assert db.get_pending_recovery_notes.call_args.kwargs['exclude_types'] == monitor.ONE_SHOT_EVENTS

    def test_failed_comment_is_left_for_next_cycle(self, mon, db):
        mon.tickets.comments_denied = False
        mon.tickets.comment.return_value = False
        db.get_pending_recovery_notes.return_value = [self._event(), self._event(id=9)]
        mon._post_recovery_notes()
        assert not db.mark_recovery_notified.called
        assert mon.tickets.comment.call_count == 2

    def test_denied_key_stops_trying(self, mon, db):
        mon.tickets.comments_denied = True
        mon._post_recovery_notes()
        assert not db.get_pending_recovery_notes.called

    def test_cable_mismatch_links_the_cutsheet(self, mon, db):
        mon.public_url = 'https://gandalf.test'
        url = mon._event_url({'event_type': 'cable_mismatch', 'target_name': 'sw', 'target_detail': 'cable 7'})
        assert url == 'https://gandalf.test/cutsheet#cable-7'


class TestTicketClientComment:
    def _client(self):
        return monitor.TicketClient({'url': 'http://t.test/create_ticket_api.php', 'api_key': 'k'})

    def test_comment_endpoint_sits_beside_create(self):
        assert self._client().comment_url == 'http://t.test/api/ticket_comment_api.php'
        sub = monitor.TicketClient({'url': 'http://t.test/tinker/create_ticket_api.php', 'api_key': 'k'})
        assert sub.comment_url == 'http://t.test/tinker/api/ticket_comment_api.php'

    def test_payload(self):
        c = self._client()
        resp = MagicMock(status_code=200)
        resp.json.return_value = {'success': True, 'comment_id': 1}
        with patch.object(monitor.requests, 'post', return_value=resp) as post:
            assert c.comment('42', 'hi') is True
        assert post.call_args.kwargs['json'] == {'ticket_id': '42', 'comment_text': 'hi', 'markdown_enabled': True}
        assert post.call_args.kwargs['headers'] == {'Authorization': 'Bearer k'}

    def test_forbidden_disables_comments(self):
        c = self._client()
        with patch.object(monitor.requests, 'post', return_value=MagicMock(status_code=403)) as post:
            assert c.comment('42', 'hi') is False
            assert c.comment('43', 'hi') is False
        assert post.call_count == 1 and c.comments_denied

    def test_missing_ticket_counts_as_done(self):
        with patch.object(monitor.requests, 'post', return_value=MagicMock(status_code=404)):
            assert self._client().comment('42', 'hi') is True

    def test_network_error_is_retryable(self):
        c = self._client()
        with patch.object(monitor.requests, 'post', side_effect=OSError('down')):
            assert c.comment('42', 'hi') is False
        assert not c.comments_denied


class TestTicketClientCreate:
    def _client(self):
        return monitor.TicketClient({'url': 'http://t.test/create_ticket_api.php', 'api_key': 'k'})

    def _post(self, body):
        resp = MagicMock(status_code=200)
        resp.json.return_value = body
        return patch.object(monitor.requests, 'post', return_value=resp)

    def test_new_and_deduplicated_tickets_are_used(self):
        c = self._client()
        with self._post({'success': True, 'ticket_id': '5', 'action': 'deduplicated'}):
            assert c.create('t', 'd') == '5'
        with self._post({'success': True, 'ticket_id': '6'}):
            assert c.create('u', 'd') == '6'

    def test_recurrence_on_a_closed_ticket_is_not_used_and_not_retried_each_poll(self):
        c = self._client()
        with self._post({'success': True, 'ticket_id': '993', 'action': 'recurrence_noted'}) as post:
            assert c.create('t', 'd') is None
            assert c.create('t', 'd') is None
        assert post.call_count == 1  # the second poll doesn't add another comment to the closed ticket
        with self._post({'success': True, 'ticket_id': '7', 'action': 'none'}) as post:
            assert c.create('other', 'd') == '7'  # other alerts are unaffected

    def test_retry_after_a_day(self):
        c = self._client()
        with self._post({'success': True, 'ticket_id': '993', 'action': 'recurrence_noted'}):
            c.create('t', 'd')
        c._closed_matches['t'] = ('993', time.time() - c.CLOSED_MATCH_RETRY_S - 1)
        with self._post({'success': True, 'ticket_id': '8', 'action': 'none'}):
            assert c.create('t', 'd') == '8'
        assert 't' not in c._closed_matches