"""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'] class TestPower: OUTLETS = [ {'index': 1, 'name': 'AT&T Modem', 'on': True, 'power_w': 9.0}, {'index': 6, 'name': 'Dream Machine Pro', 'on': True, 'power_w': 20.6}, {'index': 7, 'name': 'Outlet 7', 'on': True, 'power_w': 48.0}, {'index': 14, 'name': 'host-a', 'on': False, 'power_w': 0.0}, {'index': 16, 'name': '02:00:5e:00:0a:01', 'on': True, 'power_w': 131.7}, {'index': 17, 'name': 'USB Outlet 1', 'on': True, 'power_w': None}, ] def _graph(self, feeds=(), outlets=None): snapshot, link_stats, modem = network() link_stats['unifi_devices']['RPDU']['outlets'] = outlets or self.OUTLETS link_stats['host_macs'] = {'02:00:5e:00:0a:01': ['host-c', 'nic0']} return topology.build_graph(snapshot, link_stats, modem, [], [], [], None, list(feeds)) def _outlets(self, g, source='RPDU'): return {o['index']: o for o in next(s for s in g['power'] if s['node'] == source)['outlets']} def test_matched_by_name_alias_and_mac(self): out = self._outlets(self._graph()) assert (out[1]['target'], out[1]['match']) == ('att-modem', 'name') assert (out[6]['target'], out[6]['match']) == ('Dream Machine Pro', 'name') assert (out[14]['target'], out[14]['on']) == ('host-a', False) assert (out[16]['target'], out[16]['match']) == ('host-c', 'mac') def test_generic_names_feed_nothing(self): out = self._outlets(self._graph()) assert (out[7]['target'], out[7]['match']) == (None, 'none') assert out[17]['target'] is None def test_saved_assignment_wins_and_empty_means_nothing(self): feeds = [{'source': 'RPDU', 'outlet': 7, 'target': 'host-b'}, {'source': 'RPDU', 'outlet': 6, 'target': ''}] out = self._outlets(self._graph(feeds)) assert (out[7]['target'], out[7]['match']) == ('host-b', 'saved') assert (out[6]['target'], out[6]['match']) == (None, 'cleared') def test_devices_know_what_feeds_them(self): g = self._graph() assert node(g, 'host-c')['power_in'] == [{'source': 'RPDU', 'outlet': 16, 'on': True, 'power_w': 131.7}] assert node(g, 'host-b')['power_in'] == [] def test_a_power_device_does_not_feed_itself(self): out = self._outlets(self._graph(outlets=[{'index': 1, 'name': 'RPDU spare', 'on': True}])) assert out[1]['target'] is None def test_every_power_device_is_listed(self): assert [s['node'] for s in self._graph()['power']] == ['RPDU', 'UPS 2U Pro', 'USP Strip']