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_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'] == {}