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