diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..25a750f --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,15 @@ +name: CI + +on: + push: + branches: [main, master] + tags: ["v*"] + pull_request: + +jobs: + driver: + uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main + with: + standard-ref: main + secrets: + PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml deleted file mode 100644 index a7bc4d4..0000000 --- a/.github/workflows/ci.yml +++ /dev/null @@ -1,48 +0,0 @@ -name: CI - -on: - push: - branches: ["**"] - pull_request: - branches: ["**"] - -jobs: - test: - runs-on: ubuntu-latest - strategy: - fail-fast: false - matrix: - python-version: ["3.10", "3.11", "3.12"] - steps: - - name: Checkout - uses: actions/checkout@v4 - - - name: Setup Python - uses: actions/setup-python@v5 - with: - python-version: ${{ matrix.python-version }} - cache: pip - - - name: Install package with dev extras - run: | - python -m pip install --upgrade pip - # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this - # pip looks there, finds an unrelated 0.1.0 and the job dies before any test. - python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git" - python -m pip install -e ".[dev]" - - - name: Run unit tests - run: | - python -m pytest -q --tb=short - - - name: Build wheel and sdist - run: | - python -m pip install build - python -m build - - - name: Upload dist artifacts - # v4 refuses to run on Gitea ("not currently supported on GHES"). - uses: actions/upload-artifact@v3 - with: - name: dist-${{ matrix.python-version }} - path: dist/* \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..d911400 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,25 @@ +# Changelog + +All notable changes to napalm-openwrt. The format follows +[Keep a Changelog](https://keepachangelog.com/), versions follow +[Semantic Versioning](https://semver.org/). Versions up to 0.6.0 were not +recorded here; their history is the git log. + +## [Unreleased] + +## [0.7.0] - 2026-10-23 + +The first release on PyPI. + +### Changed +- The driver follows the NAPALM driver standard + (https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the + tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and + a build check, and a release tag publishes to PyPI. The package now ships `py.typed` + and a LICENSE file. **Python 3.8 and 3.9 are no longer supported**; + napalm-device-types never ran on them. +- napalm-device-types 4.21.1 or newer is required. +- Exceptions the driver deliberately ignores (the snmpd community in + `get_snmp_config()`, the network VLANs in `get_ssids()`) are now logged at debug + level. +- The project links point at git.netork.io instead of GitHub. diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..68d39ec --- /dev/null +++ b/LICENSE @@ -0,0 +1,191 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship made available under + the License, as indicated by a copyright notice that is included in + or attached to the work (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other transformations + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean, as submitted to the Licensor for inclusion + in the Work by the copyright owner or by an individual or Legal Entity + authorized to submit on behalf of the copyright owner. For the purposes + of this definition, "submitted" means any form of electronic, verbal, + or written communication sent to the Licensor or its representatives, + including but not limited to communication on electronic mailing lists, + source code control systems, and issue tracking systems that are managed + by, or on behalf of, the Licensor for the purpose of discussing and + improving the Work, but excluding communication that is conspicuously + marked or designated in writing by the copyright owner as "Not a + Contribution." + + "Contributor" shall mean Licensor and any Legal Entity on behalf of + whom a Contribution has been received by the Licensor and incorporated + within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by the combined Contribution(s) and the Work + to which such Contribution(s) was submitted. If You institute patent + litigation against any entity (including a cross-claim or counterclaim + in a lawsuit) alleging that the Work or any Contribution embodied + within the Work constitutes direct or contributory patent infringement, + then any patent licenses granted to You under this License for that + Work shall terminate as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or Derivative + Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, You must include a readable copy of the + attribution notices contained within such NOTICE file, in + at least one of the following places: within a NOTICE text file + distributed as part of the Derivative Works; within the Source + form or documentation, if provided along with the Derivative + Works; or, within a display generated by the Derivative Works, + if and wherever such third-party notices normally appear. The + contents of the NOTICE file are for informational purposes only + and do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own license statement for Your modifications and + may provide additional grant of rights to use, copy, modify, merge, + publish, distribute, sublicense, and/or sell copies of the Work, + and to permit persons to whom the Work is furnished to do so, + subject to the following conditions: + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any conditions of TITLE, + MERCHANTIBILITY, or FITNESS FOR A PARTICULAR PURPOSE. + See the License for the specific language governing permissions and + limitations under the License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or exemplary damages of any character arising as a + result of this License or out of the use or inability to use the + Work (even if such Contributor has been advised of the possibility + of such damages). + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may offer only + conditions consistent with this License, subject to the following + additional conditions: + + (a) You may offer additional warranty, indemnity, or other liability + terms and conditions; and + + (b) You may offer support for the Work in return for a fee. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format in question. It may also be + possible to make it available with a creative commons license. + + Copyright 2026 Christian Manivong + + 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 + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md index 6d41139..16bff6e 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ NAPALM community driver for **OpenWrt** routers and access-points. -Communicates over SSH using [Netmiko](https://github.com/ktbyers/netmiko) (`linux` device type). +Communicates over SSH using [Netmiko](https://pypi.org/project/netmiko/) (`linux` device type). Requires OpenWrt **19.07** or newer. ## Tested devices @@ -17,7 +17,7 @@ Requires OpenWrt **19.07** or newer. | Dependency | Minimum version | |---|---| -| Python | 3.8 | +| Python | 3.10 | | NAPALM | 4.0 | | Netmiko | 4.0 | @@ -26,7 +26,7 @@ Requires OpenWrt **19.07** or newer. From source: ```bash -git clone https://github.com/napalm-automation-community/napalm-openwrt +git clone https://git.netork.io/NAPALM/napalm-openwrt cd napalm-openwrt pip install -e . ``` @@ -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..9c8f26d 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") @@ -90,7 +91,8 @@ class OpenWrtConfigMixin(ConfigLifecycleMixin): """ if self._backup_config is None: raise CommandErrorException( - "No backup configuration available – commit_config has not been called in this session." + "No backup configuration available – " + "commit_config has not been called in this session." ) escaped = self._backup_config.replace("'", "'\\''") 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/py.typed b/napalm_openwrt/py.typed new file mode 100644 index 0000000..e69de29 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..63b0b9b 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 @@ -13,10 +12,11 @@ from __future__ import annotations +import logging import re from typing import Any - +logger = logging.getLogger(__name__) #: Printed after a command so its exit status survives the shell session. _RC_MARKER = "__OWRT_RC=" @@ -31,8 +31,12 @@ 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).""" + """Mixin providing system-level NAPALM getters. + + Environment, NTP, SNMP, users, services, updates and device actions. + """ def get_environment(self) -> dict[str, Any]: """Return device environment data (CPU, memory). @@ -41,7 +45,8 @@ class OpenWrtSystemMixin: Memory from ``/proc/meminfo``. """ cpu_out = self._send_command( - "awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} END{print (1-(idle1/total1))*100}' /proc/stat" + "awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} " + "END{print (1-(idle1/total1))*100}' /proc/stat" ) mem_out = self._send_command("cat /proc/meminfo") @@ -95,9 +100,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 +241,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 @@ -253,7 +257,7 @@ class OpenWrtSystemMixin: if uci_comm: community = uci_comm except Exception: - pass + logger.debug("snmpd community: ignored", exc_info=True) return SNMPConfigDict(running=True, community=community, port=161, version="2c") @@ -282,7 +286,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 +335,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 +353,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 +377,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 +426,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 +447,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 +465,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 +549,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 +580,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..d2d0e0e 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 @@ -13,9 +12,12 @@ from __future__ import annotations +import logging import re from typing import Any +logger = logging.getLogger(__name__) + class OpenWrtWirelessMixin: """Mixin providing wireless/radio NAPALM getters.""" @@ -101,8 +103,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", @@ -152,7 +158,8 @@ class OpenWrtWirelessMixin: if vlan is not None: net_vlan[iface] = vlan except Exception: - pass # Non-fatal: VLAN info is optional enrichment + # Non-fatal: VLAN info is optional enrichment. + logger.debug("network VLANs: ignored", exc_info=True) # Build result; group entries with the same SSID name, merging bands result: dict[str, Any] = {} @@ -176,11 +183,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 +344,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 +425,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 +449,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 +500,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 +676,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 +702,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..a6cd5f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,23 +1,20 @@ [build-system] -requires = ["setuptools>=64", "wheel"] +requires = ["setuptools>=77"] build-backend = "setuptools.build_meta" [project] name = "napalm-openwrt" -version = "0.6.0" +version = "0.7.0" description = "NAPALM driver for OpenWrt routers/access-points" readme = "README.md" -license = { text = "Apache-2.0" } -requires-python = ">=3.8" +license = "Apache-2.0" +requires-python = ">=3.10" authors = [ { name = "Christian Manivong" }, ] classifiers = [ "Topic :: Utilities", - "License :: OSI Approved :: Apache Software License", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.8", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", @@ -26,30 +23,71 @@ classifiers = [ ] dependencies = [ "napalm>=4.0.0", - "napalm_device_types>=4.19.0", + "napalm_device_types>=4.21.1", "netmiko>=4.0.0", "paramiko>=5.0.0", # CVE-2026-44405 - "netaddr", + "netaddr>=0.7.19", # IPAddress; below it the package has none (CI lowest job) ] [project.optional-dependencies] -dev = [ - "pytest", - "pytest-cov", - "black", - "ruff", -] +dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"] [project.entry-points."napalm.drivers"] openwrt = "napalm_openwrt:OpenWrtDriver" [project.urls] -Repository = "https://github.com/napalm-automation-community/napalm-openwrt" +Repository = "https://git.netork.io/NAPALM/napalm-openwrt" +Issues = "https://git.netork.io/NAPALM/napalm-openwrt/issues" +Changelog = "https://git.netork.io/NAPALM/napalm-openwrt/src/branch/main/CHANGELOG.md" [tool.setuptools.packages.find] where = ["."] include = ["napalm_openwrt*"] +[tool.setuptools.package-data] +napalm_openwrt = ["py.typed"] + [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"] + +[tool.mypy] +python_version = "3.10" +strict = true +# napalm itself is untyped, so every driver subclasses Any. +disallow_subclassing_any = false + +[[tool.mypy.overrides]] +module = ["napalm.*", "netmiko.*", "paramiko.*", "netaddr.*"] +ignore_missing_imports = true + +[[tool.mypy.overrides]] +# Not yet under strict mypy (quality baseline); may only shrink. +module = [ + "napalm_openwrt.config_mixin", + "napalm_openwrt.interfaces_mixin", + "napalm_openwrt.lldp_mixin", + "napalm_openwrt.openwrt", + "napalm_openwrt.packages_mixin", + "napalm_openwrt.routing_mixin", + "napalm_openwrt.system_mixin", + "napalm_openwrt.vlan_mixin", + "napalm_openwrt.wireless_mixin", +] +ignore_errors = true + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.coverage.run] +source = ["napalm_openwrt"] + +[tool.coverage.report] +fail_under = 65 diff --git a/tests/quality_baseline.json b/tests/quality_baseline.json new file mode 100644 index 0000000..0cbab18 --- /dev/null +++ b/tests/quality_baseline.json @@ -0,0 +1,55 @@ +{ + "coverage_floor": 65.0, + "function_complexity": { + "napalm_openwrt/config_mixin.py::OpenWrtConfigMixin.commit_config": 14, + "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_arp_table": 13, + "napalm_openwrt/openwrt.py::OpenWrtDriver._allow_snmp_sources": 22, + "napalm_openwrt/openwrt.py::OpenWrtDriver._write_snmp_agent_config": 22, + "napalm_openwrt/packages_mixin.py::OpenWrtPackageMixin.search_packages": 12, + "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin._parse_traceroute_output": 12, + "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin.get_route_to": 12, + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_services": 14, + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_system_config": 16, + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_users": 13, + "napalm_openwrt/vlan_mixin.py::OpenWrtVLANMixin.get_vlans": 32, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_channel_scan": 12, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_radio_status": 19, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_ssids": 51, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_wireless_clients": 27 + }, + "function_lines": { + "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_arp_table": 85, + "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_interfaces_ip": 55, + "napalm_openwrt/interfaces_mixin.py::OpenWrtInterfaceMixin.get_mac_address_table": 53, + "napalm_openwrt/openwrt.py::OpenWrtDriver._allow_snmp_sources": 90, + "napalm_openwrt/openwrt.py::OpenWrtDriver.get_facts": 53, + "napalm_openwrt/packages_mixin.py::OpenWrtPackageMixin.search_packages": 59, + "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin._parse_traceroute_output": 67, + "napalm_openwrt/routing_mixin.py::OpenWrtRoutingMixin.get_route_to": 91, + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_services": 61, + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_snmp_information": 56, + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_system_config": 92, + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_users": 64, + "napalm_openwrt/vlan_mixin.py::OpenWrtVLANMixin.get_network_instances": 61, + "napalm_openwrt/vlan_mixin.py::OpenWrtVLANMixin.get_vlans": 106, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin._parse_iw_scan": 93, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_radio_status": 133, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_ssids": 240, + "napalm_openwrt/wireless_mixin.py::OpenWrtWirelessMixin.get_wireless_clients": 152 + }, + "mypy_relaxed": [ + "napalm_openwrt.config_mixin", + "napalm_openwrt.interfaces_mixin", + "napalm_openwrt.lldp_mixin", + "napalm_openwrt.openwrt", + "napalm_openwrt.packages_mixin", + "napalm_openwrt.routing_mixin", + "napalm_openwrt.system_mixin", + "napalm_openwrt.vlan_mixin", + "napalm_openwrt.wireless_mixin" + ], + "python_file_lines": {}, + "untyped_public": [ + "napalm_openwrt/system_mixin.py::OpenWrtSystemMixin.get_snmp_config" + ] +} 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..a5f9b75 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): @@ -885,11 +911,15 @@ class TestGetRouteTo: 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 _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..03ffddd 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 @@ -74,7 +76,7 @@ def test_a_changed_key_is_refused(): try: raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY) except paramiko.SSHException as exc: # what netmiko does with it - raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") + raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc driver = _driver(seen, SshHostKey.of(OTHER_KEY)) with patch(HANDLER, side_effect=refuse):