From 512871b1bb92c52e717ce945b0a8a5ea7f1de7c7 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 24 Jun 2026 23:26:25 +0200 Subject: [PATCH] feat: get_channel_scan() + push_radio_channel() in OpenWrtWirelessMixin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_channel_scan(mode='active'|'passive'): - Discovers AP-mode interfaces via 'iw dev' - Runs 'iw dev scan [passive]' on each AP interface - Parses BSS blocks: BSSID, SSID, frequency, channel, signal, channel width (HT/VHT/HE), band (2.4/5/6 GHz) - Returns dict[iface → list[ChannelScanEntryDict]] push_radio_channel(radio, channel): - Sets channel via 'uci set wireless..channel=' - Commits and reloads: 'uci commit wireless && wifi reload' - channel=0 writes 'auto' Tests: 19 new unit tests covering parse logic, active/passive flag, AP-only interface filtering, 6 GHz band detection, push command order. Co-Authored-By: Claude Sonnet 4.6 --- napalm_openwrt/wireless_mixin.py | 158 ++++++++++++++++++ tests/unit/test_driver.py | 264 +++++++++++++++++++++++++++++++ 2 files changed, 422 insertions(+) diff --git a/napalm_openwrt/wireless_mixin.py b/napalm_openwrt/wireless_mixin.py index e0dec18..2c508b2 100644 --- a/napalm_openwrt/wireless_mixin.py +++ b/napalm_openwrt/wireless_mixin.py @@ -380,6 +380,164 @@ class OpenWrtWirelessMixin: _flush() return results + def get_channel_scan(self, mode: str = "active") -> dict[str, Any]: + """Return channel scan results for all AP-mode interfaces. + + Uses ``iw dev scan`` (active) or + ``iw dev scan passive`` (passive) on each AP interface. + + Keys are interface names (e.g. ``"wlan0"``). Values are lists of + :class:`~napalm_device_types.models.ChannelScanEntryDict`. + """ + from napalm_device_types.models import ChannelScanEntryDict + + # Step 1: discover interfaces and filter to AP mode only + iw_out = self._send_command("iw dev 2>/dev/null || true") + ap_ifaces: list[str] = [] + current_iface: str = "" + current_type: str = "" + iface_order: list[tuple[str, str]] = [] # (iface, type) + + for line in iw_out.splitlines(): + stripped = line.strip() + im = re.match(r"^Interface\s+(\S+)$", stripped) + if im: + if current_iface: + iface_order.append((current_iface, current_type)) + current_iface = im.group(1) + current_type = "" + continue + tm = re.match(r"^type\s+(\S+)$", stripped) + if tm and current_iface: + current_type = tm.group(1) + + if current_iface: + iface_order.append((current_iface, current_type)) + + ap_ifaces = [ + name for name, itype in iface_order + if itype.upper() in ("AP", "AP/VLAN") + ] + + if not ap_ifaces: + return {} + + # Step 2: run scan on each AP interface + passive_flag = " passive" if mode == "passive" else "" + result: dict[str, list[Any]] = {} + + for iface in ap_ifaces: + scan_cmd = f"iw dev {iface} scan{passive_flag} 2>/dev/null || true" + scan_out = self._send_command(scan_cmd) + entries = self._parse_iw_scan(scan_out, iface) + result[iface] = entries + + return result + + @staticmethod + def _parse_iw_scan(scan_out: str, iface: str) -> list[Any]: + """Parse ``iw dev scan`` output into ChannelScanEntryDict list.""" + from napalm_device_types.models import ChannelScanEntryDict + + entries: list[Any] = [] + current: dict[str, Any] = {} + + def _freq_to_band(freq: int) -> str: + if freq < 3000: + return "2.4GHz" + if freq < 5950: + return "5GHz" + return "6GHz" + + def _freq_to_channel(freq: int) -> int: + if 2412 <= freq <= 2484: + if freq == 2484: + return 14 + return (freq - 2407) // 5 + if 5160 <= freq <= 5885: + return (freq - 5000) // 5 + if 5955 <= freq <= 7115: + return (freq - 5950) // 5 + 1 + return 0 + + def _flush() -> None: + if current.get("bssid"): + freq = current.get("frequency", 0) + ch = current.get("channel") or _freq_to_channel(freq) + width = current.get("channel_width", 0) + # If width is still 0 try to derive from HT/VHT/HE info + if width == 0 and current.get("_ht_width"): + width = current["_ht_width"] + entries.append(ChannelScanEntryDict( + bssid=current["bssid"], + ssid=current.get("ssid", ""), + frequency=freq, + channel=ch, + signal_dbm=current.get("signal_dbm", 0), + channel_width=width, + band=_freq_to_band(freq), + )) + + for line in scan_out.splitlines(): + stripped = line.strip() + + # New BSS block: "BSS aa:bb:cc:dd:ee:ff(on wlan0)" + bss_m = re.match(r"^BSS\s+([\da-fA-F:]{17})\(", stripped) + if bss_m: + _flush() + current = {"bssid": bss_m.group(1).lower()} + continue + + # freq: 2437 + freq_m = re.match(r"^freq:\s+(\d+)$", stripped) + if freq_m: + current["frequency"] = int(freq_m.group(1)) + continue + + # signal: -72.00 dBm + sig_m = re.match(r"^signal:\s+([-\d.]+)\s+dBm", stripped) + if sig_m: + current["signal_dbm"] = int(float(sig_m.group(1))) + continue + + # SSID: NeighborNet (may be empty for hidden) + ssid_m = re.match(r"^SSID:\s*(.*)?$", stripped) + if ssid_m: + current["ssid"] = ssid_m.group(1).strip() + continue + + # HT operation primary channel: 6 + ht_ch_m = re.match(r"^\*\s+primary channel:\s+(\d+)", stripped) + if ht_ch_m: + current["channel"] = int(ht_ch_m.group(1)) + continue + + # HT STA channel width: 20 MHz + ht_w_m = re.match(r"^\*\s+STA channel width:\s+(\d+)\s+MHz", stripped) + if ht_w_m: + current["_ht_width"] = int(ht_w_m.group(1)) + continue + + # VHT channel width: 80 MHz + vht_w_m = re.match(r"^\*\s+channel width:\s+(\d+)\s+MHz", stripped) + if vht_w_m: + current["channel_width"] = int(vht_w_m.group(1)) + continue + + _flush() + return entries + + def push_radio_channel(self, radio: str, channel: int) -> None: + """Set a new channel on *radio* via UCI and reload the wireless stack. + + :param radio: UCI radio name, e.g. ``"radio0"``. + :param channel: Channel number to set; ``0`` sets UCI ``auto``. + """ + ch_val = "auto" if channel == 0 else str(channel) + self._send_command(f"uci set wireless.{radio}.channel={ch_val}") + self._send_command("uci commit wireless") + self._send_command("wifi reload") + def get_radio_status(self) -> dict[str, Any]: """Return radio status from UCI and iwinfo. diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 467910f..fe3869c 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -833,3 +833,267 @@ class TestGetNetworkInstances: ) 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_reload(self, driver): + issued: list[str] = [] + driver._send_command = lambda cmd, **kw: issued.append(cmd) or "" + driver.push_radio_channel("radio1", 36) + assert any("wifi reload" in c 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_reload(self, driver): + """UCI commit must happen before wifi reload.""" + 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) + reload_idx = next(i for i, c in enumerate(issued) if "wifi reload" in c) + assert commit_idx < reload_idx