"""Unit tests for OpenWrtDriver — no real device required.""" import pytest from unittest.mock import MagicMock, patch from napalm_openwrt.openwrt import OpenWrtDriver # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def driver(): """Return a driver instance with a mocked Netmiko connection.""" with patch("napalm_openwrt.openwrt.ConnectHandler"): drv = OpenWrtDriver( hostname="192.168.1.1", username="root", password="", ) drv.device = MagicMock() yield drv # --------------------------------------------------------------------------- # Sample command output (as would be returned by the device) # --------------------------------------------------------------------------- OPENWRT_RELEASE = """\ DISTRIB_ID="OpenWrt" DISTRIB_RELEASE="23.05.3" DISTRIB_REVISION="r23809-234f1a2efa" DISTRIB_TARGET="ath79/generic" DISTRIB_ARCH="mips_24kc" DISTRIB_CODENAME="Restoring Earth" DISTRIB_TAINTS="" """ SYSINFO_MODEL = "TP-Link TL-WR1043N/ND v5" UPTIME = "352467.12 345823.44" IP_LINK_SHOW = """\ 1: lo: mtu 65536 qdisc noqueue state UNKNOWN mode DEFAULT group default qlen 1000 link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00 2: eth0: mtu 1500 qdisc fq_codel state UP mode DEFAULT group default qlen 1000 link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff 3: eth1: mtu 1500 qdisc noop state DOWN mode DEFAULT group default qlen 1000 link/ether b0:95:75:aa:bb:dd brd ff:ff:ff:ff:ff:ff 4: br-lan: mtu 1500 qdisc noqueue state UP mode DEFAULT group default qlen 1000 link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff """ IP_ADDR_SHOW = """\ 1: lo: mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000 link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00 inet 127.0.0.1/8 scope host lo valid_lft forever preferred_lft forever 2: eth0: mtu 1500 qdisc fq_codel state UP group default qlen 1000 link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff 4: br-lan: mtu 1500 qdisc noqueue state UP group default qlen 1000 link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff inet 192.168.1.1/24 brd 192.168.1.255 scope global br-lan valid_lft forever preferred_lft forever inet6 fd00::1/64 scope global valid_lft forever preferred_lft forever """ IP_NEIGH_SHOW = """\ 192.168.1.100 dev br-lan lladdr aa:bb:cc:dd:ee:ff REACHABLE 192.168.1.101 dev br-lan lladdr 11:22:33:44:55:66 STALE 192.168.1.102 dev br-lan FAILED """ BRIDGE_FDB = """\ aa:bb:cc:dd:ee:ff dev br-lan master br-lan permanent 11:22:33:44:55:66 dev eth0 vlan 1 master br-lan 33:33:00:00:00:01 dev br-lan self permanent """ UCI_EXPORT = """\ package system config system \toption hostname 'OpenWrt' \toption timezone 'UTC' package network config interface 'loopback' \toption device 'lo' \toption proto 'static' \toption ipaddr '127.0.0.1' \toption netmask '255.0.0.0' config interface 'lan' \toption device 'br-lan' \toption proto 'static' \toption ipaddr '192.168.1.1' \toption netmask '255.255.255.0' """ LLDPCTL_KV = """\ lldp.eth0.via=LLDP lldp.eth0.rid=1 lldp.eth0.age=0 day, 01:23:45 lldp.eth0.chassis.mac=00:aa:bb:cc:dd:ee lldp.eth0.chassis.name=core-router lldp.eth0.chassis.descr=RouterOS 7.x lldp.eth0.chassis.mgmt-ip=10.0.0.1 lldp.eth0.chassis.cap.available=Router, Bridge lldp.eth0.chassis.cap.enabled=Router lldp.eth0.port.ifname=ether1 lldp.eth0.port.descr=uplink """ PROC_NET_DEV = """\ Inter-| Receive | Transmit face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed lo: 1234 12 0 0 0 0 0 0 1234 12 0 0 0 0 0 0 eth0: 9876543 12345 0 0 0 0 0 100 1234567 9876 0 0 0 0 0 0 br-lan: 8765432 11234 0 0 0 0 0 50 1123456 8765 0 0 0 0 0 0 """ HOSTNAME = "OpenWrt" # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestGetFacts: def _make_send(self): """Return a _send_command mock that handles both str and list args.""" def _send(cmd, **kw): key = cmd[0] if isinstance(cmd, list) else cmd if "openwrt_release" in key: return OPENWRT_RELEASE if "sysinfo/model" in key: return SYSINFO_MODEL if "uptime" in key: return UPTIME if "hostname" in key.lower() or "system.@system" in key: return HOSTNAME if "ip link" in key: return IP_LINK_SHOW return "" return _send def test_returns_required_keys(self, driver): driver._send_command = self._make_send() facts = driver.get_facts() assert set(facts.keys()) == { "vendor", "model", "hostname", "fqdn", "os_version", "serial_number", "uptime", "interface_list", } def test_vendor(self, driver): driver._send_command = self._make_send() assert driver.get_facts()["vendor"] == "OpenWrt" def test_os_version(self, driver): driver._send_command = self._make_send() assert driver.get_facts()["os_version"] == "23.05.3" def test_uptime_parsed(self, driver): driver._send_command = self._make_send() facts = driver.get_facts() assert facts["uptime"] == pytest.approx(352467.12) class TestParseOpenwrtRelease: def test_parses_release(self): result = OpenWrtDriver._parse_openwrt_release(OPENWRT_RELEASE) assert result["DISTRIB_RELEASE"] == "23.05.3" assert result["DISTRIB_ID"] == "OpenWrt" assert result["DISTRIB_TARGET"] == "ath79/generic" class TestGetInterfaces: def test_returns_dict(self, driver): driver._send_command = lambda cmd, **kw: IP_LINK_SHOW result = driver.get_interfaces() assert isinstance(result, dict) def test_eth0_is_up(self, driver): driver._send_command = lambda cmd, **kw: IP_LINK_SHOW result = driver.get_interfaces() assert "eth0" in result assert result["eth0"]["is_up"] is True assert result["eth0"]["is_enabled"] is True def test_eth1_is_down(self, driver): driver._send_command = lambda cmd, **kw: IP_LINK_SHOW result = driver.get_interfaces() assert "eth1" in result assert result["eth1"]["is_up"] is False def test_mac_address_populated(self, driver): driver._send_command = lambda cmd, **kw: IP_LINK_SHOW result = driver.get_interfaces() assert result["eth0"]["mac_address"] != "" def test_required_keys(self, driver): driver._send_command = lambda cmd, **kw: IP_LINK_SHOW result = driver.get_interfaces() for iface_data in result.values(): assert set(iface_data.keys()) >= { "is_up", "is_enabled", "description", "last_flapped", "speed", "mtu", "mac_address", } class TestGetInterfacesIP: def test_br_lan_ipv4(self, driver): driver._send_command = lambda cmd, **kw: IP_ADDR_SHOW result = driver.get_interfaces_ip() assert "br-lan" in result assert "192.168.1.1" in result["br-lan"]["ipv4"] assert result["br-lan"]["ipv4"]["192.168.1.1"]["prefix_length"] == 24 def test_br_lan_ipv6(self, driver): driver._send_command = lambda cmd, **kw: IP_ADDR_SHOW result = driver.get_interfaces_ip() assert "ipv6" in result["br-lan"] assert "fd00::1" in result["br-lan"]["ipv6"] class TestGetArpTable: def test_returns_list(self, driver): driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW result = driver.get_arp_table() assert isinstance(result, list) def test_entries_count(self, driver): driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW result = driver.get_arp_table() # FAILED entry should be skipped assert len(result) == 2 def test_entry_keys(self, driver): driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW result = driver.get_arp_table() for entry in result: assert set(entry.keys()) >= {"interface", "mac", "ip", "age"} def test_ip_value(self, driver): driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW result = driver.get_arp_table() ips = {e["ip"] for e in result} assert "192.168.1.100" in ips assert "192.168.1.101" in ips class TestGetMacAddressTable: def test_returns_list(self, driver): driver._send_command = lambda cmd, **kw: BRIDGE_FDB result = driver.get_mac_address_table() assert isinstance(result, list) def test_multicast_skipped(self, driver): driver._send_command = lambda cmd, **kw: BRIDGE_FDB result = driver.get_mac_address_table() macs = [e["mac"] for e in result] assert not any("33:33" in m for m in macs) def test_vlan_parsed(self, driver): driver._send_command = lambda cmd, **kw: BRIDGE_FDB result = driver.get_mac_address_table() vlan1_entries = [e for e in result if e["vlan"] == 1] assert len(vlan1_entries) >= 1 class TestGetConfig: def test_returns_uci_export(self, driver): driver._send_command = lambda cmd, **kw: UCI_EXPORT result = driver.get_config() assert "running" in result assert "startup" in result assert "candidate" in result assert "package" in result["running"] def test_candidate_always_empty(self, driver): driver._send_command = lambda cmd, **kw: UCI_EXPORT result = driver.get_config() assert result["candidate"] == "" class TestGetLldpNeighbors: def test_returns_dict(self, driver): driver._send_command = lambda cmd, **kw: LLDPCTL_KV result = driver.get_lldp_neighbors() assert isinstance(result, dict) def test_eth0_neighbor(self, driver): driver._send_command = lambda cmd, **kw: LLDPCTL_KV result = driver.get_lldp_neighbors() assert "eth0" in result assert result["eth0"][0]["hostname"] == "core-router" assert result["eth0"][0]["port"] == "ether1" BRIDGE_VLAN_SHOW = """\ port vlan-id eth0 1 PVID Egress Untagged 10 20 br-lan 1 PVID Egress Untagged 10 20 eth1 20 PVID Egress Untagged """ UCI_NETWORK_VLANS = """\ network.@bridge-vlan[0]=bridge-vlan network.@bridge-vlan[0].device='br-lan' network.@bridge-vlan[0].vlan='10' network.@bridge-vlan[0].name='management' network.@bridge-vlan[1]=bridge-vlan network.@bridge-vlan[1].device='br-lan' network.@bridge-vlan[1].vlan='20' network.@bridge-vlan[1].name='iot' """ class TestGetVlans: def _send(self, cmd, **kw): if "bridge vlan" in cmd: return BRIDGE_VLAN_SHOW if "uci show network" in cmd: return UCI_NETWORK_VLANS return "" def test_returns_dict(self, driver): driver._send_command = self._send result = driver.get_vlans() assert isinstance(result, dict) def test_vlan_ids_present(self, driver): driver._send_command = self._send result = driver.get_vlans() assert "1" in result assert "10" in result assert "20" in result def test_required_keys(self, driver): driver._send_command = self._send result = driver.get_vlans() for vlan_data in result.values(): assert "name" in vlan_data assert "interfaces" in vlan_data def test_interfaces_for_vlan10(self, driver): driver._send_command = self._send result = driver.get_vlans() assert set(result["10"]["interfaces"]) == {"eth0", "br-lan"} def test_interfaces_for_vlan20(self, driver): driver._send_command = self._send result = driver.get_vlans() assert set(result["20"]["interfaces"]) == {"eth0", "br-lan", "eth1"} def test_uci_names_applied(self, driver): driver._send_command = self._send result = driver.get_vlans() assert result["10"]["name"] == "management" assert result["20"]["name"] == "iot" def test_vlan_without_uci_name_is_empty_string(self, driver): driver._send_command = self._send result = driver.get_vlans() assert result["1"]["name"] == "" def test_no_duplicate_interfaces(self, driver): driver._send_command = self._send result = driver.get_vlans() for vlan_data in result.values(): assert len(vlan_data["interfaces"]) == len(set(vlan_data["interfaces"])) def test_empty_bridge_output(self, driver): driver._send_command = lambda cmd, **kw: "" result = driver.get_vlans() assert result == {} class TestConfigManagement: def test_load_merge_candidate(self, driver): driver.load_merge_candidate(config="uci set system.@system[0].hostname='MyRouter'") assert driver._candidate_config is not None assert driver._candidate_mode == "merge" def test_load_replace_candidate(self, driver): driver.load_replace_candidate(config=UCI_EXPORT) assert driver._candidate_config is not None assert driver._candidate_mode == "replace" def test_discard_config(self, driver): driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'") driver.discard_config() assert driver._candidate_config is None assert not driver.has_pending_commit() def test_compare_merge_candidate(self, driver): driver._send_command = lambda cmd, **kw: UCI_EXPORT driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'") diff = driver.compare_config() assert diff.startswith("+") def test_compare_no_candidate(self, driver): assert driver.compare_config() == "" def test_has_pending_commit_false_initially(self, driver): assert not driver.has_pending_commit() def test_has_pending_commit_true_after_load(self, driver): driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'") assert driver.has_pending_commit() # --------------------------------------------------------------------------- # Sample data for new methods # --------------------------------------------------------------------------- PASSWD = """\ root:$1$xyz:0:0:root:/root:/bin/ash daemon:*:1:1:daemon:/var:/bin/false nobody:*:65534:65534:nobody:/var:/bin/false alice:$6$abc:1001:1001:Alice:/home/alice:/bin/ash """ ROOT_AUTHORIZED_KEYS = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABroot@host" ALICE_AUTHORIZED_KEYS = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5alice@host" UCI_SYSTEM_NTP = """\ system.@system[0]=system system.@system[0].hostname='OpenWrt' system.@system[0].timezone='UTC' system.ntp=timeserver system.ntp.server='0.openwrt.pool.ntp.org 1.openwrt.pool.ntp.org 2.openwrt.pool.ntp.org' system.ntp.enabled='1' system.ntp.enable_server='0' """ NTPQ_OUTPUT = """\ remote refid st t when poll reach delay offset jitter ============================================================================== *188.114.101.4 188.114.100.1 4 u 107 256 377 164.228 -13.866 2.695 +37.187.56.220 192.53.103.108 2 u 22 64 377 30.112 5.123 1.100 """ UCI_SNMPD = """\ snmpd.@agent[0]=agent snmpd.@agent[0].agentaddress='UDP:161' snmpd.@com2sec[0]=com2sec snmpd.@com2sec[0].secname='public' snmpd.@com2sec[0].source='default' snmpd.@com2sec[0].community='public' snmpd.@com2sec[1]=com2sec snmpd.@com2sec[1].secname='private' snmpd.@com2sec[1].source='10.0.0.0/8' snmpd.@com2sec[1].community='private' snmpd.@system[0]=system snmpd.@system[0].sysContact='admin@example.com' snmpd.@system[0].sysLocation='Server Room' snmpd.@system[0].sysName='MyRouter' """ PING_SUCCESS = """\ PING 8.8.8.8 (8.8.8.8): 56 data bytes 64 bytes from 8.8.8.8: seq=0 ttl=120 time=7.123 ms 64 bytes from 8.8.8.8: seq=1 ttl=120 time=6.987 ms 64 bytes from 8.8.8.8: seq=2 ttl=120 time=7.234 ms --- 8.8.8.8 ping statistics --- 3 packets transmitted, 3 packets received, 0% packet loss round-trip min/avg/max = 6.987/7.115/7.234 ms """ PING_LOSS = """\ PING 10.0.0.99 (10.0.0.99): 56 data bytes --- 10.0.0.99 ping statistics --- 3 packets transmitted, 0 packets received, 100% packet loss """ PING_ERROR = "ping: bad address 'invalid.host'" IPV6_NEIGH = """\ 2001:db8::1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE fe80::1 dev br-lan lladdr 11:22:33:44:55:66 STALE 2001:db8::2 dev eth0 FAILED """ IP_ROUTE_SHOW = """\ default via 192.168.1.1 dev br-wan proto dhcp src 203.0.113.1 metric 100 192.168.1.0/24 dev br-lan proto kernel scope link src 192.168.1.1 10.0.0.0/8 via 192.168.1.254 dev br-wan proto static metric 50 """ TRACEROUTE_OUTPUT = """\ traceroute to 8.8.8.8 (8.8.8.8), 30 hops max, 38 byte packets 1 192.168.1.1 (192.168.1.1) 1.123 ms 1.456 ms 1.789 ms 2 10.0.0.1 (10.0.0.1) 5.123 ms 5.456 ms 5.789 ms 3 * * * 4 8.8.8.8 (8.8.8.8) 7.001 ms 6.999 ms 7.100 ms """ TRACEROUTE_ERROR = "traceroute: unknown host invalid.host" IP_NETNS_LIST = """\ vpn (id: 1) mgmt (id: 2) """ # --------------------------------------------------------------------------- # Tests for new methods # --------------------------------------------------------------------------- class TestCli: def test_returns_dict_keyed_by_command(self, driver): driver._send_command = lambda cmd, **kw: f"output of {cmd}" result = driver.cli(["uname -a", "uptime"]) assert set(result.keys()) == {"uname -a", "uptime"} def test_output_content(self, driver): driver._send_command = lambda cmd, **kw: "Linux OpenWrt" result = driver.cli(["uname -a"]) assert result["uname -a"] == "Linux OpenWrt" def test_empty_command_list(self, driver): result = driver.cli([]) assert result == {} class TestGetUsers: def _send(self, cmd, **kw): key = cmd[0] if isinstance(cmd, list) else cmd if "/etc/passwd" in key: return PASSWD if "root/.ssh/authorized_keys" in key or "dropbear/authorized_keys" in key: return ROOT_AUTHORIZED_KEYS if "alice" in key: return ALICE_AUTHORIZED_KEYS return "" def test_returns_dict(self, driver): driver._send_command = self._send assert isinstance(driver.get_users(), dict) def test_root_level_15(self, driver): driver._send_command = self._send users = driver.get_users() assert "root" in users assert users["root"]["level"] == 15 def test_regular_user_level_1(self, driver): driver._send_command = self._send users = driver.get_users() assert "alice" in users assert users["alice"]["level"] == 1 def test_system_accounts_excluded(self, driver): driver._send_command = self._send users = driver.get_users() assert "daemon" not in users assert "nobody" not in users def test_root_has_ssh_key(self, driver): driver._send_command = self._send users = driver.get_users() assert len(users["root"]["sshkeys"]) >= 1 assert users["root"]["sshkeys"][0].startswith("ssh-rsa") def test_required_keys(self, driver): driver._send_command = self._send users = driver.get_users() for data in users.values(): assert "level" in data assert "password" in data assert "sshkeys" in data class TestGetNtpServers: def test_returns_dict(self, driver): driver._send_command = lambda cmd, **kw: UCI_SYSTEM_NTP result = driver.get_ntp_servers() assert isinstance(result, dict) def test_servers_present(self, driver): driver._send_command = lambda cmd, **kw: UCI_SYSTEM_NTP result = driver.get_ntp_servers() assert "0.openwrt.pool.ntp.org" in result assert "1.openwrt.pool.ntp.org" in result assert "2.openwrt.pool.ntp.org" in result def test_empty_when_no_ntp(self, driver): driver._send_command = lambda cmd, **kw: "" assert driver.get_ntp_servers() == {} class TestGetNtpStats: def test_returns_list(self, driver): driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT result = driver.get_ntp_stats() assert isinstance(result, list) def test_synchronized_entry(self, driver): driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT result = driver.get_ntp_stats() synced = [e for e in result if e["synchronized"]] assert len(synced) == 1 assert synced[0]["remote"] == "188.114.101.4" def test_required_keys(self, driver): driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT result = driver.get_ntp_stats() for entry in result: assert set(entry.keys()) >= { "remote", "referenceid", "synchronized", "stratum", "type", "when", "hostpoll", "reachability", "delay", "offset", "jitter", } def test_empty_on_no_tool(self, driver): driver._send_command = lambda cmd, **kw: "sh: ntpq: not found" result = driver.get_ntp_stats() assert result == [] class TestGetSnmpInformation: def test_returns_dict(self, driver): driver._send_command = lambda cmd, **kw: UCI_SNMPD result = driver.get_snmp_information() assert isinstance(result, dict) def test_required_keys(self, driver): driver._send_command = lambda cmd, **kw: UCI_SNMPD result = driver.get_snmp_information() assert set(result.keys()) >= {"chassis_id", "community", "contact", "location"} def test_contact_and_location(self, driver): driver._send_command = lambda cmd, **kw: UCI_SNMPD result = driver.get_snmp_information() assert result["contact"] == "admin@example.com" assert result["location"] == "Server Room" def test_community_entries(self, driver): driver._send_command = lambda cmd, **kw: UCI_SNMPD result = driver.get_snmp_information() assert "public" in result["community"] assert "private" in result["community"] def test_community_mode(self, driver): driver._send_command = lambda cmd, **kw: UCI_SNMPD result = driver.get_snmp_information() assert result["community"]["public"]["mode"] == "ro" assert result["community"]["private"]["mode"] == "rw" class TestPing: def test_success_result(self, driver): driver._send_command = lambda cmd, **kw: PING_SUCCESS result = driver.ping("8.8.8.8") assert "success" in result assert result["success"]["probes_sent"] == 3 assert result["success"]["packet_loss"] == 0 def test_rtt_values(self, driver): driver._send_command = lambda cmd, **kw: PING_SUCCESS result = driver.ping("8.8.8.8") s = result["success"] assert s["rtt_min"] == 6.987 assert s["rtt_max"] == 7.234 assert s["rtt_avg"] == 7.115 def test_probe_results(self, driver): driver._send_command = lambda cmd, **kw: PING_SUCCESS result = driver.ping("8.8.8.8") assert len(result["success"]["results"]) == 3 assert result["success"]["results"][0]["ip_address"] == "8.8.8.8" def test_100_percent_loss(self, driver): driver._send_command = lambda cmd, **kw: PING_LOSS result = driver.ping("10.0.0.99") assert "success" in result assert result["success"]["packet_loss"] == 3 def test_error_on_bad_host(self, driver): driver._send_command = lambda cmd, **kw: PING_ERROR result = driver.ping("invalid.host") assert "error" in result class TestGetIpv6NeighborsTable: def test_returns_list(self, driver): driver._send_command = lambda cmd, **kw: IPV6_NEIGH assert isinstance(driver.get_ipv6_neighbors_table(), list) def test_failed_entries_excluded(self, driver): driver._send_command = lambda cmd, **kw: IPV6_NEIGH result = driver.get_ipv6_neighbors_table() assert len(result) == 2 def test_required_keys(self, driver): driver._send_command = lambda cmd, **kw: IPV6_NEIGH for entry in driver.get_ipv6_neighbors_table(): assert set(entry.keys()) >= {"interface", "mac", "ip", "age", "state"} def test_state_values(self, driver): driver._send_command = lambda cmd, **kw: IPV6_NEIGH states = {e["state"] for e in driver.get_ipv6_neighbors_table()} assert "REACHABLE" in states assert "STALE" in states class TestGetRouteTo: def test_returns_dict(self, driver): driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW assert isinstance(driver.get_route_to(), dict) def test_default_route_present(self, driver): driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW result = driver.get_route_to() assert "0.0.0.0/0" in result def test_next_hop(self, driver): driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW result = driver.get_route_to() default = result["0.0.0.0/0"][0] assert default["next_hop"] == "192.168.1.1" assert default["outgoing_interface"] == "br-wan" def test_connected_route(self, driver): driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW result = driver.get_route_to() assert "192.168.1.0/24" in result assert result["192.168.1.0/24"][0]["protocol"] == "connected" def test_protocol_filter(self, driver): driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW result = driver.get_route_to(protocol="static") assert "10.0.0.0/8" in result assert "192.168.1.0/24" not in result def test_required_keys(self, driver): driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW for prefix, routes in driver.get_route_to().items(): for route in routes: assert set(route.keys()) >= { "protocol", "current_active", "next_hop", "outgoing_interface", "preference", "routing_table", } class TestTraceroute: def test_success_result(self, driver): driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT result = driver.traceroute("8.8.8.8") assert "success" in result def test_hop_count(self, driver): driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT result = driver.traceroute("8.8.8.8") assert len(result["success"]) == 4 def test_hop_1_rtt(self, driver): driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT result = driver.traceroute("8.8.8.8") hop1 = result["success"][1]["probes"][1] assert hop1["rtt"] == 1.123 assert hop1["ip_address"] == "192.168.1.1" def test_star_hop(self, driver): driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT result = driver.traceroute("8.8.8.8") hop3 = result["success"][3]["probes"][1] assert hop3["ip_address"] == "*" def test_error_on_unknown_host(self, driver): driver._send_command = lambda cmd, **kw: TRACEROUTE_ERROR result = driver.traceroute("invalid.host") assert "error" in result class TestGetNetworkInstances: def test_default_instance_always_present(self, driver): driver._send_command = lambda cmd, **kw: ( IP_LINK_SHOW if "ip link" in cmd else "" ) result = driver.get_network_instances() assert "default" in result def test_default_instance_type(self, driver): driver._send_command = lambda cmd, **kw: ( IP_LINK_SHOW if "ip link" in cmd else "" ) result = driver.get_network_instances() assert result["default"]["type"] == "DEFAULT_INSTANCE" def test_default_interfaces_populated(self, driver): driver._send_command = lambda cmd, **kw: ( IP_LINK_SHOW if "ip link" in cmd else "" ) result = driver.get_network_instances() ifaces = result["default"]["interfaces"]["interface"] assert "eth0" in ifaces def test_named_namespaces(self, driver): def _send(cmd, **kw): if "netns list" in cmd: return IP_NETNS_LIST if "netns exec" in cmd: return "" # empty namespace if "ip link" in cmd: return IP_LINK_SHOW return "" driver._send_command = _send result = driver.get_network_instances() assert "vpn" in result assert "mgmt" in result assert result["vpn"]["type"] == "L3VRF" def test_name_filter(self, driver): driver._send_command = lambda cmd, **kw: ( IP_LINK_SHOW if "ip link" in cmd else "" ) result = driver.get_network_instances(name="default") assert list(result.keys()) == ["default"] def test_required_keys(self, driver): driver._send_command = lambda cmd, **kw: ( IP_LINK_SHOW if "ip link" in cmd else "" ) for inst in driver.get_network_instances().values(): assert set(inst.keys()) >= {"name", "type", "state", "interfaces"}