NAPALM/napalm-openwrt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d45c082355b69a231ab8c32fe09a34a369df3c3c
napalm-openwrt/tests/unit/test_driver.py
T
Christian Manivong d45c082355 feat(system): report remote syslog, LuCI state and bridge STP 
get_system_config() already parsed the whole system section but returned
only a slice of it, so netOrk had no IST side for three AP-profile fields
and could not compare them at all — the fields were editable, stored and
silently ineffective.

Adds:
  * syslog_remote / syslog_ip / syslog_port / syslog_proto — read from the
    system section that was already being parsed. log_remote gates the
    others: OpenWrt ships nothing without it, so a leftover log_ip must
    not read as an active target.
  * luci_enabled — from the uhttpd init script rather than its listener
    config, so the answer stays "is the web UI served" and re-enabling
    restores whatever was configured before.
  * bridge_stp — None when there is no br-ap at all, which is a different
    statement from "bridge without STP" and has to stay distinguishable.

netOrk #164
2026-08-29 22:41:36 +07:00

1870 lines
68 KiB
Python
Raw Blame History

"""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: <LOOPBACK,UP,LOWER_UP> 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: <BROADCAST,MULTICAST,UP,LOWER_UP> 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: <BROADCAST,MULTICAST> 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: <BROADCAST,MULTICAST,UP,LOWER_UP> 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: <LOOPBACK,UP,LOWER_UP> 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: <BROADCAST,MULTICAST,UP,LOWER_UP> 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: <BROADCAST,MULTICAST,UP,LOWER_UP> 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", "number_of_interfaces",
}
def test_number_of_interfaces_matches_list(self, driver):
driver._send_command = self._make_send()
facts = driver.get_facts()
assert facts["number_of_interfaces"] == len(facts["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 "tagged" in vlan_data
assert "untagged" in vlan_data
def test_pvid_port_is_untagged_member(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
assert set(result["1"]["untagged"]) == {"eth0", "br-lan"}
assert result["1"]["tagged"] == []
def test_continuation_lines_are_tagged_members(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
assert set(result["10"]["tagged"]) == {"eth0", "br-lan"}
assert result["10"]["untagged"] == []
def test_same_vlan_can_mix_tagged_and_untagged_ports(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
assert set(result["20"]["tagged"]) == {"eth0", "br-lan"}
assert set(result["20"]["untagged"]) == {"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_ports(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
for vlan_data in result.values():
for key in ("tagged", "untagged"):
assert len(vlan_data[key]) == len(set(vlan_data[key]))
def test_port_is_never_both_tagged_and_untagged(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
for vlan_data in result.values():
assert not set(vlan_data["tagged"]) & set(vlan_data["untagged"])
def test_empty_bridge_output(self, driver):
driver._send_command = lambda cmd, **kw: ""
result = driver.get_vlans()
assert result == {}
# A VLAN-aware bridge as configured by an AP profile: management VLAN untagged
# (native/PVID) on the uplink, SSID VLANs tagged. Sections are *named*, which
# is what OpenWrt writes for a hand-built or tool-provisioned bridge.
UCI_NAMED_BRIDGE_VLANS = """\
network.apbr=device
network.apbr.name='br-ap'
network.apbr.type='bridge'
network.apbr.ports='eth0'
network.apbr.vlan_filtering='1'
network.apbr_vlan10=bridge-vlan
network.apbr_vlan10.device='br-ap'
network.apbr_vlan10.vlan='10'
network.apbr_vlan10.ports='eth0:u*'
network.apbr_vlan30=bridge-vlan
network.apbr_vlan30.device='br-ap'
network.apbr_vlan30.vlan='30'
network.apbr_vlan30.ports='eth0:t'
network.mgmt=interface
network.mgmt.device='br-ap.10'
network.mgmt.proto='dhcp'
network.esche=interface
network.esche.device='br-ap.30'
network.esche.proto='none'
"""
IP_LINK_AP = """\
1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel master br-ap
3: br-ap: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue
4: br-ap.10@br-ap: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue
5: br-ap.30@br-ap: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue
"""
class TestGetVlansWithoutBridgeBinary:
"""Membership must come from UCI when `bridge` is absent (BusyBox-only APs).
Devices such as the Sophos AP100 ship BusyBox without the `bridge`/`ip-full`
packages, so `bridge vlan show` fails. Falling back to sub-interface
topology alone reports `br-ap`/`br-ap.10` and never names the uplink port,
which is the only port whose tagging actually matters.
"""
def _send(self, cmd, **kw):
if "bridge vlan" in cmd:
return "ash: bridge: not found"
if "uci show network" in cmd:
return UCI_NAMED_BRIDGE_VLANS
if "ip link show" in cmd:
return IP_LINK_AP
return ""
def test_untagged_pvid_uplink_membership(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
assert "eth0" in result["10"]["untagged"]
assert "eth0" not in result["10"]["tagged"]
def test_tagged_uplink_membership(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
assert "eth0" in result["30"]["tagged"]
assert "eth0" not in result["30"]["untagged"]
def test_named_sections_are_parsed(self, driver):
"""The old parser only matched anonymous @bridge-vlan[N] sections."""
driver._send_command = self._send
result = driver.get_vlans()
assert {"10", "30"} <= set(result)
def test_shell_error_is_not_parsed_as_membership(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
for vlan_data in result.values():
assert "ash:" not in vlan_data["tagged"] + vlan_data["untagged"]
def test_name_falls_back_to_interface_section(self, driver):
driver._send_command = self._send
result = driver.get_vlans()
assert result["10"]["name"] == "mgmt"
assert result["30"]["name"] == "esche"
def test_runtime_membership_wins_over_uci(self, driver):
"""`bridge vlan show` describes what the kernel actually does."""
def _send(cmd, **kw):
if "bridge vlan" in cmd:
return "port vlan-id\neth0 10 PVID Egress Untagged\n"
if "uci show network" in cmd:
# UCI claims tagged; the kernel says untagged.
return (
"network.apbr_vlan10=bridge-vlan\n"
"network.apbr_vlan10.device='br-ap'\n"
"network.apbr_vlan10.vlan='10'\n"
"network.apbr_vlan10.ports='eth0:t'\n"
)
return ""
driver._send_command = _send
result = driver.get_vlans()
assert "eth0" in result["10"]["untagged"]
assert "eth0" not in result["10"]["tagged"]
def test_multi_port_list_is_split(self, driver):
def _send(cmd, **kw):
if "uci show network" in cmd:
return (
"network.brv=bridge-vlan\n"
"network.brv.device='br-lan'\n"
"network.brv.vlan='20'\n"
"network.brv.ports='lan1:t' 'lan2:t' 'lan3'\n"
)
return ""
driver._send_command = _send
result = driver.get_vlans()
assert set(result["20"]["tagged"]) == {"lan1", "lan2"}
assert set(result["20"]["untagged"]) == {"lan3"}
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"}
# ---------------------------------------------------------------------------
# Sample data for get_channel_scan()
# ---------------------------------------------------------------------------
IW_DEV_OUT = """\
phy#0
\tInterface wlan0
\t\tifindex 4
\t\twdev 0x1
\t\taddr aa:bb:cc:dd:ee:f0
\t\tssid HomeNet
\t\ttype AP
\t\tchannel 6 (2437 MHz), width: 20 MHz, center1: 2437 MHz
phy#1
\tInterface wlan1
\t\tifindex 5
\t\twdev 0x2
\t\taddr aa:bb:cc:dd:ee:f1
\t\tssid HomeNet
\t\ttype AP
\t\tchannel 36 (5180 MHz), width: 80 MHz, center1: 5210 MHz
\tInterface wlan2
\t\tifindex 6
\t\twdev 0x3
\t\taddr aa:bb:cc:dd:ee:f2
\t\tssid GuestNet
\t\ttype managed
"""
# Minimal iw scan output for wlan0 (2.4 GHz AP, ch 6 + ch 1)
IW_SCAN_WLAN0 = """\
BSS aa:bb:cc:dd:01:ff(on wlan0)
\tfreq: 2437
\tsignal: -72.00 dBm
\tSSID: NeighborNet1
\tHT operation:
\t\t * primary channel: 6
\t\t * secondary channel offset: no secondary
\t\t * STA channel width: 20 MHz
BSS aa:bb:cc:dd:02:ff(on wlan0)
\tfreq: 2412
\tsignal: -85.00 dBm
\tSSID: FarAway
\tHT operation:
\t\t * primary channel: 1
\t\t * secondary channel offset: no secondary
\t\t * STA channel width: 20 MHz
BSS aa:bb:cc:dd:03:ff(on wlan0)
\tfreq: 2452
\tsignal: -90.00 dBm
\tSSID: Hidden
\tHT operation:
\t\t * primary channel: 9
\t\t * secondary channel offset: no secondary
\t\t * STA channel width: 20 MHz
"""
# Minimal iw scan output for wlan1 (5 GHz AP, ch 36 + 44)
IW_SCAN_WLAN1 = """\
BSS 11:22:33:44:01:ff(on wlan1)
\tfreq: 5180
\tsignal: -58.00 dBm
\tSSID: Corp5G
\tVHT operation:
\t\t * channel width: 80 MHz
\t\t * center freq segment 1: 42
\t\t * center freq segment 2: 0
BSS 11:22:33:44:02:ff(on wlan1)
\tfreq: 5220
\tsignal: -77.00 dBm
\tSSID: Office5G
\tVHT operation:
\t\t * channel width: 80 MHz
"""
class TestGetChannelScan:
"""Tests for OpenWrtWirelessMixin.get_channel_scan()."""
def _make_send(self, mode: str = "active"):
"""Return a _send_command mock routing iw dev / iw scan output."""
def _send(cmd, **kw):
if cmd.strip().startswith("iw dev") and "scan" not in cmd:
return IW_DEV_OUT
if "wlan0" in cmd and "scan" in cmd:
return IW_SCAN_WLAN0
if "wlan1" in cmd and "scan" in cmd:
return IW_SCAN_WLAN1
return ""
return _send
def test_returns_dict(self, driver):
driver._send_command = self._make_send()
result = driver.get_channel_scan()
assert isinstance(result, dict)
def test_only_ap_interfaces_included(self, driver):
"""wlan2 is managed (client) mode — must not appear in results."""
driver._send_command = self._make_send()
result = driver.get_channel_scan()
assert "wlan0" in result
assert "wlan1" in result
assert "wlan2" not in result
def test_wlan0_entry_count(self, driver):
driver._send_command = self._make_send()
result = driver.get_channel_scan()
assert len(result["wlan0"]) == 3
def test_wlan0_entry_required_keys(self, driver):
driver._send_command = self._make_send()
entry = driver.get_channel_scan()["wlan0"][0]
assert set(entry.keys()) >= {
"bssid", "ssid", "frequency", "channel", "signal_dbm",
"channel_width", "band",
}
def test_wlan0_first_entry_values(self, driver):
driver._send_command = self._make_send()
entry = driver.get_channel_scan()["wlan0"][0]
assert entry["bssid"] == "aa:bb:cc:dd:01:ff"
assert entry["ssid"] == "NeighborNet1"
assert entry["frequency"] == 2437
assert entry["channel"] == 6
assert entry["signal_dbm"] == -72
assert entry["channel_width"] == 20
assert entry["band"] == "2.4GHz"
def test_wlan0_ch1_entry(self, driver):
driver._send_command = self._make_send()
entries = driver.get_channel_scan()["wlan0"]
ch1 = next(e for e in entries if e["channel"] == 1)
assert ch1["signal_dbm"] == -85
assert ch1["band"] == "2.4GHz"
def test_wlan0_hidden_ssid(self, driver):
"""Empty SSID (hidden) should be stored as empty string."""
driver._send_command = self._make_send()
entries = driver.get_channel_scan()["wlan0"]
ch9 = next(e for e in entries if e["channel"] == 9)
assert ch9["ssid"] == "Hidden"
def test_wlan1_band_is_5ghz(self, driver):
driver._send_command = self._make_send()
entries = driver.get_channel_scan()["wlan1"]
assert all(e["band"] == "5GHz" for e in entries)
def test_wlan1_80mhz_width(self, driver):
driver._send_command = self._make_send()
entry = driver.get_channel_scan()["wlan1"][0]
assert entry["channel_width"] == 80
def test_passive_mode_uses_passive_flag(self, driver):
"""With mode='passive', the scan command must contain 'passive'."""
issued_cmds: list[str] = []
def _send(cmd, **kw):
issued_cmds.append(cmd)
if cmd.strip().startswith("iw dev") and "scan" not in cmd:
return IW_DEV_OUT
return ""
driver._send_command = _send
driver.get_channel_scan(mode="passive")
scan_cmds = [c for c in issued_cmds if "scan" in c]
assert all("passive" in c for c in scan_cmds)
def test_active_mode_no_passive_flag(self, driver):
"""With mode='active' (default), scan command must NOT contain 'passive'."""
issued_cmds: list[str] = []
def _send(cmd, **kw):
issued_cmds.append(cmd)
if cmd.strip().startswith("iw dev") and "scan" not in cmd:
return IW_DEV_OUT
return ""
driver._send_command = _send
driver.get_channel_scan(mode="active")
scan_cmds = [c for c in issued_cmds if "scan" in c]
assert all("passive" not in c for c in scan_cmds)
def test_no_ap_interfaces_returns_empty(self, driver):
"""If no AP-mode interfaces exist, return empty dict."""
no_ap = IW_DEV_OUT.replace("type AP", "type managed")
driver._send_command = lambda cmd, **kw: (
no_ap if "iw dev" in cmd and "scan" not in cmd else ""
)
result = driver.get_channel_scan()
assert result == {}
def test_empty_iw_output_returns_empty(self, driver):
driver._send_command = lambda cmd, **kw: ""
result = driver.get_channel_scan()
assert result == {}
def test_6ghz_band_detected(self, driver):
iw_dev_6g = """\
phy#0
\tInterface wlan0
\t\taddr aa:bb:cc:dd:ee:f0
\t\tssid Net6G
\t\ttype AP
"""
scan_6g = """\
BSS aa:bb:cc:dd:01:ff(on wlan0)
\tfreq: 5975
\tsignal: -65.00 dBm
\tSSID: Neighbor6G
"""
def _send(cmd, **kw):
if "iw dev" in cmd and "scan" not in cmd:
return iw_dev_6g
if "scan" in cmd:
return scan_6g
return ""
driver._send_command = _send
entries = driver.get_channel_scan().get("wlan0", [])
assert len(entries) == 1
assert entries[0]["band"] == "6GHz"
assert entries[0]["frequency"] == 5975
class TestPushRadioChannel:
"""Tests for OpenWrtWirelessMixin.push_radio_channel()."""
def test_issues_uci_set(self, driver):
"""Must call 'uci set wireless.radio0.channel=11'."""
issued: list[str] = []
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
driver.push_radio_channel("radio0", 11)
assert any("uci set wireless.radio0.channel=11" in c for c in issued)
def test_issues_uci_commit(self, driver):
issued: list[str] = []
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
driver.push_radio_channel("radio0", 11)
assert any("uci commit wireless" in c for c in issued)
def test_issues_wifi_restart(self, driver):
"""Must call 'wifi' (full restart) — NOT just 'wifi reload'."""
issued: list[str] = []
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
driver.push_radio_channel("radio1", 36)
# 'wifi' alone is a full restart; 'wifi reload' is insufficient for channel changes
assert any(c.strip() == "wifi" for c in issued)
def test_auto_channel_zero(self, driver):
"""channel=0 should set 'auto' in UCI."""
issued: list[str] = []
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
driver.push_radio_channel("radio0", 0)
assert any("channel=auto" in c or "channel='auto'" in c for c in issued)
def test_commit_before_restart(self, driver):
"""UCI commit must happen before wifi restart."""
issued: list[str] = []
driver._send_command = lambda cmd, **kw: issued.append(cmd) or ""
driver.push_radio_channel("radio0", 6)
commit_idx = next(i for i, c in enumerate(issued) if "commit" in c)
wifi_idx = next(i for i, c in enumerate(issued) if c.strip() == "wifi")
assert commit_idx < wifi_idx
UCI_WIRELESS_ACL = """\
wireless.radio0=wifi-device
wireless.radio0.band='2g'
wireless.radio1=wifi-device
wireless.radio1.band='5g'
wireless.@wifi-iface[0]=wifi-iface
wireless.@wifi-iface[0].device='radio0'
wireless.@wifi-iface[0].ssid='CorpWiFi'
wireless.@wifi-iface[0].encryption='psk2'
wireless.@wifi-iface[0].macfilter='allow'
wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:01'
wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:02'
wireless.@wifi-iface[1]=wifi-iface
wireless.@wifi-iface[1].device='radio1'
wireless.@wifi-iface[1].ssid='CorpWiFi'
wireless.@wifi-iface[1].encryption='psk2'
wireless.@wifi-iface[2]=wifi-iface
wireless.@wifi-iface[2].device='radio0'
wireless.@wifi-iface[2].ssid='GuestNet'
wireless.@wifi-iface[2].encryption='none'
"""
class TestGetSsidsAcl:
"""Tests for the macfilter/maclist parsing in OpenWrtWirelessMixin.get_ssids()."""
def _send(self, cmd, **kw):
if cmd.strip() == "uci show wireless":
return UCI_WIRELESS_ACL
return ""
def test_whitelist_mode_parsed(self, driver):
driver._send_command = self._send
result = driver.get_ssids()
assert result["CorpWiFi"]["acl_mode"] == "whitelist"
def test_maclist_multiple_entries_parsed(self, driver):
"""UCI list values repeat the same key across lines — must not overwrite."""
driver._send_command = self._send
result = driver.get_ssids()
assert result["CorpWiFi"]["mac_list"] == ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"]
def test_acl_mode_merged_across_radios(self, driver):
"""Only wifi-iface[0] has macfilter set; wifi-iface[1] (same SSID) must inherit it."""
driver._send_command = self._send
result = driver.get_ssids()
# Both wifi-iface sections belong to CorpWiFi — the merged result carries one acl_mode.
assert result["CorpWiFi"]["acl_mode"] == "whitelist"
assert result["CorpWiFi"]["mac_list"] == ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"]
def test_no_macfilter_defaults_to_off(self, driver):
driver._send_command = self._send
result = driver.get_ssids()
assert result["GuestNet"]["acl_mode"] == "off"
assert result["GuestNet"]["mac_list"] == []
def test_deny_maps_to_blacklist(self, driver):
deny_uci = UCI_WIRELESS_ACL.replace("macfilter='allow'", "macfilter='deny'")
driver._send_command = lambda cmd, **kw: deny_uci if cmd.strip() == "uci show wireless" else ""
result = driver.get_ssids()
assert result["CorpWiFi"]["acl_mode"] == "blacklist"
class TestPushMacAcl:
"""Tests for OpenWrtWirelessMixin.push_mac_acl()."""
def _make_send(self, sections="wireless.@wifi-iface[0]\nwireless.@wifi-iface[1]", ssid_by_section=None):
ssid_by_section = ssid_by_section or {
"wireless.@wifi-iface[0]": "CorpWiFi",
"wireless.@wifi-iface[1]": "GuestNet",
}
issued: list[str] = []
def _send(cmd, **kw):
issued.append(cmd)
if "grep -oE" in cmd and "sort -u" in cmd:
return sections
for sec, ssid in ssid_by_section.items():
if f"uci -q get {sec}.ssid" in cmd:
return ssid
return ""
return _send, issued
def test_whitelist_sets_macfilter_allow(self, driver):
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
assert any("wireless.@wifi-iface[0].macfilter='allow'" in c for c in issued)
def test_blacklist_sets_macfilter_deny(self, driver):
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "blacklist", ["AA:BB:CC:DD:EE:01"])
assert any("wireless.@wifi-iface[0].macfilter='deny'" in c for c in issued)
def test_off_deletes_macfilter_option_instead_of_setting_disable(self, driver):
"""OpenWrt's validator rejects macfilter='disable' outright (confirmed on
real hardware — it puts netifd in a permanent restart crash loop with
the radio stuck down). "off" must delete the option, never set it."""
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "off", [])
assert not any("macfilter='disable'" in c for c in issued)
assert any("delete wireless.@wifi-iface[0].macfilter" in c for c in issued)
assert any("delete wireless.@wifi-iface[0].maclist" in c for c in issued)
assert not any("add_list wireless.@wifi-iface[0].maclist" in c for c in issued)
def test_whitelist_with_zero_macs_treated_as_off(self, driver):
"""An empty whitelist blocks every client outright — must not be pushed
as macfilter='allow' with an empty list."""
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "whitelist", [])
assert not any("macfilter='allow'" in c for c in issued)
assert any("delete wireless.@wifi-iface[0].macfilter" in c for c in issued)
def test_blacklist_with_zero_macs_still_pushed(self, driver):
"""An empty blacklist is safe (blocks nobody) — no guard needed."""
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "blacklist", [])
assert any("macfilter='deny'" in c for c in issued)
def test_maclist_entries_added(self, driver):
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01", "AA:BB:CC:DD:EE:02"])
assert any("add_list wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:01'" in c for c in issued)
assert any("add_list wireless.@wifi-iface[0].maclist='AA:BB:CC:DD:EE:02'" in c for c in issued)
def test_maclist_cleared_before_readd(self, driver):
"""Full-rebuild: existing maclist must be deleted before new entries are added."""
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
delete_idx = next(i for i, c in enumerate(issued) if "delete wireless.@wifi-iface[0].maclist" in c)
add_idx = next(i for i, c in enumerate(issued) if "add_list wireless.@wifi-iface[0].maclist" in c)
assert delete_idx < add_idx
def test_only_matching_ssid_sections_touched(self, driver):
"""GuestNet section must not be modified when pushing CorpWiFi's ACL."""
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
assert not any("wireless.@wifi-iface[1].macfilter" in c for c in issued)
def test_issues_uci_commit_and_wifi_reload(self, driver):
send, issued = self._make_send()
driver._send_command = send
driver.push_mac_acl("CorpWiFi", "whitelist", ["AA:BB:CC:DD:EE:01"])
assert any("uci commit wireless" in c for c in issued)
assert any(c.strip() == "wifi reload" for c in issued)
# ---------------------------------------------------------------------------
# fix_snmp — firewall zone handling
# ---------------------------------------------------------------------------
UCI_FIREWALL_HEALTHY = """\
firewall.@defaults[0]=defaults
firewall.@defaults[0].input='REJECT'
firewall.@zone[0]=zone
firewall.@zone[0].name='lan'
firewall.@zone[0].network='lan'
firewall.@zone[0].input='ACCEPT'
firewall.@zone[1]=zone
firewall.@zone[1].name='wan'
firewall.@zone[1].network='wan' 'wan6'
firewall.@zone[1].input='REJECT'
firewall.@rule[0]=rule
firewall.@rule[0].name='Allow-DHCP-Renew'
firewall.@rule[0].src='wan'
firewall.@rule[0].dest_port='68'
"""
# The zone that owns the management network lost its 'name' — fw4 skips the
# whole section and every rule pointing at it.
UCI_FIREWALL_NAMELESS_ZONE = """\
firewall.@defaults[0]=defaults
firewall.@defaults[0].input='REJECT'
firewall.@zone[0]=zone
firewall.@zone[0].network='lan'
firewall.@zone[0].input='ACCEPT'
firewall.@rule[0]=rule
firewall.@rule[0].name='Allow-DHCP-Renew'
firewall.@rule[0].src='wan'
"""
# A dedicated management zone — the AP layout the action is meant to handle.
UCI_FIREWALL_MGMT_ZONE = """\
firewall.@zone[0]=zone
firewall.@zone[0].name='lan'
firewall.@zone[0].network='lan'
firewall.@zone[1]=zone
firewall.@zone[1].name='mgmt'
firewall.@zone[1].network='mgmt'
firewall.@zone[1].input='REJECT'
firewall.@rule[0]=rule
firewall.@rule[0].name='Allow-SSH'
firewall.@rule[0].src='mgmt'
firewall.@rule[0].dest_port='22'
"""
UCI_NETWORK_STATIC = """\
network.loopback=interface
network.loopback.device='lo'
network.lan=interface
network.lan.device='br-lan'
network.lan.proto='static'
network.lan.ipaddr='192.168.1.1'
network.mgmt=interface
network.mgmt.device='br-lan.9'
network.mgmt.proto='static'
network.mgmt.ipaddr='10.10.0.5'
"""
UCI_NETWORK_DHCP = """\
network.lan=interface
network.lan.device='br-lan'
network.lan.proto='dhcp'
"""
IP_ADDR_BRLAN = """\
1: lo inet 127.0.0.1/8 scope host lo\\ valid_lft forever preferred_lft forever
7: br-lan inet 10.10.0.5/24 brd 10.10.0.255 scope global br-lan\\ valid_lft forever
"""
class _FakeShell:
"""Collects issued commands and answers them from a canned config."""
def __init__(self, firewall="", network="", ip_addr="", ssh_connection="",
reload_out="", nft_hits="1", listen_hits="1", snmpd="running"):
self.firewall = firewall
self.network = network
self.ip_addr = ip_addr
self.ssh_connection = ssh_connection
self.reload_out = reload_out
self.nft_hits = nft_hits
self.listen_hits = listen_hits
self.snmpd = snmpd
self.issued: list[str] = []
def __call__(self, cmd, **kw):
self.issued.append(cmd)
if cmd.startswith("uci show firewall"):
return self.firewall
if cmd.startswith("uci show network"):
return self.network
if "$SSH_CONNECTION" in cmd:
return self.ssh_connection
if "ip -o -4 addr" in cmd:
return self.ip_addr
if "fw4 reload" in cmd or "firewall reload" in cmd:
return self.reload_out
if "dport 161" in cmd:
return self.nft_hits
if ":161" in cmd:
return self.listen_hits
if "snmpd status" in cmd:
return self.snmpd
return ""
class TestFixSnmpZoneDetection:
"""_action_fix_snmp() must resolve the real zone that owns the mgmt address."""
def test_static_mgmt_address_selects_owning_zone(self, driver):
"""10.10.0.5 lives on network 'mgmt' → zone 'mgmt', not the 'lan' fallback."""
shell = _FakeShell(
firewall=UCI_FIREWALL_MGMT_ZONE,
network=UCI_NETWORK_STATIC,
ssh_connection="10.10.0.1 51234 10.10.0.5 22",
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert "Management zone: 'mgmt'" in result["output"]
assert any("firewall.allow_snmp_from_mgmt.src='mgmt'" in c for c in shell.issued)
def test_anonymous_ssh_rule_does_not_decide_the_zone(self, driver):
"""The old per-line 'src= and ssh' heuristic never matched anonymous rules."""
shell = _FakeShell(
firewall=UCI_FIREWALL_HEALTHY,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert "Management zone: 'lan'" in result["output"]
def test_dhcp_mgmt_address_resolved_via_l3_device(self, driver):
"""No ipaddr in UCI → resolve address → device → network section → zone."""
shell = _FakeShell(
firewall=UCI_FIREWALL_HEALTHY,
network=UCI_NETWORK_DHCP,
ip_addr=IP_ADDR_BRLAN,
ssh_connection="10.10.0.1 51234 10.10.0.5 22",
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert "Management zone: 'lan'" in result["output"]
def test_nameless_zone_is_reported_and_action_fails(self, driver):
"""A zone without 'name' is skipped by fw4 — say so instead of writing a dead rule."""
shell = _FakeShell(
firewall=UCI_FIREWALL_NAMELESS_ZONE,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert result["success"] is False
assert "@zone[0]" in result["output"]
assert "name" in result["output"]
def test_nameless_zone_does_not_get_a_rule_written(self, driver):
"""No SNMP rule may be committed while the owning zone is invalid."""
shell = _FakeShell(
firewall=UCI_FIREWALL_NAMELESS_ZONE,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
driver._action_fix_snmp()
assert not any("allow_snmp" in c for c in shell.issued)
class TestFixSnmpRuleWriting:
"""The rule must be (re)written idempotently, not skipped when present."""
def test_existing_rule_is_repaired_not_skipped(self, driver):
"""A rule that exists but lacks src must be rewritten, not left broken."""
broken = UCI_FIREWALL_HEALTHY + (
"firewall.allow_snmp_from_lan=rule\n"
"firewall.allow_snmp_from_lan.name='Allow-SNMP-from-lan'\n"
"firewall.allow_snmp_from_lan.dest_port='161'\n"
)
shell = _FakeShell(
firewall=broken,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
driver._action_fix_snmp()
assert any("firewall.allow_snmp_from_lan.src='lan'" in c for c in shell.issued)
def test_stale_rules_are_deleted_highest_index_first(self, driver):
"""Anonymous sections shift on delete — descending order keeps the keys valid."""
stale = UCI_FIREWALL_HEALTHY + (
"firewall.@rule[1]=rule\n"
"firewall.@rule[1].name='Allow-SNMP'\n"
"firewall.@rule[1].dest_port='161'\n"
"firewall.@rule[2]=rule\n"
"firewall.@rule[2].name='Allow-SNMP-netOrk'\n"
"firewall.@rule[2].dest_port='161'\n"
)
shell = _FakeShell(
firewall=stale,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
driver._action_fix_snmp()
deletes = [c for c in shell.issued if "delete" in c and "@rule" in c]
joined = " ".join(deletes)
assert joined.index("@rule[2]") < joined.index("@rule[1]")
def test_deletion_is_committed(self, driver):
"""Old code staged deletes in /tmp/.uci and never committed them."""
stale = UCI_FIREWALL_HEALTHY + (
"firewall.snmp_netork=rule\n"
"firewall.snmp_netork.name='Allow-SNMP-from-mgmt'\n"
"firewall.snmp_netork.dest_port='161'\n"
)
shell = _FakeShell(
firewall=stale,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
driver._action_fix_snmp()
delete_idx = next(i for i, c in enumerate(shell.issued) if "snmp_netork" in c and "delete" in c)
assert any("uci commit firewall" in c for c in shell.issued[delete_idx:])
def test_legacy_snmp_netork_rule_is_recognised_as_stale(self, driver):
"""The pre-0.x rule was named 'Allow-SNMP-from-mgmt' — hyphens, not underscores."""
stale = UCI_FIREWALL_HEALTHY + (
"firewall.snmp_netork=rule\n"
"firewall.snmp_netork.name='Allow-SNMP-from-mgmt'\n"
"firewall.snmp_netork.src='*'\n"
"firewall.snmp_netork.dest_port='161'\n"
)
shell = _FakeShell(
firewall=stale,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
driver._action_fix_snmp()
assert any("snmp_netork" in c and "delete" in c for c in shell.issued)
class TestFixSnmpReloadVerification:
"""A failing fw4 reload must fail the action, not be swallowed."""
FW4_ZONE_ERROR = (
"Section @zone[0] option 'name' is mandatory but not set\n"
"Section @zone[0] skipped due to invalid options\n"
"Section @rule[0] references unknown zone 'lan'\n"
"Section @rule[0] skipped due to invalid options"
)
def test_reload_error_fails_the_action(self, driver):
shell = _FakeShell(
firewall=UCI_FIREWALL_HEALTHY,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
reload_out=self.FW4_ZONE_ERROR,
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert result["success"] is False
def test_reload_output_is_not_truncated(self, driver):
"""The old 120-char cap hid the 'references unknown zone' line."""
shell = _FakeShell(
firewall=UCI_FIREWALL_HEALTHY,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
reload_out=self.FW4_ZONE_ERROR,
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert "references unknown zone" in result["output"]
def test_missing_live_rule_fails_the_action(self, driver):
"""snmpd up + clean reload, but no udp/161 accept in the packet filter."""
shell = _FakeShell(
firewall=UCI_FIREWALL_HEALTHY,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
nft_hits="0",
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert result["success"] is False
def test_snmpd_not_listening_fails_the_action(self, driver):
shell = _FakeShell(
firewall=UCI_FIREWALL_HEALTHY,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
listen_hits="0",
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert result["success"] is False
def test_fully_healthy_run_succeeds(self, driver):
shell = _FakeShell(
firewall=UCI_FIREWALL_HEALTHY,
network=UCI_NETWORK_STATIC,
ssh_connection="192.168.1.50 5000 192.168.1.1 22",
)
driver._send_command = shell
result = driver._action_fix_snmp()
assert result["success"] is True
class TestUciSectionParser:
"""_parse_uci_sections() underpins all of the above."""
def test_section_type_captured(self, driver):
parsed = driver._parse_uci_sections(UCI_FIREWALL_HEALTHY)
assert parsed["@zone[0]"]["_type"] == "zone"
def test_option_value_unquoted_on_read(self, driver):
parsed = driver._parse_uci_sections(UCI_FIREWALL_HEALTHY)
assert driver._uci_scalar(parsed["@zone[0]"]["name"]) == "lan"
def test_list_values_split_into_tokens(self, driver):
parsed = driver._parse_uci_sections(UCI_FIREWALL_HEALTHY)
assert driver._uci_tokens(parsed["@zone[1]"]["network"]) == ["wan", "wan6"]
def test_value_containing_equals_is_kept_whole(self, driver):
parsed = driver._parse_uci_sections("firewall.x=rule\nfirewall.x.name='a=b'\n")
assert driver._uci_scalar(parsed["x"]["name"]) == "a=b"
def test_blank_and_malformed_lines_ignored(self, driver):
parsed = driver._parse_uci_sections("\n\nnot a uci line\nfirewall.x=rule\n")
assert list(parsed) == ["x"]
# ---------------------------------------------------------------------------
# get_system_config — remote syslog, LuCI and bridge STP (netOrk #164)
# ---------------------------------------------------------------------------
UCI_SYSTEM_WITH_SYSLOG = """\
system.@system[0]=system
system.@system[0].hostname='ap-eze-Garten'
system.@system[0].timezone='CET-1CEST,M3.5.0,M10.5.0/3'
system.@system[0].zonename='Europe/Berlin'
system.@system[0].log_remote='1'
system.@system[0].log_ip='10.10.40.2'
system.@system[0].log_port='5514'
system.@system[0].log_proto='udp'
system.ntp=timeserver
system.ntp.server='0.openwrt.pool.ntp.org' '1.openwrt.pool.ntp.org'
"""
UCI_SYSTEM_NO_SYSLOG = """\
system.@system[0]=system
system.@system[0].hostname='ap-eze-Parkplatz'
system.@system[0].timezone='GMT0'
system.@system[0].zonename='UTC'
system.ntp=timeserver
system.ntp.server='0.openwrt.pool.ntp.org'
"""
UCI_DROPBEAR_SYS = """\
dropbear.@dropbear[0]=dropbear
dropbear.@dropbear[0].Port='22'
dropbear.@dropbear[0].PasswordAuth='on'
dropbear.@dropbear[0].RootPasswordAuth='on'
"""
UCI_NETWORK_STP_ON = """\
network.ap_bridge=device
network.ap_bridge.name='br-ap'
network.ap_bridge.type='bridge'
network.ap_bridge.stp='1'
network.lan=interface
"""
UCI_NETWORK_STP_OFF = """\
network.ap_bridge=device
network.ap_bridge.name='br-ap'
network.ap_bridge.type='bridge'
network.lan=interface
"""
def _system_send(system_out=UCI_SYSTEM_WITH_SYSLOG, luci_out="1", network_out=UCI_NETWORK_STP_ON):
"""Dispatch _send_command by the command it receives."""
def _send(cmd, **kw):
if "uci show system" in cmd:
return system_out
if "uci show dropbear" in cmd:
return UCI_DROPBEAR_SYS
if "uci show network" in cmd:
return network_out
if "uhttpd" in cmd:
return luci_out
return ""
return _send
class TestGetSystemConfigSyslog:
def test_remote_syslog_target_is_reported(self, driver):
driver._send_command = _system_send()
cfg = driver.get_system_config()
assert cfg["syslog_remote"] is True
assert cfg["syslog_ip"] == "10.10.40.2"
assert cfg["syslog_port"] == 5514
def test_absent_syslog_reports_as_disabled(self, driver):
driver._send_command = _system_send(UCI_SYSTEM_NO_SYSLOG)
cfg = driver.get_system_config()
assert cfg["syslog_remote"] is False
assert cfg["syslog_ip"] == ""
def test_log_ip_without_log_remote_is_not_active(self, driver):
# OpenWrt only ships logs when log_remote is set, whatever log_ip says.
out = UCI_SYSTEM_NO_SYSLOG + "system.@system[0].log_ip='10.10.40.2'\n"
driver._send_command = _system_send(out)
assert driver.get_system_config()["syslog_remote"] is False
def test_port_falls_back_to_the_openwrt_default(self, driver):
out = UCI_SYSTEM_NO_SYSLOG + (
"system.@system[0].log_remote='1'\nsystem.@system[0].log_ip='10.10.40.2'\n"
)
driver._send_command = _system_send(out)
assert driver.get_system_config()["syslog_port"] == 514
def test_existing_fields_are_untouched(self, driver):
driver._send_command = _system_send()
cfg = driver.get_system_config()
assert cfg["timezone"] == "CET-1CEST,M3.5.0,M10.5.0/3"
assert cfg["zonename"] == "Europe/Berlin"
assert cfg["dropbear_port"] == 22
assert cfg["ntp_servers"] == ["0.openwrt.pool.ntp.org", "1.openwrt.pool.ntp.org"]
class TestGetSystemConfigLuci:
def test_enabled_uhttpd_reports_luci_as_reachable(self, driver):
driver._send_command = _system_send(luci_out="1")
assert driver.get_system_config()["luci_enabled"] is True
def test_disabled_uhttpd_reports_luci_as_unreachable(self, driver):
driver._send_command = _system_send(luci_out="0")
assert driver.get_system_config()["luci_enabled"] is False
def test_missing_uhttpd_reports_as_unreachable(self, driver):
# No uhttpd installed at all — LuCI cannot be served.
driver._send_command = _system_send(luci_out="")
assert driver.get_system_config()["luci_enabled"] is False
class TestGetSystemConfigBridgeStp:
def test_stp_enabled_is_reported(self, driver):
driver._send_command = _system_send(network_out=UCI_NETWORK_STP_ON)
assert driver.get_system_config()["bridge_stp"] is True
def test_absent_stp_option_means_disabled(self, driver):
# UCI defaults stp to 0 when the option is not present.
driver._send_command = _system_send(network_out=UCI_NETWORK_STP_OFF)
assert driver.get_system_config()["bridge_stp"] is False
def test_no_ap_bridge_reports_none(self, driver):
# Nothing to have an opinion about — distinct from "STP is off".
driver._send_command = _system_send(network_out="network.lan=interface\n")
assert driver.get_system_config()["bridge_stp"] is None
Reference in New Issue View Git Blame Copy Permalink