gandalf/tests/test_topology.py

"""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']