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.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_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