gandalf/tests/test_devices.py

"""Tests for devices.py: the inventory behind search and device pages."""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

import devices  # noqa: E402

SNAPSHOT = {
    'hosts': {
        'large1': {'ip': '10.0.0.2', 'status': 'degraded', 'source': 'prometheus',
                   'interfaces': {'eth0': 'up', 'eth1': 'down'}},
    },
    'unifi': [
        {'name': 'Dream Machine Pro', 'type': 'udm', 'model': 'UDMPRO', 'ip': '203.0.113.5',
         'mac': 'aa:bb:cc:dd:ee:ff', 'connected': True},
        {'name': '24RJ45-POE', 'type': 'usw', 'model': 'US24PRO', 'ip': '10.0.0.40',
         'mac': '11:22:33:44:55:66', 'connected': False},
        {'name': 'RPDU', 'type': 'usw', 'model': 'USPPDUP', 'ip': '10.0.0.50', 'mac': '', 'connected': True},
        {'name': 'AP', 'type': 'uap', 'model': 'UAL6', 'ip': '10.0.0.60', 'mac': '', 'connected': True},
    ],
}
LINKS = {
    'hosts': {'large1': {'eth0': {'host_ip': '10.0.0.2', 'speed_mbps': 1000}}},
    'unifi_switches': {
        '24RJ45-POE': {'ip': '10.0.0.40', 'model': 'US24PRO', 'ports': {
            'Port 3': {'port_idx': 3, 'up': True, 'speed_mbps': 1000,
                       'lldp': {'system_name': 'large1', 'port_id': 'eth0'}},
            'Port 4': {'port_idx': 4, 'up': False, 'lldp': None},
        }},
    },
    'unifi_devices': {'Dream Machine Pro': {'firmware': '6.0.17', 'wan': {'wan1': {'up': True}}}},
    'updated': '2026-11-05 00:00:00 UTC',
}
MODEM = {'name': 'att-modem', 'reachable': True, 'lan_ip': '192.168.1.254', 'wan_ip': '203.0.113.5',
         'model': 'BGW320-505', 'wan_problems': []}


def index():
    return {d['name']: d for d in devices.build_index(SNAPSHOT, LINKS, MODEM)}


class TestKinds:
    def test_kinds(self):
        i = index()
        assert i['large1']['kind'] == 'host'
        assert i['Dream Machine Pro']['kind'] == 'router'
        assert i['24RJ45-POE']['kind'] == 'switch'
        assert i['AP']['kind'] == 'ap'
        assert i['att-modem']['kind'] == 'modem'

    def test_pdu_model_beats_reported_type(self):
        assert index()['RPDU']['kind'] == 'pdu'


class TestSearchText:
    def test_finds_host_by_ip_and_interface(self):
        s = index()['large1']['search']
        assert '10.0.0.2' in s and 'eth1' in s

    def test_mac_with_and_without_separators(self):
        s = index()['Dream Machine Pro']['search']
        assert 'aa:bb:cc:dd:ee:ff' in s and 'aabbccddeeff' in s

    def test_switch_found_by_port_neighbor(self):
        assert 'large1' in index()['24RJ45-POE']['search']

    def test_modem_found_by_either_address_or_word(self):
        s = index()['att-modem']['search']
        assert '192.168.1.254' in s and '203.0.113.5' in s and 'fiber' in s


class TestStatus:
    def test_offline_unifi_device_is_down(self):
        assert index()['24RJ45-POE']['status'] == 'down'

    def test_modem_status(self):
        def status(**kw):
            return {d['name']: d for d in devices.build_index({}, {}, dict(MODEM, **kw))}['att-modem']['status']
        assert status() == 'up'
        assert status(wan_problems=['optical WAN is Down']) == 'down'
        assert status(reachable=False) == 'unknown'


class TestPartialData:
    def test_empty_state(self):
        assert devices.build_index({}, {}, {}) == []
        assert devices.build_index(None, None, None) == []

    def test_switch_known_only_from_link_stats(self):
        d = {x['name']: x for x in devices.build_index({}, LINKS, None)}['24RJ45-POE']
        assert d['status'] == 'unknown' and d['kind'] == 'switch'


class TestUnconfiguredHosts:
    LINKS = {'hosts': {'10.0.0.77': {'eth0': {'host_ip': '10.0.0.77', 'speed_mbps': 1000}}}}

    def test_host_only_in_link_stats_is_kept(self):
        d = {x['name']: x for x in devices.build_index({}, self.LINKS, None)}['10.0.0.77']
        assert d['kind'] == 'host' and d['status'] == 'unknown' and d['ip'] == '10.0.0.77'

    def test_its_interfaces_are_available(self):
        d = devices.device_detail('10.0.0.77', {}, self.LINKS, None)
        assert d['interfaces']['eth0']['speed_mbps'] == 1000


