jaredandClaude Opus 5.5
aa7341af06
Lint / JS (eslint) (push) Successful in 21s
Lint / Python (flake8) (push) Successful in 1m20s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 1m0s
Test / Python Tests (pytest) (push) Successful in 1m48s
Lint / Python (flake8) (pull_request) Successful in 1m44s
Lint / JS (eslint) (pull_request) Successful in 30s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 1m42s
Test / Python Tests (pytest) (pull_request) Successful in 1m16s
The dashboard's hand-drawn diagram was out of date. /topology replaces it with the network as the switches report it, every device drawn to scale from its faceplate or cutsheet front, and one cable per physical link leaving the port it is plugged into. - topology.py builds the graph: switch-to-switch cables seen from both ends are merged, servers get one cable per NIC, the gateway's internet port goes to the modem and on to the internet, unknown neighbours are stubs, and gear with no live cable is drawn where it was last seen. - Ports named after a different device than the one plugged in are flagged. - Cables show medium, speed, state and traffic; click one for its details, hover a device to trace its path to the internet; pan, zoom and filters. - The monitor now keeps host_macs (MAC to host NIC) in link_stats. - The old dashboard diagram and its JS and CSS are removed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
339 lines
17 KiB
Python
339 lines
17 KiB
Python
"""Tests for topology.build_graph, on a small network shaped like the real one."""
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
|
|
|
import topology # noqa: E402
|
|
|
|
|
|
def port(idx, name=None, up=True, speed=1000, neighbour='', port_id='', chassis='', uplink=False, sfp=None,
|
|
tx=0.0, rx=0.0, source=None):
|
|
lldp = {}
|
|
if neighbour or chassis:
|
|
lldp = {'system_name': neighbour, 'port_id': port_id, 'chassis_id': chassis, 'port_desc': '', 'mgmt_ips': []}
|
|
if source:
|
|
lldp['source'] = source
|
|
p = {'port_idx': idx, 'up': up, 'speed_mbps': speed if up else 0, 'is_uplink': uplink, 'media': 'GE',
|
|
'lldp': lldp, 'tx_bytes_rate': tx, 'rx_bytes_rate': rx, 'rx_errs_rate': 0.0, 'tx_errs_rate': 0.0,
|
|
'rx_drops_rate': 0.0, 'tx_drops_rate': 0.0}
|
|
if sfp:
|
|
p['sfp'] = dict({'found': True, 'vendor': 'OEM', 'serial': 'SN1'}, **sfp)
|
|
p['media'] = 'SFP+'
|
|
return (name or f'Port {idx}'), p
|
|
|
|
|
|
def ports(*items):
|
|
return dict(items)
|
|
|
|
|
|
def dac(peer=None, peer_port=None):
|
|
s = {'kind': 'dac', 'part': 'SFP-H10GB-CU1M'}
|
|
if peer:
|
|
s['cable_peer'] = {'switch': peer, 'port_idx': peer_port}
|
|
return s
|
|
|
|
|
|
def network(poe_port_11='Port 11', pbs_up=True):
|
|
"""Gateway -> aggregation -> PoE switch, two dual-homed hosts, an AP, a PDU and unknown devices."""
|
|
switches = {
|
|
'Dream Machine Pro': {'model': 'UDMPRO', 'ports': ports(
|
|
port(1, neighbour='host-a', port_id='eno2', source='unifi-clients'),
|
|
port(8), # up, nothing reported: a cable to something Gandalf can't see
|
|
port(9, uplink=True, chassis='02:00:5e:00:00:09', port_id='02:00:5e:00:00:0a'),
|
|
port(11, 'Downlink to USW-Aggregation', speed=10000, neighbour='8SFP-AGG', port_id='local Port 7',
|
|
sfp=dac('8SFP-AGG', 7), tx=2000.0, rx=1000.0),
|
|
)},
|
|
'8SFP-AGG': {'model': 'USL8A', 'ports': ports(
|
|
port(1, 'SFP+ 1', speed=10000, neighbour='host-a', port_id='eno1', sfp={'kind': 'dac', 'part': 'X'}),
|
|
port(2, 'SFP+ 2', speed=10000, neighbour='host-b', port_id='enp1s0', sfp={'kind': '', 'part': 'L56'}),
|
|
port(5, 'Downlink to 24RJ45-POE', speed=10000, neighbour='24RJ45-POE', port_id='Port 26',
|
|
sfp=dac('24RJ45-POE', 26)),
|
|
port(7, 'SFP+ 7', speed=10000, neighbour='Dream Machine Pro', port_id='02:00:5e:00:00:11',
|
|
uplink=True, sfp=dac('Dream Machine Pro', 11), tx=1000.0, rx=2000.0),
|
|
port(8, 'SFP+ 8', speed=10000, neighbour='host-c', port_id='nic4', sfp={'kind': 'optic', 'part': 'SR'}),
|
|
)},
|
|
'24RJ45-POE': {'model': 'US24PRO', 'ports': ports(
|
|
port(1, 'Micro 2 Eth2', chassis='02:00:5e:00:01:01', port_id='02:00:5E:00:01:01'),
|
|
port(3, 'Living Room Downlink', neighbour='USW Flex', port_id='02:00:5E:00:02:01'),
|
|
port(4, neighbour='host-b', port_id='enp2s0', source='unifi-clients'),
|
|
port(8, neighbour='192.0.2.44', port_id='eth0', chassis='02:00:5e:00:01:08', source='unifi-clients'),
|
|
port(11, poe_port_11, neighbour='AP', port_id='02:00:5E:00:03:01'),
|
|
port(12, up=False),
|
|
port(15, 'PBS', up=pbs_up, neighbour='pbs' if pbs_up else '', port_id='enp0s25' if pbs_up else ''),
|
|
port(19, neighbour='RPDU', port_id='02:00:5E:00:04:01', speed=100),
|
|
port(26, 'SFP+ 2', speed=10000, neighbour='8SFP-AGG', port_id='Port 5', uplink=True,
|
|
sfp=dac('8SFP-AGG', 5)),
|
|
)},
|
|
'USW Flex': {'model': 'USF5P', 'ports': ports(
|
|
port(1, uplink=True, neighbour='24RJ45-POE', port_id='Port 3'),
|
|
port(5, neighbour='UPS 2U Pro', port_id='Port 1', speed=100),
|
|
)},
|
|
'RPDU': {'model': 'USPPDUP', 'ports': ports(port(1, uplink=True, speed=100))},
|
|
}
|
|
unifi = [
|
|
{'name': 'Dream Machine Pro', 'type': 'udm', 'model': 'UDMPRO', 'connected': True, 'mac': '', 'ip': ''},
|
|
{'name': '8SFP-AGG', 'type': 'usw', 'model': 'USL8A', 'connected': True, 'mac': '', 'ip': ''},
|
|
{'name': '24RJ45-POE', 'type': 'usw', 'model': 'US24PRO', 'connected': True, 'mac': '', 'ip': ''},
|
|
{'name': 'USW Flex', 'type': 'usw', 'model': 'USF5P', 'connected': True, 'mac': '', 'ip': ''},
|
|
{'name': 'AP', 'type': 'uap', 'model': 'UAL6', 'connected': True, 'mac': '', 'ip': ''},
|
|
{'name': 'RPDU', 'type': 'usw', 'model': 'USPPDUP', 'connected': True, 'mac': '', 'ip': ''},
|
|
{'name': 'UPS 2U Pro', 'type': 'usp', 'model': 'USPDA2B', 'connected': True, 'mac': '', 'ip': ''},
|
|
{'name': 'USP Strip', 'type': 'uap', 'model': 'UP6', 'connected': True, 'mac': '', 'ip': ''},
|
|
]
|
|
hosts = {
|
|
'host-a': {'status': 'up', 'interfaces': {'eno1': 'up', 'eno2': 'up', 'eno3': 'unused', 'vmbr0': 'up'}},
|
|
'host-b': {'status': 'degraded',
|
|
'interfaces': {'bond0': 'up', 'enp1s0': 'up', 'enp2s0': 'up', 'enp3s0': 'down'}},
|
|
'host-c': {'status': 'up', 'interfaces': {'nic4': 'up'}},
|
|
'pbs': {'status': 'up' if pbs_up else 'down', 'interfaces': {'enp0s25': 'up' if pbs_up else 'down'}},
|
|
}
|
|
snapshot = {'hosts': hosts, 'unifi': unifi}
|
|
link_stats = {
|
|
'hosts': {'192.0.2.44': {'eth0': {}}}, # an IP-only Prometheus target
|
|
'unifi_switches': switches,
|
|
'unifi_devices': {'Dream Machine Pro': {'wan': {'wan1': {'up': True}}},
|
|
'RPDU': {'power_w': 120.5, 'outlets': [{'index': 1, 'name': 'AT&T Modem', 'on': True}]}},
|
|
'updated': '2026-11-14 12:00:00 UTC',
|
|
}
|
|
modem = {'name': 'att-modem', 'model': 'BGW320-505', 'reachable': True, 'broadband_state': 'Up',
|
|
'source': 'FIBER', 'line_speed_mbps': 10000, 'rates': {'rx_bytes': 5000.0, 'tx_bytes': 700.0},
|
|
'wan_problems': []}
|
|
return snapshot, link_stats, modem
|
|
|
|
|
|
def build(events=(), items=(), cables=(), known=None, **kw):
|
|
snapshot, link_stats, modem = network(**kw)
|
|
return topology.build_graph(snapshot, link_stats, modem, list(events), list(items), list(cables), known)
|
|
|
|
|
|
def find(graph, a, b):
|
|
"""The links between two nodes, either way round."""
|
|
return [lk for lk in graph['links'] if {lk['a']['node'], lk['b']['node']} == {a, b}]
|
|
|
|
|
|
def node(graph, name):
|
|
return next(n for n in graph['nodes'] if n['id'] == name)
|
|
|
|
|
|
class TestSwitchCables:
|
|
def test_cable_seen_from_both_ends_is_drawn_once(self):
|
|
g = build()
|
|
assert len(find(g, 'Dream Machine Pro', '8SFP-AGG')) == 1
|
|
assert len(find(g, '8SFP-AGG', '24RJ45-POE')) == 1
|
|
assert len(find(g, '24RJ45-POE', 'USW Flex')) == 1
|
|
|
|
def test_both_ports_of_a_switch_cable_are_known(self):
|
|
g = build()
|
|
(link,) = find(g, 'Dream Machine Pro', '8SFP-AGG')
|
|
ends = {link['a']['node']: link['a']['port_idx'], link['b']['node']: link['b']['port_idx']}
|
|
assert ends == {'Dream Machine Pro': 11, '8SFP-AGG': 7}
|
|
(flex,) = find(g, '24RJ45-POE', 'USW Flex')
|
|
assert {flex['a']['port_idx'], flex['b']['port_idx']} == {3, 1}
|
|
|
|
def test_gear_without_lldp_lands_on_its_uplink(self):
|
|
(pdu,) = find(build(), '24RJ45-POE', 'RPDU')
|
|
assert pdu['b'] == {'node': 'RPDU', 'port_idx': 1, 'port_name': 'Port 1', 'iface': ''}
|
|
|
|
def test_cables_point_away_from_the_internet(self):
|
|
g = build()
|
|
(link,) = find(g, 'Dream Machine Pro', '8SFP-AGG')
|
|
assert link['a']['node'] == 'Dream Machine Pro'
|
|
# The gateway sends 2000 B/s down the cable and gets 1000 back, whichever end measured it
|
|
assert (link['a_to_b_bps'], link['b_to_a_bps']) == (2000.0, 1000.0)
|
|
(down,) = find(g, '8SFP-AGG', '24RJ45-POE')
|
|
assert down['a']['node'] == '8SFP-AGG'
|
|
|
|
def test_medium_and_module(self):
|
|
g = build()
|
|
assert find(g, 'Dream Machine Pro', '8SFP-AGG')[0]['medium'] == 'dac'
|
|
assert find(g, '8SFP-AGG', 'host-c')[0]['medium'] == 'fibre'
|
|
assert find(g, '8SFP-AGG', 'host-b')[0]['medium'] == 'sfp'
|
|
assert find(g, '24RJ45-POE', 'host-b')[0]['medium'] == 'copper'
|
|
assert find(g, 'Dream Machine Pro', '8SFP-AGG')[0]['sfp_part'] == 'SFP-H10GB-CU1M'
|
|
|
|
|
|
class TestWanEdge:
|
|
def test_gateway_to_modem_to_internet(self):
|
|
g = build()
|
|
(wan,) = find(g, 'Dream Machine Pro', 'att-modem')
|
|
assert wan['wan'] and wan['b']['node'] == 'Dream Machine Pro' and wan['a']['node'] == 'att-modem'
|
|
assert wan['b']['port_idx'] == 9
|
|
(fibre,) = find(g, 'att-modem', topology.INTERNET)
|
|
assert fibre['medium'] == 'fibre' and fibre['speed_mbps'] == 10000 and fibre['up']
|
|
# Down from the internet is what the modem received
|
|
assert fibre['a_to_b_bps'] == 5000.0
|
|
|
|
def test_the_wan_port_is_not_a_stub(self):
|
|
assert 'stub:Dream Machine Pro:9' not in {n['id'] for n in build()['nodes']}
|
|
|
|
def test_without_a_modem_the_gateway_goes_straight_out(self):
|
|
snapshot, link_stats, _ = network()
|
|
g = topology.build_graph(snapshot, link_stats, None, [], [], [], None)
|
|
assert len(find(g, 'Dream Machine Pro', topology.INTERNET)) == 1
|
|
|
|
def test_internet_status_follows_the_gateway_wan(self):
|
|
snapshot, link_stats, modem = network()
|
|
link_stats['unifi_devices']['Dream Machine Pro']['wan']['wan1']['up'] = False
|
|
g = topology.build_graph(snapshot, link_stats, modem, [], [], [], None)
|
|
assert node(g, topology.INTERNET)['status'] == 'down'
|
|
|
|
|
|
class TestHosts:
|
|
def test_dual_homed_host_has_two_cables(self):
|
|
g = build()
|
|
a = [lk for lk in g['links'] if 'host-a' in (lk['a']['node'], lk['b']['node'])]
|
|
assert sorted((lk['speed_mbps'], lk['b']['iface']) for lk in a) == [(1000, 'eno2'), (10000, 'eno1')]
|
|
|
|
def test_physical_nics_only_with_the_cables_on_them(self):
|
|
nics = {n['name']: n for n in node(build(), 'host-a')['interfaces']}
|
|
assert set(nics) == {'eno1', 'eno2', 'eno3'}
|
|
assert nics['eno3']['unused'] and 'links' not in nics['eno3']
|
|
assert len(nics['eno1']['links']) == 1
|
|
|
|
def test_ip_only_target_is_a_stub_not_a_host(self):
|
|
g = build()
|
|
assert '192.0.2.44' not in {n['id'] for n in g['nodes'] if n['kind'] != 'stub'}
|
|
stub = node(g, 'stub:24RJ45-POE:8')
|
|
assert stub['name'] == '192.0.2.44' and stub['ip'] == '192.0.2.44' and stub['iface'] == 'eth0'
|
|
|
|
def test_chassis_comes_from_the_cutsheet_item(self):
|
|
items = [{'name': 'Server A', 'device': 'host-a', 'chassis': 'sliger-cx4712-red'},
|
|
{'name': 'host-b', 'device': '', 'chassis': 'zimaboard-832-cx3'}]
|
|
g = build(items=items)
|
|
assert node(g, 'host-a')['chassis'] == 'sliger-cx4712-red'
|
|
assert node(g, 'host-b')['chassis'] == 'zimaboard-832-cx3'
|
|
assert node(g, 'host-c')['chassis'] == ''
|
|
|
|
|
|
class TestStubs:
|
|
def test_unnamed_neighbour_is_a_stub_named_by_its_port(self):
|
|
stub = node(build(), 'stub:24RJ45-POE:1')
|
|
assert stub['kind'] == 'stub' and stub['name'] == 'Micro 2 Eth2'
|
|
assert stub['mac'] == '02:00:5e:00:01:01' and not stub['anonymous']
|
|
|
|
def test_link_with_no_neighbour_is_an_anonymous_stub(self):
|
|
stub = node(build(), 'stub:Dream Machine Pro:8')
|
|
assert stub['anonymous'] and stub['name'] == 'unknown device'
|
|
|
|
def test_ports_with_no_link_are_left_out(self):
|
|
assert 'stub:24RJ45-POE:12' not in {n['id'] for n in build()['nodes']}
|
|
|
|
|
|
class TestNameMismatch:
|
|
def test_port_named_after_another_device(self):
|
|
(link,) = find(build(poe_port_11='Large1 Management'), '24RJ45-POE', 'AP')
|
|
assert link['name_mismatch'] is None # no device called large1 in this network
|
|
|
|
def test_flags_when_the_named_device_exists(self):
|
|
(link,) = find(build(poe_port_11='host-b management'), '24RJ45-POE', 'AP')
|
|
assert link['name_mismatch'] == {'named': 'host-b', 'actual': 'AP', 'end': 'a'}
|
|
|
|
def test_downlink_and_generic_names_are_fine(self):
|
|
g = build()
|
|
assert all(lk['name_mismatch'] is None for lk in g['links'])
|
|
|
|
@pytest.mark.parametrize('name', ['Port 4', 'SFP+ 2', 'Uplink', 'Downlink to 24RJ45-POE'])
|
|
def test_names_that_never_flag(self, name):
|
|
known = {topology._norm(n): n for n in ('24RJ45-POE', 'host-a')}
|
|
assert topology.name_mismatch(name, '24RJ45-POE', 'x', known) is None
|
|
|
|
def test_matches_ignore_case_spacing_and_dashes(self):
|
|
known = {topology._norm('compute-storage-01'): 'compute-storage-01'}
|
|
assert topology.name_mismatch('Compute Storage 01 mgmt', 'AP', 'sw', known)['named'] == 'compute-storage-01'
|
|
|
|
def test_a_stub_is_checked_against_what_was_seen(self):
|
|
snapshot, link_stats, modem = network()
|
|
link_stats['unifi_switches']['24RJ45-POE']['ports']['host-c'] = port(
|
|
20, 'host-c', chassis='02:00:5e:00:09:09')[1]
|
|
g = topology.build_graph(snapshot, link_stats, modem, [], [], [], None)
|
|
(link,) = find(g, '24RJ45-POE', 'stub:24RJ45-POE:20')
|
|
assert link['name_mismatch']['named'] == 'host-c'
|
|
|
|
|
|
class TestGhosts:
|
|
KNOWN = {'pbs': {'seen': '2026-11-14 20:49:08 UTC', 'uplinks': [
|
|
{'switch': '24RJ45-POE', 'port': 'PBS', 'port_idx': 15, 'host_interface': 'enp0s25'}]}}
|
|
|
|
def test_missing_host_is_drawn_where_last_seen(self):
|
|
g = build(known=self.KNOWN, pbs_up=False)
|
|
(ghost,) = find(g, '24RJ45-POE', 'pbs')
|
|
assert ghost['ghost'] and not ghost['up'] and ghost['seen'] == '2026-11-14 20:49:08 UTC'
|
|
assert ghost['b'] == {'node': 'pbs', 'port_idx': None, 'port_name': '', 'iface': 'enp0s25'}
|
|
|
|
def test_no_ghost_while_the_host_is_linked(self):
|
|
g = build(known=self.KNOWN)
|
|
assert [lk['ghost'] for lk in find(g, '24RJ45-POE', 'pbs')] == [False]
|
|
|
|
def test_no_ghost_onto_a_port_now_used_by_something_else(self):
|
|
known = {'pbs': {'seen': 'x', 'uplinks': [{'switch': '24RJ45-POE', 'port_idx': 4, 'host_interface': 'e'}]}}
|
|
assert find(build(known=known, pbs_up=False), '24RJ45-POE', 'pbs') == []
|
|
|
|
|
|
class TestDepth:
|
|
def test_hops_from_the_internet(self):
|
|
g = build()
|
|
depths = {n['id']: n['depth'] for n in g['nodes']}
|
|
assert depths[topology.INTERNET] == 0
|
|
assert depths['att-modem'] == 1
|
|
assert depths['Dream Machine Pro'] == 2
|
|
assert depths['8SFP-AGG'] == 3
|
|
assert depths['24RJ45-POE'] == 4
|
|
assert depths['USW Flex'] == 5 and depths['AP'] == 5
|
|
assert depths['UPS 2U Pro'] == 6
|
|
# A host's depth is its nearest switch's plus one; hosts never pass the path on
|
|
assert depths['host-a'] == 3
|
|
|
|
def test_unlinked_gear_has_no_depth(self):
|
|
assert node(build(), 'USP Strip')['depth'] is None
|
|
|
|
def test_power_gear_is_marked(self):
|
|
g = build()
|
|
assert {n['id']: n['kind'] for n in g['nodes'] if n['power']} == {
|
|
'RPDU': 'pdu', 'UPS 2U Pro': 'ups', 'USP Strip': 'strip'}
|
|
assert node(g, 'RPDU')['power_w'] == 120.5
|
|
|
|
|
|
class TestAlerts:
|
|
def test_active_alerts_count_on_their_device(self):
|
|
events = [
|
|
{'id': 1, 'event_type': 'interface_down', 'severity': 'critical', 'target_name': 'host-b',
|
|
'target_detail': 'enp3s0', 'description': ''},
|
|
{'id': 2, 'event_type': 'unifi_port_errors', 'severity': 'warning', 'target_name': '24RJ45-POE',
|
|
'target_detail': 'Port 9', 'description': ''},
|
|
{'id': 3, 'event_type': 'host_down', 'severity': 'critical', 'target_name': 'host-a',
|
|
'target_detail': '', 'description': '', 'is_suppressed': True},
|
|
]
|
|
g = build(events=events)
|
|
b = node(g, 'host-b')
|
|
assert (b['alert_count'], b['severity']) == (1, 'critical')
|
|
assert next(n for n in b['interfaces'] if n['name'] == 'enp3s0')['alert'] == 'critical'
|
|
assert node(g, '24RJ45-POE')['severity'] == 'warning'
|
|
assert node(g, 'host-a')['alert_count'] == 0
|
|
|
|
def test_status_comes_from_the_snapshot(self):
|
|
assert node(build(), 'host-b')['status'] == 'degraded'
|
|
|
|
|
|
class TestCutsheetLabels:
|
|
def test_cable_label_on_either_end(self):
|
|
items = [{'name': 'Patch', 'device': '', 'chassis': ''}, {'name': 'UDM-Pro', 'device': 'Dream Machine Pro'}]
|
|
cables = [{'id': 7, 'label': 'C07', 'a_device': 'UDM-Pro', 'a_port': '8', 'b_device': 'Patch',
|
|
'b_port': '1', 'status': 'installed', 'cable_type': 'Cat6', 'color': 'Black'},
|
|
{'id': 8, 'label': 'C08', 'a_device': 'host-a', 'a_port': 'eno2', 'b_device': '',
|
|
'b_port': '', 'status': 'retired'}]
|
|
g = build(items=items, cables=cables)
|
|
(stub,) = find(g, 'Dream Machine Pro', 'stub:Dream Machine Pro:8')
|
|
assert stub['cable'] == {'id': 7, 'label': 'C07', 'type': 'Cat6', 'color': 'Black'}
|
|
# Retired cables are not on the network any more
|
|
assert find(g, 'Dream Machine Pro', 'host-a')[0]['cable'] is None
|
|
|
|
|
|
class TestEmpty:
|
|
def test_no_data_yet(self):
|
|
g = topology.build_graph({}, {}, None, [], [], [], None)
|
|
assert [n['id'] for n in g['nodes']] == [topology.INTERNET]
|
|
assert g['links'] == [] and g['warnings']
|