diff --git a/README.md b/README.md index 6d41139..17ee077 100644 --- a/README.md +++ b/README.md @@ -74,10 +74,12 @@ Configuration is managed via [UCI](https://openwrt.org/docs/guide-user/base-syst ### Merge candidate ```python -device.load_merge_candidate(config=""" +device.load_merge_candidate( + config=""" uci set system.@system[0].hostname='my-router' uci set network.lan.ipaddr='10.0.0.1' -""") +""" +) print(device.compare_config()) device.commit_config() diff --git a/napalm_openwrt/config_mixin.py b/napalm_openwrt/config_mixin.py index e671ba5..9559a96 100644 --- a/napalm_openwrt/config_mixin.py +++ b/napalm_openwrt/config_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -50,7 +49,9 @@ class OpenWrtConfigMixin(ConfigLifecycleMixin): if self._candidate_config is None: raise MergeConfigException("No candidate configuration is staged.") - ex_cls = ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException + ex_cls = ( + ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException + ) self._backup_config = self._send_command("uci export") diff --git a/napalm_openwrt/interfaces_mixin.py b/napalm_openwrt/interfaces_mixin.py index 81b4966..43a4824 100644 --- a/napalm_openwrt/interfaces_mixin.py +++ b/napalm_openwrt/interfaces_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -33,9 +32,8 @@ class OpenWrtInterfaceMixin: m = re.match(r"^\d+:\s+(\S+?)[@:]", line) if m: name = m.group(1) - if ( - name not in self._EXCLUDED_INTERFACES - and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES) + if name not in self._EXCLUDED_INTERFACES and not name.startswith( + self._EXCLUDED_INTERFACE_PREFIXES ): interfaces.append(name) return sorted(set(interfaces)) @@ -368,12 +366,14 @@ class OpenWrtInterfaceMixin: except Exception: mac_addr = mac_raw - table.append({ - "interface": interface, - "mac": mac_addr, - "ip": ip_addr, - "age": -1.0, - "state": state, - }) + table.append( + { + "interface": interface, + "mac": mac_addr, + "ip": ip_addr, + "age": -1.0, + "state": state, + } + ) return table diff --git a/napalm_openwrt/lldp_mixin.py b/napalm_openwrt/lldp_mixin.py index 3da4351..168677e 100644 --- a/napalm_openwrt/lldp_mixin.py +++ b/napalm_openwrt/lldp_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at diff --git a/napalm_openwrt/openwrt.py b/napalm_openwrt/openwrt.py index 08434cd..f48b3ee 100644 --- a/napalm_openwrt/openwrt.py +++ b/napalm_openwrt/openwrt.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -21,12 +20,13 @@ from __future__ import annotations import ipaddress import re -import socket from typing import Any -from netmiko import ConnectHandler -from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException - +from napalm.base.exceptions import ( + ConnectionClosedException, + ConnectionException, +) +from napalm.base.netmiko_helpers import netmiko_args from napalm_device_types import ( AccessPointDriver, ByteStream, @@ -37,22 +37,18 @@ from napalm_device_types import ( ) from napalm_device_types.channel import open_stream_on_transport, run_on_transport from napalm_device_types.listening import ListeningSocketsMixin -from napalm.base.exceptions import ( - ConnectionException, - ConnectionClosedException, -) -from napalm.base.netmiko_helpers import netmiko_args from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch +from netmiko import ConnectHandler +from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException -from napalm_openwrt.interfaces_mixin import OpenWrtInterfaceMixin -from napalm_openwrt.wireless_mixin import OpenWrtWirelessMixin -from napalm_openwrt.lldp_mixin import OpenWrtLLDPMixin -from napalm_openwrt.vlan_mixin import OpenWrtVLANMixin from napalm_openwrt.config_mixin import OpenWrtConfigMixin -from napalm_openwrt.system_mixin import _RC_MARKER, OpenWrtSystemMixin, _split_status +from napalm_openwrt.interfaces_mixin import OpenWrtInterfaceMixin +from napalm_openwrt.lldp_mixin import OpenWrtLLDPMixin from napalm_openwrt.packages_mixin import OpenWrtPackageMixin from napalm_openwrt.routing_mixin import OpenWrtRoutingMixin - +from napalm_openwrt.system_mixin import _RC_MARKER, OpenWrtSystemMixin, _split_status +from napalm_openwrt.vlan_mixin import OpenWrtVLANMixin +from napalm_openwrt.wireless_mixin import OpenWrtWirelessMixin #: What may stand in a UCI value written in single quotes: no whitespace, quote or #. _SNMP_WORD = re.compile(r"^[^\s'\"#]+$") @@ -146,13 +142,9 @@ class OpenWrtDriver( except SshHostKeyMismatch: raise # refused before the login; nothing was sent except NetmikoTimeoutException as exc: - raise ConnectionException( - f"Cannot connect to {self.hostname}: {exc}" - ) from exc + raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc except NetmikoAuthenticationException as exc: - raise ConnectionException( - f"Authentication failed for {self.hostname}: {exc}" - ) from exc + raise ConnectionException(f"Authentication failed for {self.hostname}: {exc}") from exc self._host_key.observe(self.device.remote_conn_pre.get_transport()) def close(self) -> None: @@ -167,7 +159,7 @@ class OpenWrtDriver( return {"is_alive": False} try: return {"is_alive": self.device.remote_conn.transport.is_active()} - except (socket.error, EOFError, AttributeError): + except (OSError, EOFError, AttributeError): return {"is_alive": False} # ------------------------------------------------------------------ @@ -180,6 +172,7 @@ class OpenWrtDriver( When a list is supplied, commands are tried in order and the first one that does not return an error indicator is returned. """ + def _do_send(cmd: str) -> str: return self.device.send_command( cmd, @@ -195,7 +188,7 @@ class OpenWrtDriver( return output return output return _do_send(command) - except (socket.error, EOFError) as exc: + except (OSError, EOFError) as exc: raise ConnectionClosedException(str(exc)) from exc # ------------------------------------------------------------------ @@ -317,18 +310,14 @@ class OpenWrtDriver( if model_raw and not model_raw.startswith("cat: "): facts["model"] = model_raw.strip().replace("\x00", "") else: - facts["model"] = ( - release.get("DISTRIB_TARGET", "").replace("/", " ") - ) + facts["model"] = release.get("DISTRIB_TARGET", "").replace("/", " ") facts["os_version"] = release.get( "DISTRIB_DESCRIPTION", release.get("DISTRIB_RELEASE", "unknown"), ) facts["serial_number"] = "" - facts["uptime"] = self._parse_uptime_seconds( - self._send_command("cat /proc/uptime") - ) + facts["uptime"] = self._parse_uptime_seconds(self._send_command("cat /proc/uptime")) # hostname and FQDN uname_out = self._send_command("uname -a") @@ -370,27 +359,15 @@ class OpenWrtDriver( return configs - def cli( - self, commands: list[str], encoding: str = "text" - ) -> dict[str, str]: + def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]: """Execute a list of CLI commands and return their output.""" - return { - cmd: self._send_command(cmd) - for cmd in commands - } + return {cmd: self._send_command(cmd) for cmd in commands} # ------------------------------------------------------------------ # Service management (forwarded to mixins because AccessPointDriver # defines NotImplementedError stubs for these) # ------------------------------------------------------------------ - - - - - - - # ------------------------------------------------------------------ # UCI parsing helpers # ------------------------------------------------------------------ @@ -523,9 +500,7 @@ class OpenWrtDriver( return "" @classmethod - def _stale_snmp_rules( - cls, fw_sections: dict[str, dict[str, str]], keep: str - ) -> list[str]: + def _stale_snmp_rules(cls, fw_sections: dict[str, dict[str, str]], keep: str) -> list[str]: """Return every SNMP rule section except ``keep``, in delete-safe order.""" hits = [ key diff --git a/napalm_openwrt/packages_mixin.py b/napalm_openwrt/packages_mixin.py index 2281049..65cdbfb 100644 --- a/napalm_openwrt/packages_mixin.py +++ b/napalm_openwrt/packages_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -99,14 +98,16 @@ class OpenWrtPackageMixin: name = token[: vm.start()] if vm else token version = vm.group(1) if vm else "" origin = "" - packages.append({ - "name": name, - "version": version, - "installed": True, - "description": "", - "size": 0, - "source": origin, - }) + packages.append( + { + "name": name, + "version": version, + "installed": True, + "description": "", + "size": 0, + "source": origin, + } + ) return sorted(packages, key=lambda p: p["name"].lower()) def _pm_type(self) -> str: @@ -121,6 +122,7 @@ class OpenWrtPackageMixin: index is populated (OpenWrt stores it in RAM and loses it on reboot). """ import shlex + safe_q = shlex.quote(query) if self._pm_type() == "apk": # Refresh index (no-ops if already current, safe to run every time) @@ -137,14 +139,16 @@ class OpenWrtPackageMixin: name, version, description = m.group(1), m.group(2), (m.group(3) or "") else: name, version, description = line, "", "" - packages.append({ - "name": name, - "version": version, - "installed": name in installed, - "description": description, - "size": 0, - "source": "", - }) + packages.append( + { + "name": name, + "version": version, + "installed": name in installed, + "description": description, + "size": 0, + "source": "", + } + ) else: # opkg lists live in /var/opkg-lists/ (RAM) — cleared on reboot self._send_command("opkg update 2>/dev/null || true") @@ -159,14 +163,16 @@ class OpenWrtPackageMixin: name = parts[0].strip() version = parts[1].strip() if len(parts) > 1 else "" description = parts[2].strip() if len(parts) > 2 else "" - packages.append({ - "name": name, - "version": version, - "installed": name in installed, - "description": description, - "size": 0, - "source": "", - }) + packages.append( + { + "name": name, + "version": version, + "installed": name in installed, + "description": description, + "size": 0, + "source": "", + } + ) return packages @staticmethod @@ -175,27 +181,29 @@ class OpenWrtPackageMixin: # Strip CSI sequences (\x1b[...X), OSC, charset designations, and # 2-byte DEC private sequences like ESC 7 (cursor save) / ESC 8 (restore) cleaned = re.sub( - r'\x1b(?:\[[0-9;?]*[a-zA-Z]|\][^\x07]*\x07|[()][0-9A-Za-z]|[\x30-\x7e])', - '', raw, + r"\x1b(?:\[[0-9;?]*[a-zA-Z]|\][^\x07]*\x07|[()][0-9A-Za-z]|[\x30-\x7e])", + "", + raw, ) # After stripping cursor-save/restore sequences, apk progress updates # end up concatenated on a single line. Strip those inline patterns. - cleaned = re.sub(r'\s*\d{1,3}%\s*#*', ' ', cleaned) + cleaned = re.sub(r"\s*\d{1,3}%\s*#*", " ", cleaned) lines = [] - for segment in cleaned.split('\n'): + for segment in cleaned.split("\n"): # \r overwrites the line; keep only the portion after the last \r - part = segment.split('\r')[-1].strip() + part = segment.split("\r")[-1].strip() if not part: continue # Drop pure progress-bar lines (only #, spaces, digits, %) - if re.match(r'^[#\s\d%]*$', part): + if re.match(r"^[#\s\d%]*$", part): continue lines.append(part) - return '\n'.join(lines) + return "\n".join(lines) def install_package(self, name: str) -> dict[str, Any]: """Install a package by name. Returns ``{"success": bool, "output": str}``.""" import shlex + safe_name = shlex.quote(name) if self._pm_type() == "apk": raw = self._send_command(f"apk add {safe_name} 2>&1") @@ -209,6 +217,7 @@ class OpenWrtPackageMixin: def uninstall_package(self, name: str) -> dict[str, Any]: """Remove a package by name. Returns ``{"success": bool, "output": str}``.""" import shlex + safe_name = shlex.quote(name) if self._pm_type() == "apk": raw = self._send_command(f"apk del {safe_name} 2>&1") diff --git a/napalm_openwrt/routing_mixin.py b/napalm_openwrt/routing_mixin.py index ea027e7..07d504f 100644 --- a/napalm_openwrt/routing_mixin.py +++ b/napalm_openwrt/routing_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -145,7 +144,7 @@ class OpenWrtRoutingMixin: # "192.168.1.0/24 dev ..." → prefix as-is if line_s.startswith("default"): prefix = "0.0.0.0/0" - rest = line_s[len("default"):].strip() + rest = line_s[len("default") :].strip() else: parts = line_s.split() prefix = parts[0] diff --git a/napalm_openwrt/system_mixin.py b/napalm_openwrt/system_mixin.py index db91e0c..d302d72 100644 --- a/napalm_openwrt/system_mixin.py +++ b/napalm_openwrt/system_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -16,8 +15,6 @@ from __future__ import annotations import re from typing import Any - - #: Printed after a command so its exit status survives the shell session. _RC_MARKER = "__OWRT_RC=" _RC_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) @@ -31,6 +28,7 @@ def _split_status(raw: str) -> tuple[str, int | None]: output = _RC_RE.sub("", raw or "").strip() return output, int(statuses[-1]) if statuses else None + class OpenWrtSystemMixin: """Mixin providing system-level NAPALM getters (environment, NTP, SNMP, users, services, updates, device actions).""" @@ -95,9 +93,7 @@ class OpenWrtSystemMixin: """ sys_out = self._send_command("uci show system 2>/dev/null || true") db_out = self._send_command("uci show dropbear 2>/dev/null || true") - luci_out = self._send_command( - "/etc/init.d/uhttpd enabled 2>/dev/null && echo 1 || echo 0" - ) + luci_out = self._send_command("/etc/init.d/uhttpd enabled 2>/dev/null && echo 1 || echo 0") net_out = self._send_command("uci show network 2>/dev/null || true") sys_cfg: dict[str, str] = {} @@ -238,7 +234,8 @@ class OpenWrtSystemMixin: running = ( self._send_command( "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive" - ).strip() == "active" + ).strip() + == "active" ) if not running: return None @@ -282,7 +279,13 @@ class OpenWrtSystemMixin: if len(parts) < 7: continue username, password_hash, uid_str, _, _, home, shell = ( - parts[0], parts[1], parts[2], parts[3], parts[4], parts[5], parts[6], + parts[0], + parts[1], + parts[2], + parts[3], + parts[4], + parts[5], + parts[6], ) if shell not in valid_shells: continue @@ -325,9 +328,7 @@ class OpenWrtSystemMixin: import json as _json # -- enabled set: names from /etc/rc.d/S symlinks ---- - rc_out = self._send_command( - "ls /etc/rc.d/ 2>/dev/null | grep '^S' | sed 's/^S[0-9]*//'" - ) + rc_out = self._send_command("ls /etc/rc.d/ 2>/dev/null | grep '^S' | sed 's/^S[0-9]*//'") enabled: set[str] = {s.strip() for s in rc_out.splitlines() if s.strip()} # -- running info from procd via ubus --------------------------------- @@ -345,9 +346,7 @@ class OpenWrtSystemMixin: continue instances = svc_info.get("instances", {}) running = any( - inst.get("running", False) - for inst in instances.values() - if isinstance(inst, dict) + inst.get("running", False) for inst in instances.values() if isinstance(inst, dict) ) pid = next( ( @@ -371,21 +370,24 @@ class OpenWrtSystemMixin: result: list[dict[str, Any]] = [] for name in sorted(all_names): info = service_map.get(name, {}) - result.append({ - "name": name, - "running": info.get("running", False), - "enabled": name in enabled, - "pid": info.get("pid", 0), - }) + result.append( + { + "name": name, + "running": info.get("running", False), + "enabled": name in enabled, + "pid": info.get("pid", 0), + } + ) return result def manage_service(self, name: str, action: str) -> dict[str, Any]: """Execute a lifecycle action (start/stop/restart/enable/disable) on a service.""" import re as _re - if not _re.match(r'^[a-zA-Z0-9_\-]+$', name): + + if not _re.match(r"^[a-zA-Z0-9_\-]+$", name): raise ValueError(f"Invalid service name: {name!r}") - if action not in ('start', 'stop', 'restart', 'enable', 'disable'): + if action not in ("start", "stop", "restart", "enable", "disable"): raise ValueError(f"Invalid action: {action!r}") output = self._send_command(f"/etc/init.d/{name} {action} 2>&1") return {"success": True, "output": output} @@ -417,11 +419,13 @@ class OpenWrtSystemMixin: # Output format: "pkgname-current_ver < new_ver" m = _re.match(r"^(.+)-(\d\S*)\s+<\s+(\S+)", line.strip()) if m: - updates.append({ - "name": m.group(1), - "current_version": m.group(2), - "new_version": m.group(3), - }) + updates.append( + { + "name": m.group(1), + "current_version": m.group(2), + "new_version": m.group(3), + } + ) return updates def _updates_opkg(self) -> list[dict[str, Any]]: @@ -436,11 +440,13 @@ class OpenWrtSystemMixin: # opkg output: "pkgname - current_ver - new_ver" parts = [p.strip() for p in line.split(" - ")] if len(parts) == 3: - updates.append({ - "name": parts[0], - "current_version": parts[1], - "new_version": parts[2], - }) + updates.append( + { + "name": parts[0], + "current_version": parts[1], + "new_version": parts[2], + } + ) return updates def refresh_available_updates(self) -> dict[str, Any]: @@ -452,8 +458,9 @@ class OpenWrtSystemMixin: def apply_updates(self, packages: list[str]) -> dict[str, Any]: """Upgrade the given packages using the device's package manager.""" import re as _re + for pkg in packages: - if not _re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg): + if not _re.match(r"^[a-zA-Z0-9_\-\+\.]+$", pkg): raise ValueError(f"Invalid package name: {pkg!r}") pm = self._pm_type() pkg_args = " ".join(packages) @@ -535,19 +542,21 @@ class OpenWrtSystemMixin: if len(parts) < 10: continue try: - stats.append({ - "remote": parts[0], - "referenceid": parts[1], - "synchronized": tally == "*", - "stratum": int(parts[2]), - "type": parts[3], - "when": parts[4], - "hostpoll": int(parts[5]), - "reachability": int(parts[6], 8), # octal - "delay": float(parts[7]), - "offset": float(parts[8]), - "jitter": float(parts[9]), - }) + stats.append( + { + "remote": parts[0], + "referenceid": parts[1], + "synchronized": tally == "*", + "stratum": int(parts[2]), + "type": parts[3], + "when": parts[4], + "hostpoll": int(parts[5]), + "reachability": int(parts[6], 8), # octal + "delay": float(parts[7]), + "offset": float(parts[8]), + "jitter": float(parts[9]), + } + ) except (ValueError, IndexError): continue return stats @@ -564,19 +573,23 @@ class OpenWrtSystemMixin: continue tally = m.group(1) try: - stats.append({ - "remote": m.group(2), - "referenceid": "", - "synchronized": tally == "^*", - "stratum": int(m.group(3)), - "type": "u", - "when": m.group(6), - "hostpoll": int(m.group(4)), - "reachability": int(m.group(5), 8) if re.match(r"^[0-7]+$", m.group(5)) else 0, - "delay": 0.0, - "offset": 0.0, - "jitter": 0.0, - }) + stats.append( + { + "remote": m.group(2), + "referenceid": "", + "synchronized": tally == "^*", + "stratum": int(m.group(3)), + "type": "u", + "when": m.group(6), + "hostpoll": int(m.group(4)), + "reachability": int(m.group(5), 8) + if re.match(r"^[0-7]+$", m.group(5)) + else 0, + "delay": 0.0, + "offset": 0.0, + "jitter": 0.0, + } + ) except (ValueError, IndexError): continue return stats diff --git a/napalm_openwrt/vlan_mixin.py b/napalm_openwrt/vlan_mixin.py index 9b29e25..575f419 100644 --- a/napalm_openwrt/vlan_mixin.py +++ b/napalm_openwrt/vlan_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -172,7 +171,7 @@ class OpenWrtVLANMixin: vm = re.match(r"^([\w-]+)\.(\d+)$", iface) # allow hyphens (br-ap) if not vm: continue - parent = vm.group(1) # e.g. "eth0" or "br-ap" + parent = vm.group(1) # e.g. "eth0" or "br-ap" vlan_id = str(int(vm.group(2))) # e.g. "10" vlans.setdefault(vlan_id, {"name": "", "tagged": [], "untagged": []}) # sub-interface itself -> untagged egress @@ -210,11 +209,13 @@ class OpenWrtVLANMixin: # No explicit bridge-vlan section -- nothing to delete via UCI return - self.cli([ - f"uci delete network.@bridge-vlan[{idx}]", - "uci commit network", - "/etc/init.d/network reload", - ]) + self.cli( + [ + f"uci delete network.@bridge-vlan[{idx}]", + "uci commit network", + "/etc/init.d/network reload", + ] + ) def get_network_instances(self, name: str = "") -> dict[str, Any]: """Return network instances (Linux network namespaces + default). diff --git a/napalm_openwrt/wireless_mixin.py b/napalm_openwrt/wireless_mixin.py index a6c240f..5e5812b 100644 --- a/napalm_openwrt/wireless_mixin.py +++ b/napalm_openwrt/wireless_mixin.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at @@ -101,8 +100,12 @@ class OpenWrtWirelessMixin: return "" _ENC_MAP = { - "": "Open", "none": "Open", "0": "Open", - "wep": "WEP", "wep-open": "WEP (Open)", "wep-shared": "WEP (Shared)", + "": "Open", + "none": "Open", + "0": "Open", + "wep": "WEP", + "wep-open": "WEP (Open)", + "wep-shared": "WEP (Shared)", "psk": "WPA-PSK", "psk+ccmp": "WPA-PSK", "psk-mixed": "WPA/WPA2-PSK", @@ -176,11 +179,17 @@ class OpenWrtWirelessMixin: ft_over_ds = entry.get("ft_over_ds", "1") == "1" client_isolation = entry.get("isolate", "0") == "1" _max_raw = entry.get("maxassoc") - max_clients: int | None = int(_max_raw) if _max_raw and str(_max_raw).isdigit() else None + max_clients: int | None = ( + int(_max_raw) if _max_raw and str(_max_raw).isdigit() else None + ) _disassoc_raw = entry.get("disassoc_low_ack") - disassoc_low_ack: bool | None = (_disassoc_raw == "1") if _disassoc_raw is not None else None + disassoc_low_ack: bool | None = ( + (_disassoc_raw == "1") if _disassoc_raw is not None else None + ) _max_inact_raw = entry.get("max_inactivity") - max_inactivity: int | None = int(_max_inact_raw) if _max_inact_raw and str(_max_inact_raw).isdigit() else None + max_inactivity: int | None = ( + int(_max_inact_raw) if _max_inact_raw and str(_max_inact_raw).isdigit() else None + ) key: str = entry.get("key", "") or "" acl_mode = _ACL_MODE_MAP.get(entry.get("macfilter", ""), "off") mac_list = sorted(set(iface_maclists.get(idx, []))) @@ -331,16 +340,18 @@ class OpenWrtWirelessMixin: def _flush() -> None: if current_station and current_station.get("mac"): info = ap_ifaces.get(active_iface, {}) - results.append(WirelessClientDict( - mac=current_station["mac"], - ssid=info.get("ssid", ""), - radio=info.get("band") or info.get("radio", ""), - signal=current_station.get("signal", 0), - noise=0, - tx_rate=current_station.get("tx_rate", 0.0), - rx_rate=current_station.get("rx_rate", 0.0), - uptime=current_station.get("uptime", 0), - )) + results.append( + WirelessClientDict( + mac=current_station["mac"], + ssid=info.get("ssid", ""), + radio=info.get("band") or info.get("radio", ""), + signal=current_station.get("signal", 0), + noise=0, + tx_rate=current_station.get("tx_rate", 0.0), + rx_rate=current_station.get("rx_rate", 0.0), + uptime=current_station.get("uptime", 0), + ) + ) for line in station_out.splitlines(): stripped = line.strip() @@ -410,7 +421,6 @@ class OpenWrtWirelessMixin: 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") @@ -435,10 +445,7 @@ class OpenWrtWirelessMixin: 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") - ] + ap_ifaces = [name for name, itype in iface_order if itype.upper() in ("AP", "AP/VLAN")] if not ap_ifaces: return {} @@ -489,15 +496,17 @@ class OpenWrtWirelessMixin: # 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), - )) + 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() @@ -663,10 +672,22 @@ class OpenWrtWirelessMixin: # Derive channel_width from htmode string (e.g. VHT80 → 80 MHz) _HTMODE_WIDTH = { - "HT20": 20, "HT40": 40, - "VHT20": 20, "VHT40": 40, "VHT80": 80, "VHT80+80": 80, "VHT160": 160, - "HE20": 20, "HE40": 40, "HE80": 80, "HE160": 160, - "EHT20": 20, "EHT40": 40, "EHT80": 80, "EHT160": 160, "EHT320": 320, + "HT20": 20, + "HT40": 40, + "VHT20": 20, + "VHT40": 40, + "VHT80": 80, + "VHT80+80": 80, + "VHT160": 160, + "HE20": 20, + "HE40": 40, + "HE80": 80, + "HE160": 160, + "EHT20": 20, + "EHT40": 40, + "EHT80": 80, + "EHT160": 160, + "EHT320": 320, } channel_width = _HTMODE_WIDTH.get(htmode.upper(), 0) @@ -677,7 +698,7 @@ class OpenWrtWirelessMixin: channel=channel, channel_width=channel_width, tx_power=tx_power, - frequency=0.0, # enriched below via iwinfo + frequency=0.0, # enriched below via iwinfo ), "htmode": htmode, "country": country, diff --git a/pyproject.toml b/pyproject.toml index ae770cb..4a7f49f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,4 +52,11 @@ include = ["napalm_openwrt*"] [tool.ruff] line-length = 100 -target-version = "py38" +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] + +[tool.ruff.lint.per-file-ignores] +# Recorded device output keeps its lines. +"tests/**" = ["E501"] diff --git a/tests/unit/test_command_channel.py b/tests/unit/test_command_channel.py index ccd5116..f1f30f2 100644 --- a/tests/unit/test_command_channel.py +++ b/tests/unit/test_command_channel.py @@ -86,7 +86,9 @@ class TestRebootHost: def test_detached_so_the_answer_comes_back_first(self): drv = _driver() - with patch("napalm_openwrt.openwrt.run_on_transport", return_value=CommandResult("", "", 0)) as run: + with patch( + "napalm_openwrt.openwrt.run_on_transport", return_value=CommandResult("", "", 0) + ) as run: drv.reboot_host() command = run.call_args.args[1] @@ -96,8 +98,9 @@ class TestRebootHost: def test_a_refusal_is_raised(self): drv = _driver() failed = CommandResult("", "reboot: Operation not permitted", 1) - with patch("napalm_openwrt.openwrt.run_on_transport", return_value=failed), pytest.raises( - RuntimeError, match="not permitted" + with ( + patch("napalm_openwrt.openwrt.run_on_transport", return_value=failed), + pytest.raises(RuntimeError, match="not permitted"), ): drv.reboot_host() diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index fd6a793..77ffb82 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1,15 +1,16 @@ """Unit tests for OpenWrtDriver — no real device required.""" -import pytest from unittest.mock import MagicMock, patch -from napalm_openwrt.openwrt import OpenWrtDriver +import pytest +from napalm_openwrt.openwrt import OpenWrtDriver # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- + @pytest.fixture def driver(): """Return a driver instance with a mocked Netmiko connection.""" @@ -130,9 +131,11 @@ 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: @@ -146,6 +149,7 @@ class TestGetFacts: if "ip link" in key: return IP_LINK_SHOW return "" + return _send def test_returns_required_keys(self, driver): @@ -153,8 +157,15 @@ class TestGetFacts: facts = driver.get_facts() assert set(facts.keys()) == { - "vendor", "model", "hostname", "fqdn", "os_version", - "serial_number", "uptime", "interface_list", "number_of_interfaces", + "vendor", + "model", + "hostname", + "fqdn", + "os_version", + "serial_number", + "uptime", + "interface_list", + "number_of_interfaces", } def test_number_of_interfaces_matches_list(self, driver): @@ -213,8 +224,13 @@ class TestGetInterfaces: 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", + "is_up", + "is_enabled", + "description", + "last_flapped", + "speed", + "mtu", + "mac_address", } @@ -658,6 +674,7 @@ 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}" @@ -758,8 +775,17 @@ class TestGetNtpStats: result = driver.get_ntp_stats() for entry in result: assert set(entry.keys()) >= { - "remote", "referenceid", "synchronized", "stratum", - "type", "when", "hostpoll", "reachability", "delay", "offset", "jitter", + "remote", + "referenceid", + "synchronized", + "stratum", + "type", + "when", + "hostpoll", + "reachability", + "delay", + "offset", + "jitter", } def test_empty_on_no_tool(self, driver): @@ -888,8 +914,12 @@ class TestGetRouteTo: 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", + "protocol", + "current_active", + "next_hop", + "outgoing_interface", + "preference", + "routing_table", } @@ -925,23 +955,17 @@ class TestTraceroute: 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 "" - ) + 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 "" - ) + 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 "" - ) + 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 @@ -955,6 +979,7 @@ class TestGetNetworkInstances: if "ip link" in cmd: return IP_LINK_SHOW return "" + driver._send_command = _send result = driver.get_network_instances() assert "vpn" in result @@ -962,16 +987,12 @@ class TestGetNetworkInstances: 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 "" - ) + 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 "" - ) + 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"} @@ -1057,6 +1078,7 @@ class TestGetChannelScan: 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 @@ -1065,6 +1087,7 @@ class TestGetChannelScan: if "wlan1" in cmd and "scan" in cmd: return IW_SCAN_WLAN1 return "" + return _send def test_returns_dict(self, driver): @@ -1089,8 +1112,13 @@ class TestGetChannelScan: 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", + "bssid", + "ssid", + "frequency", + "channel", + "signal_dbm", + "channel_width", + "band", } def test_wlan0_first_entry_values(self, driver): @@ -1300,7 +1328,9 @@ class TestGetSsidsAcl: 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 "" + 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" @@ -1308,7 +1338,9 @@ class TestGetSsidsAcl: 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): + 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", @@ -1370,16 +1402,24 @@ class TestPushMacAcl: 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) + 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) + 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): @@ -1474,8 +1514,17 @@ IP_ADDR_BRLAN = """\ 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"): + 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 @@ -1633,7 +1682,9 @@ class TestFixSnmpRuleWriting: ) driver._send_command = shell driver.configure_snmp_agent(_SETTINGS) - delete_idx = next(i for i, c in enumerate(shell.issued) if "snmp_netork" in c and "delete" in c) + 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): diff --git a/tests/unit/test_ssh_host_key.py b/tests/unit/test_ssh_host_key.py index 5193dfa..7250df4 100644 --- a/tests/unit/test_ssh_host_key.py +++ b/tests/unit/test_ssh_host_key.py @@ -24,7 +24,9 @@ HANDLER = "napalm_openwrt.openwrt.ConnectHandler" def _device() -> MagicMock: device = MagicMock() - device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY + device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = ( + SERVER_KEY + ) return device