class TestDetail:
    def test_unknown_device(self):
        assert devices.device_detail('nope', SNAPSHOT, LINKS, MODEM) is None

    def test_host_merges_state_and_stats(self):
        d = devices.device_detail('large1', SNAPSHOT, LINKS, MODEM)
        assert d['interfaces']['eth0'] == {'state': 'up', 'host_ip': '10.0.0.2', 'speed_mbps': 1000}
        assert d['interfaces']['eth1'] == {'state': 'down'}
        assert 'search' not in d

    def test_host_shows_switch_port_from_lldp(self):
        up = devices.device_detail('large1', SNAPSHOT, LINKS, MODEM)['uplinks']
        assert up == [{'switch': '24RJ45-POE', 'port': 'Port 3', 'port_idx': 3,
                       'host_interface': 'eth0', 'up': True, 'speed_mbps': 1000}]

    def test_switch_has_ports(self):
        assert set(devices.device_detail('24RJ45-POE', SNAPSHOT, LINKS, MODEM)['ports']) == {'Port 3', 'Port 4'}

    def test_switch_gets_only_its_lldp_neighbors_for_the_path_check(self):
        links = dict(LINKS, hosts=dict(LINKS['hosts'], other={'eth9': {}}))
        d = devices.device_detail('24RJ45-POE', SNAPSHOT, links, MODEM)
        assert set(d['neighbor_hosts']) == {'large1'}
        assert 'eth0' in d['neighbor_hosts']['large1']

    def test_router_has_info_even_without_port_data(self):
        d = devices.device_detail('Dream Machine Pro', SNAPSHOT, LINKS, MODEM)
        assert d['info']['firmware'] == '6.0.17' and d['ports'] == {}


class TestLldpResolution:
    HOSTS = {'9c:6b:00:8a:17:e6': ('micro1', 'enp1s0')}
    DEVS = {'78:45:58:b1:bd:3a': '24RJ45-POE'}
    IPS = {'10.0.0.9': 'storage-01'}

    def resolve(self, **lldp):
        sw = {'s': {'ports': {'p': {'lldp': lldp}}}}
        devices.resolve_lldp_neighbors(sw, self.HOSTS, self.DEVS, self.IPS)
        return sw['s']['ports']['p']['lldp']

    def test_port_mac_gives_host_and_interface(self):
        got = self.resolve(chassis_id='aa:aa:aa:aa:aa:aa', port_id='9C:6B:00:8A:17:E6')
        assert (got['system_name'], got['port_id']) == ('micro1', 'enp1s0')

    def test_chassis_mac_keeps_a_named_port(self):
        got = self.resolve(chassis_id='9c:6b:00:8a:17:e6', port_id='enp9s0')
        assert (got['system_name'], got['port_id']) == ('micro1', 'enp9s0')

    def test_unifi_device_by_chassis(self):
        got = self.resolve(chassis_id='78:45:58:b1:bd:3a', port_id='Port 26')
        assert (got['system_name'], got['port_id']) == ('24RJ45-POE', 'Port 26')

    def test_management_ip_fallback(self):
        got = self.resolve(chassis_id='aa:aa:aa:aa:aa:aa', port_id='x', mgmt_ips=['10.0.0.9'])
        assert got['system_name'] == 'storage-01'

    def test_unknown_neighbor_stays_unnamed(self):
        got = self.resolve(chassis_id='aa:aa:aa:aa:aa:aa', port_id='aa:aa:aa:aa:aa:ab')
        assert got['system_name'] == ''

    def test_a_reported_name_is_kept(self):
        got = self.resolve(system_name='lldp-name', chassis_id='78:45:58:b1:bd:3a', port_id='x')
        assert got['system_name'] == 'lldp-name'

    def test_port_without_lldp_is_left_alone(self):
        sw = {'s': {'ports': {'p': {'lldp': None}}}}
        devices.resolve_lldp_neighbors(sw, self.HOSTS, self.DEVS, self.IPS)
        assert sw['s']['ports']['p']['lldp'] is None


class TestClientLocations:
    HOSTS = {'aa:aa:aa:aa:aa:01': ('large1', 'enp35s0')}
    DEVS = {'bb:bb:bb:bb:bb:01': 'sw'}

    def apply(self, port, sw_port=10):
        switches = {'sw': {'ports': {'Port 10': dict({'port_idx': 10}, **port)}}}
        client = {'mac': 'AA:AA:AA:AA:AA:01', 'sw_mac': 'bb:bb:bb:bb:bb:01', 'sw_port': sw_port}
        devices.apply_client_locations(switches, [client], self.HOSTS, self.DEVS)
        return switches['sw']['ports']['Port 10']['lldp']

    def test_server_without_lldp_is_located(self):
        got = self.apply({'lldp': None})
        assert (got['system_name'], got['port_id'], got['source']) == ('large1', 'enp35s0', 'unifi-clients')

    def test_uplink_port_is_skipped(self):
        assert self.apply({'lldp': None, 'is_uplink': True}) is None

    def test_named_lldp_neighbor_is_kept(self):
        assert self.apply({'lldp': {'system_name': 'other'}})['system_name'] == 'other'

    def test_unnamed_lldp_entry_is_replaced(self):
        assert self.apply({'lldp': {'system_name': '', 'chassis_id': 'x'}})['system_name'] == 'large1'

    def test_other_port_is_untouched(self):
        assert self.apply({'lldp': None}, sw_port=11) is None

    def test_non_host_client_is_ignored(self):
        switches = {'sw': {'ports': {'Port 10': {'port_idx': 10, 'lldp': None}}}}
        devices.apply_client_locations(
            switches, [{'mac': 'cc:cc:cc:cc:cc:cc', 'sw_mac': 'bb:bb:bb:bb:bb:01', 'sw_port': 10}],
            self.HOSTS, self.DEVS,
        )
        assert switches['sw']['ports']['Port 10']['lldp'] is None