LotusGuild/gandalf
3
1
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
afc7aae5803369a4981caa41e09c0e710caadcae
gandalf/tests/test_monitor.py
T
jaredandClaude Opus 5.5 7e588e5b02
Security / Python Security (bandit) (push) Successful in 1m50s
Details
Test / Python Tests (pytest) (push) Successful in 1m52s
Details
Lint / Python (flake8) (pull_request) Successful in 1m20s
Details
Lint / Python (flake8) (push) Successful in 1m16s
Details
Lint / JS (eslint) (pull_request) Successful in 10s
Details
Lint / JS (eslint) (push) Successful in 15s
Details
Lint / Notify on failure (push) Skipped
Details
Lint / Deploy (push) Skipped
Details
Lint / Notify on failure (pull_request) Skipped
Details
Lint / Deploy (pull_request) Skipped
Details
Security / Python Security (bandit) (pull_request) Successful in 1m5s
Details
Test / Python Tests (pytest) (pull_request) Successful in 1m7s
Details
Keep every switch port and show SFP modules 
Ports were keyed by their UniFi name, so when two ports shared a name the
later one replaced the earlier. The 24-port switch lost port 13 this way
(ports 13 and 14 are both "Office TV"). A repeated name is now kept as
"<name> (Port N)".

Traffic rates are now matched to the previous poll by port number rather
than name. Names are re-read from UniFi every poll, so a port renamed
there keeps its rates instead of losing them for a cycle.

Collect what UniFi reports about each SFP cage: whether a module is
present, its type (DAC or optic), part, serial, fault flags, and DOM
readings when the module has them. The faceplate draws empty cages,
optics, DACs and unidentified modules differently and outlines a faulted
module in red. The port panel gets an SFP Module section. Both ends of a
DAC report the same cable serial, so each end links to the other.

Closes #7
Closes #8

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-10 14:37:08 -05:00

516 lines
24 KiB
Python
Raw Blame History

"""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() == []
Reference in New Issue View Git Blame Copy Permalink