Christian Manivong
6b78ebcacb
OpenWrt's wifi-scripts validator rejects any macfilter value other than "allow"/"deny" outright — confirmed on real hardware, setting macfilter='disable' puts netifd in a permanent restart crash loop with the radio stuck down. The only way to disable filtering is to delete the option entirely. Also guards against pushing an empty whitelist (macfilter='allow' with zero MACs blocks every client outright) by treating it as equivalent to "off".
1257 lines
45 KiB
Python
1257 lines
45 KiB
Python
"""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",
|
|
}
|
|
|
|
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"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)
|