NAPALM/napalm-openwrt
1
0 Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

style: ruff format and the lint rules of the driver standard

Browse Source 
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM
driver uses (E, F, I, UP, B at line length 100, Python 3.10). What ruff
cannot fix comes in the next commit.
This commit is contained in:
Christian Manivong committed 2026-10-23 08:32:47 +02:00
1 parent 7f2ae6ac58
commit a212bc8715
14 files changed
+322 -239
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

README.md
+4 -2
View File
@@ -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()
napalm_openwrt/config_mixin.py
+3 -2
View File
@@ -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")
napalm_openwrt/interfaces_mixin.py
+11 -11
View File
@@ -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
napalm_openwrt/lldp_mixin.py
-1
View File
@@ -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
napalm_openwrt/openwrt.py
+22 -47
View File
@@ -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
napalm_openwrt/packages_mixin.py
+41 -32
View File
@@ -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")
napalm_openwrt/routing_mixin.py
+1 -2
View File
@@ -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]
napalm_openwrt/system_mixin.py
+72 -59
View File
@@ -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<priority><name> 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
napalm_openwrt/vlan_mixin.py
+8 -7
View File
@@ -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).
napalm_openwrt/wireless_mixin.py
+56 -35
View File
@@ -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,
pyproject.toml
+8 -1
View File
@@ -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"]
tests/unit/test_command_channel.py
+6 -3
View File
@@ -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()
tests/unit/test_driver.py
+87 -36
View File
@@ -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):
tests/unit/test_ssh_host_key.py
+3 -1
View File
@@ -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