From 571d9abb5fc395e398a97f41342e935ac7b6c232 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 23 Oct 2026 08:32:41 +0200 Subject: [PATCH 1/4] style: ruff format and the lint rules of the driver standard 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. --- README.md | 8 +- napalm_opnsense/listening.py | 21 +- napalm_opnsense/opnsense.py | 566 ++++++++++++++++------------ napalm_opnsense/ospf_mixin.py | 4 +- napalm_opnsense/ping_mixin.py | 18 +- napalm_opnsense/port_forwards.py | 28 +- napalm_opnsense/resolver_mixin.py | 14 +- napalm_opnsense/virtual_ips.py | 12 +- pyproject.toml | 9 +- tests/unit/test_driver.py | 330 ++++++++++------ tests/unit/test_ha.py | 10 +- tests/unit/test_ping.py | 14 +- tests/unit/test_port_forwards.py | 16 +- tests/unit/test_public_contracts.py | 53 ++- tests/unit/test_snmp_agent.py | 6 +- 15 files changed, 690 insertions(+), 419 deletions(-) diff --git a/README.md b/README.md index 68092ef..6496f1f 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,7 @@ API keys**. Pass the credentials via `optional_args`: ```python -optional_args={ +optional_args = { "api_key": "your-api-key", "api_secret": "your-api-secret", "verify": False, # set to a CA bundle path or True in production @@ -173,9 +173,9 @@ d.open() routes = json.dumps([{"network": "10.0.0.0/8", "gateway": "WAN_GW"}]) d.load_merge_candidate(config=routes) -print(d.compare_config()) # unified diff -d.commit_config() # applies routes and calls reconfigure -d.rollback() # removes the routes just added +print(d.compare_config()) # unified diff +d.commit_config() # applies routes and calls reconfigure +d.rollback() # removes the routes just added d.close() ``` diff --git a/napalm_opnsense/listening.py b/napalm_opnsense/listening.py index da29596..a20c306 100644 --- a/napalm_opnsense/listening.py +++ b/napalm_opnsense/listening.py @@ -21,14 +21,15 @@ told by the listen ports of the WireGuard interfaces. from __future__ import annotations -from typing import Any, Dict, Iterable, List, Optional, Set +from collections.abc import Iterable +from typing import Any from napalm_device_types.models import ListeningSocketDict #: FRR's daemons, all under the one service ``frr``. _FRR_DAEMONS = "zebra mgmtd staticd ospfd ospf6d bgpd bfdd ripd ripngd isisd pimd watchfrr".split() #: Daemons whose OPNsense service has another name, by the command as reported. -_SERVICE_OF: Dict[str, str] = { +_SERVICE_OF: dict[str, str] = { "sshd": "openssh", "kea-ctrl-a": "kea-dhcp", "kea-ctrl-agent": "kea-dhcp", @@ -43,7 +44,7 @@ _NOBODY = "??" _PEERLESS = "*" -def _service_of(command: str, user: str, services: Set[str]) -> Optional[str]: +def _service_of(command: str, user: str, services: set[str]) -> str | None: if command == "lighttpd": name = "captiveportal" if user == "www" else "webgui" return name if name in services else None @@ -70,8 +71,8 @@ def _address(protocol: str, address: str) -> tuple: def _socket( - entry: Dict[str, Any], services: Set[str], wireguard_ports: Set[int] -) -> Optional[ListeningSocketDict]: + entry: dict[str, Any], services: set[str], wireguard_ports: set[int] +) -> ListeningSocketDict | None: protocol = str(entry.get("protocol") or "") proto = protocol[:3] local, remote = entry.get("local") or {}, entry.get("remote") or {} @@ -101,9 +102,9 @@ def _socket( } -def wireguard_listen_ports(show: Dict[str, Any]) -> Set[int]: +def wireguard_listen_ports(show: dict[str, Any]) -> set[int]: """The listen ports of the WireGuard interfaces in ``/api/wireguard/service/show``.""" - ports: Set[int] = set() + ports: set[int] = set() for row in show.get("rows") or []: port = str(row.get("listen-port") or "") if row.get("type") == "interface" and port.isdigit(): @@ -112,12 +113,12 @@ def wireguard_listen_ports(show: Dict[str, Any]) -> Set[int]: def parse_socket_statistics( - statistics: Dict[str, Any], services: Iterable[str], wireguard_ports: Set[int] -) -> List[ListeningSocketDict]: + statistics: dict[str, Any], services: Iterable[str], wireguard_ports: set[int] +) -> list[ListeningSocketDict]: """The listening sockets of ``get_socket_statistics``, sorted by protocol, port and address. Every section is read: the UNIX sockets have no ``local``.""" names = set(services) - sockets: List[ListeningSocketDict] = [] + sockets: list[ListeningSocketDict] = [] for section in (statistics.get("statistics") or {}).values(): for entry in (section or {}).values() if isinstance(section, dict) else (): socket = _socket(entry, names, wireguard_ports) if isinstance(entry, dict) else None diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 623e35a..5722eba 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.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 @@ -52,29 +51,31 @@ _REFRESH_POLLS = 20 _REFRESH_INTERVAL = 3 import requests -from requests.exceptions import RequestException - +from napalm.base.exceptions import ( + ConnectionClosedException, + ConnectionException, + MergeConfigException, +) from napalm_device_types import ConfigSyncMixin, FingerprintRule, FirewallDriver, SnmpAgentMixin from napalm_device_types.dhcp import normalize_mac from napalm_device_types.models import ( - FixStepOutcomeDict, - SnmpAgentSettingsDict, ConfigSyncDict, DhcpHaDict, + FixStepOutcomeDict, ListeningSocketsDict, + SnmpAgentSettingsDict, VirtualIpsDict, ) -from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException +from requests.exceptions import RequestException from napalm_opnsense.ha import config_sync, dhcp_ha from napalm_opnsense.listening import parse_socket_statistics, wireguard_listen_ports -from napalm_opnsense.ping_mixin import OPNsensePingMixin from napalm_opnsense.ospf_mixin import OPNsenseOspfMixin -from napalm_opnsense.resolver_mixin import OPNsenseResolverMixin +from napalm_opnsense.ping_mixin import OPNsensePingMixin from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces +from napalm_opnsense.resolver_mixin import OPNsenseResolverMixin from napalm_opnsense.virtual_ips import carp_addresses - _ARCHITECTURE_SUFFIX = re.compile(r"-(amd64|aarch64|arm64|i386|armv7)$") @@ -137,7 +138,10 @@ class OPNsenseDriver( # OPNsense REST API settings self.base_url = self.optional_args.get("base_url") or f"https://{hostname}" # Accept "verify", "verify_ssl", or "ssl_verify" (NetOrk passes "verify_ssl") - _v = self.optional_args.get("verify", self.optional_args.get("verify_ssl", self.optional_args.get("ssl_verify", True))) + _v = self.optional_args.get( + "verify", + self.optional_args.get("verify_ssl", self.optional_args.get("ssl_verify", True)), + ) self.verify = bool(_v) self.api_key = self.optional_args.get("api_key") or username self.api_secret = self.optional_args.get("api_secret") or password @@ -190,7 +194,7 @@ class OPNsenseDriver( with socket.create_connection((host, port), timeout=5): pass return {"is_alive": True} - except (socket.error, OSError): + except OSError: return {"is_alive": False} # ------------------------------------------------------------------ @@ -319,7 +323,9 @@ class OPNsenseDriver( if not name: continue interfaces[name] = { - "is_up": iface.get("status", "") == "up" if "status" in iface else bool(iface.get("up", False)), + "is_up": iface.get("status", "") == "up" + if "status" in iface + else bool(iface.get("up", False)), "is_enabled": bool(iface.get("enabled", True)), "description": iface.get("description") or iface.get("descr") or "", "last_flapped": -1.0, @@ -434,14 +440,16 @@ class OPNsenseDriver( net = iface_obj.network if net.is_loopback or net.is_link_local: return - networks.append({ - "network": str(net), - "interface": ifname, - "gateway": str(iface_obj.ip), - "family": family, - "prefix_length": net.prefixlen, - "vlan_id": vlan_id, - }) + networks.append( + { + "network": str(net), + "interface": ifname, + "gateway": str(iface_obj.ip), + "family": family, + "prefix_length": net.prefixlen, + "vlan_id": vlan_id, + } + ) except ValueError: pass @@ -605,7 +613,9 @@ class OPNsenseDriver( ospf_networks: set = set() try: ospf_data = self._get("/api/quagga/ospf/routes") - for prefix in (ospf_data if isinstance(ospf_data, list) else ospf_data.get("routes", {}).keys()): + for prefix in ( + ospf_data if isinstance(ospf_data, list) else ospf_data.get("routes", {}).keys() + ): ospf_networks.add(str(prefix)) except Exception as exc: logger.debug("Failed to fetch OSPF routes: %s", exc) @@ -624,7 +634,11 @@ class OPNsenseDriver( iface = route.get("netif") or route.get("interface", "") # Clean up BSD link-layer gateway references - clean_gateway = "" if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") else gateway + clean_gateway = ( + "" + if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") + else gateway + ) # Determine address family from network address family = "ipv6" if (":" in network or (gateway and ":" in gateway)) else "ipv4" @@ -711,9 +725,7 @@ class OPNsenseDriver( ) return neighbors - def get_lldp_neighbors_detail( - self, interface: str = "" - ) -> dict[str, list[dict[str, Any]]]: + def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[dict[str, Any]]]: """Return detailed LLDP neighbor information. Calls ``GET /api/lldpd/service/neighbor``. @@ -978,7 +990,7 @@ class OPNsenseDriver( if filename is not None: try: - with open(filename, "r", encoding="utf-8") as fh: + with open(filename, encoding="utf-8") as fh: config = fh.read() except OSError as exc: raise MergeConfigException(f"Cannot read file {filename!r}: {exc}") from exc @@ -989,16 +1001,12 @@ class OPNsenseDriver( raise MergeConfigException(f"Invalid JSON in candidate config: {exc}") from exc if not isinstance(routes, list): - raise MergeConfigException( - "Candidate config must be a JSON array of route objects." - ) + raise MergeConfigException("Candidate config must be a JSON array of route objects.") for i, route in enumerate(routes): if not isinstance(route, dict): raise MergeConfigException(f"Route at index {i} must be a JSON object.") if "network" not in route or "gateway" not in route: - raise MergeConfigException( - f"Route at index {i} is missing 'network' or 'gateway'." - ) + raise MergeConfigException(f"Route at index {i} is missing 'network' or 'gateway'.") self._candidate_config = routes @@ -1041,7 +1049,9 @@ class OPNsenseDriver( API returns an error for any route. """ if self._candidate_config is None: - raise MergeConfigException("No candidate config loaded. Call load_merge_candidate() first.") + raise MergeConfigException( + "No candidate config loaded. Call load_merge_candidate() first." + ) # Record the current backup ID so rollback() can restore exactly # this state after OPNsense writes the new config to disk. @@ -1147,14 +1157,16 @@ class OPNsenseDriver( for p in info.get("plugin", []): if p.get("installed") != "1": continue - result.append({ - "name": p.get("name", ""), - "version": p.get("version", ""), - "installed": True, - "description": p.get("comment", ""), - "size": 0, - "source": "opnsense-plugins", - }) + result.append( + { + "name": p.get("name", ""), + "version": p.get("version", ""), + "installed": True, + "description": p.get("comment", ""), + "size": 0, + "source": "opnsense-plugins", + } + ) return sorted(result, key=lambda x: x["name"].lower()) def get_certificates(self) -> list[dict[str, Any]]: @@ -1180,13 +1192,15 @@ class OPNsenseDriver( result: list[dict[str, Any]] = [] for row in data.get("rows", []): - result.append({ - "name": row.get("commonname") or row.get("descr") or row.get("refid", ""), - "issuer": row.get("%caref") or "", - "valid_from": _as_int(row.get("valid_from")), - "valid_to": _as_int(row.get("valid_to")), - "in_use_by": _as_int(row.get("in_use")), - }) + result.append( + { + "name": row.get("commonname") or row.get("descr") or row.get("refid", ""), + "issuer": row.get("%caref") or "", + "valid_from": _as_int(row.get("valid_from")), + "valid_to": _as_int(row.get("valid_to")), + "in_use_by": _as_int(row.get("in_use")), + } + ) return result def get_ddns_status(self) -> dict[str, Any] | None: @@ -1336,13 +1350,16 @@ class OPNsenseDriver( * ``lease_end`` — Unix timestamp when the lease expires (0 if unknown) * ``state`` — raw state string """ + def _parse_kea(rows: list[dict]) -> list[dict[str, Any]]: result = [] for row in rows: # OPNsense Kea uses "hwaddr"; standard Kea uses "hw-address"; ISC DHCP uses "mac" mac = ( - row.get("hwaddr") or row.get("hw-address") or row.get("mac") or "" - ).lower().strip() + (row.get("hwaddr") or row.get("hw-address") or row.get("mac") or "") + .lower() + .strip() + ) if not mac: continue raw_end = row.get("expire") or row.get("ends") or 0 @@ -1350,13 +1367,17 @@ class OPNsenseDriver( lease_end = int(raw_end) except (ValueError, TypeError): lease_end = 0 - result.append({ - "mac": mac, - "ip": (row.get("address") or row.get("ip-address") or row.get("ip") or "").strip(), - "hostname": (row.get("hostname") or "").strip(), - "lease_end": lease_end, - "state": str(row.get("state") or ""), - }) + result.append( + { + "mac": mac, + "ip": ( + row.get("address") or row.get("ip-address") or row.get("ip") or "" + ).strip(), + "hostname": (row.get("hostname") or "").strip(), + "lease_end": lease_end, + "state": str(row.get("state") or ""), + } + ) return result failures: list[str] = [] @@ -1365,7 +1386,9 @@ class OPNsenseDriver( # 1. Kea DHCPv4 plugin try: data = self._get("/api/kea/leases4/search") - rows = data.get("rows") or data.get("leases") or (data if isinstance(data, list) else []) + rows = ( + data.get("rows") or data.get("leases") or (data if isinstance(data, list) else []) + ) leases = _parse_kea(rows) if leases: return leases @@ -1654,8 +1677,7 @@ class OPNsenseDriver( return [] cidr_by_uuid = { - str(row.get("uuid") or ""): str(row.get("subnet") or "") - for row in self._kea_subnets() + str(row.get("uuid") or ""): str(row.get("subnet") or "") for row in self._kea_subnets() } known_cidrs = set(cidr_by_uuid.values()) @@ -1667,14 +1689,16 @@ class OPNsenseDriver( else: subnet = cidr_by_uuid.get(raw_subnet, "") - reservations.append({ - "uuid": str(row.get("uuid") or ""), - "mac": str(row.get("hw_address") or ""), - "ip": str(row.get("ip_address") or ""), - "hostname": str(row.get("hostname") or ""), - "description": str(row.get("description") or ""), - "subnet": subnet, - }) + reservations.append( + { + "uuid": str(row.get("uuid") or ""), + "mac": str(row.get("hw_address") or ""), + "ip": str(row.get("ip_address") or ""), + "hostname": str(row.get("hostname") or ""), + "description": str(row.get("description") or ""), + "subnet": subnet, + } + ) return reservations def apply_dhcp_reservation( @@ -1704,9 +1728,7 @@ class OPNsenseDriver( } } path = ( - f"/api/kea/dhcpv4/setReservation/{uuid}" - if uuid - else "/api/kea/dhcpv4/addReservation" + f"/api/kea/dhcpv4/setReservation/{uuid}" if uuid else "/api/kea/dhcpv4/addReservation" ) result = self._post(path, payload) if result.get("result") != "saved": @@ -1824,19 +1846,19 @@ class OPNsenseDriver( option_data[name] = value pools = [ - line.strip() - for line in str(record.get("pools") or "").splitlines() - if line.strip() + line.strip() for line in str(record.get("pools") or "").splitlines() if line.strip() ] - subnets.append({ - "uuid": uuid, - "subnet": str(record.get("subnet") or row.get("subnet") or ""), - "description": str(record.get("description") or ""), - "pools": pools, - "option_data": option_data, - "match_client_id": str(record.get("match-client-id") or "0") == "1", - }) + subnets.append( + { + "uuid": uuid, + "subnet": str(record.get("subnet") or row.get("subnet") or ""), + "description": str(record.get("description") or ""), + "pools": pools, + "option_data": option_data, + "match_client_id": str(record.get("match-client-id") or "0") == "1", + } + ) return subnets def apply_dhcp_subnet( @@ -1894,14 +1916,10 @@ class OPNsenseDriver( record["option_data"] = merged_options record["option_data_autocollect"] = "0" - path = ( - f"/api/kea/dhcpv4/setSubnet/{uuid}" if uuid else "/api/kea/dhcpv4/addSubnet" - ) + path = f"/api/kea/dhcpv4/setSubnet/{uuid}" if uuid else "/api/kea/dhcpv4/addSubnet" result = self._post(path, {"subnet": record}) if result.get("result") != "saved": - raise RuntimeError( - f"Kea rejected DHCP subnet {subnet.get('subnet')}: {result}" - ) + raise RuntimeError(f"Kea rejected DHCP subnet {subnet.get('subnet')}: {result}") return {"success": True} def commit_dhcp_subnets(self) -> dict[str, Any]: @@ -1919,12 +1937,14 @@ class OPNsenseDriver( data = self._get("/api/core/service/search") result: list[dict[str, Any]] = [] for row in data.get("rows", []): - result.append({ - "name": row.get("name") or row.get("id", ""), - "running": bool(row.get("running", 0)), - "enabled": True, # OPNsense has no separate enabled/disabled state - "pid": 0, - }) + result.append( + { + "name": row.get("name") or row.get("id", ""), + "running": bool(row.get("running", 0)), + "enabled": True, # OPNsense has no separate enabled/disabled state + "pid": 0, + } + ) return sorted(result, key=lambda x: x["name"].lower()) def get_listening_sockets(self) -> ListeningSocketsDict: @@ -1970,9 +1990,7 @@ class OPNsenseDriver( :raises: whatever either request raises -- an unread answer is not "off". """ - return dhcp_ha( - self._get("/api/kea/dhcpv4/get"), self._get("/api/kea/dhcpv4/searchPeer") - ) + return dhcp_ha(self._get("/api/kea/dhcpv4/get"), self._get("/api/kea/dhcpv4/searchPeer")) def get_config_sync(self) -> ConfigSyncDict: """Where this firewall pushes its configuration (HA sync), and for which services. @@ -1993,10 +2011,14 @@ class OPNsenseDriver( are enabled/disabled by installing or removing them). """ 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'): - return {"success": False, "output": f"Action '{action}' is not supported on OPNsense services"} + if action not in ("start", "stop", "restart"): + return { + "success": False, + "output": f"Action '{action}' is not supported on OPNsense services", + } try: result = self._post(f"/api/{name}/service/{action}") return {"success": True, "output": str(result)} @@ -2061,17 +2083,17 @@ class OPNsenseDriver( # Byte counters may live in child-sa entries bytes_in = 0 bytes_out = 0 - for child in session.get("child-sas", {}).values() if isinstance(session.get("child-sas"), dict) else []: + for child in ( + session.get("child-sas", {}).values() + if isinstance(session.get("child-sas"), dict) + else [] + ): try: bytes_in += int(child.get("bytes-in", 0) or 0) bytes_out += int(child.get("bytes-out", 0) or 0) except (ValueError, TypeError): pass - description = ( - session.get("local-id") - or session.get("description") - or "" - ) + description = session.get("local-id") or session.get("description") or "" key = f"ipsec-{name}" tunnels[key] = { "type": "IPsec", @@ -2271,7 +2293,9 @@ class OPNsenseDriver( """ try: if packages: - result = self._post("/api/core/firmware/upgrade", {"upgrade_packages": list(packages)}) + result = self._post( + "/api/core/firmware/upgrade", {"upgrade_packages": list(packages)} + ) else: result = self._post("/api/core/firmware/upgrade") return {"success": True, "output": str(result)} @@ -2302,14 +2326,16 @@ class OPNsenseDriver( name = p.get("name", "") comment = p.get("comment", "") if q in name.lower() or q in comment.lower(): - result.append({ - "name": name, - "version": p.get("version", ""), - "installed": p.get("installed") == "1", - "description": comment, - "size": 0, - "source": "opnsense-plugins", - }) + result.append( + { + "name": name, + "version": p.get("version", ""), + "installed": p.get("installed") == "1", + "description": comment, + "size": 0, + "source": "opnsense-plugins", + } + ) return sorted(result, key=lambda x: x["name"].lower()) def install_package(self, name: str) -> dict[str, Any]: @@ -2318,7 +2344,8 @@ class OPNsenseDriver( Calls ``POST /api/core/firmware/install/{name}``. """ 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 package name: {name!r}") try: result = self._post(f"/api/core/firmware/install/{name}") @@ -2347,7 +2374,8 @@ class OPNsenseDriver( a decision of its own rather than a detail of this one. """ 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 package name: {name!r}") if not name.startswith("os-"): raise NotImplementedError( @@ -2464,57 +2492,72 @@ class OPNsenseDriver( def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: community = settings["community"] - if not community or any(c.isspace() or c in "\"#" for c in community): + if not community or any(c.isspace() or c in '"#' for c in community): raise ValueError("the community must be one word without quotes or #") listen = settings.get("listen") or self.hostname try: listen_val = {str(ipaddress.ip_address(listen)): {"selected": 1}} except ValueError: listen_val = {} # a DNS name binds nothing; snmpd then listens everywhere - result = self._post("/api/netsnmp/general/set", { - "general": { - "enabled": "1", - "community": community, - "contact": settings.get("contact", ""), - "location": settings.get("location", ""), - "sysobjid": "", - "bindip": "", - "listen": listen_val, - } - }) + result = self._post( + "/api/netsnmp/general/set", + { + "general": { + "enabled": "1", + "community": community, + "contact": settings.get("contact", ""), + "location": settings.get("location", ""), + "sysobjid": "", + "bindip": "", + "listen": listen_val, + } + }, + ) if result.get("result") != "saved": return {"ok": False, "output": f"configuration refused: {result}"} - return {"ok": True, "output": f"community set, listen {list(listen_val) or 'all addresses'}"} + return { + "ok": True, + "output": f"community set, listen {list(listen_val) or 'all addresses'}", + } def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: - sources = [str(ipaddress.ip_network(s.strip(), strict=False)) for s in settings.get("allowed_sources", [])] + sources = [ + str(ipaddress.ip_network(s.strip(), strict=False)) + for s in settings.get("allowed_sources", []) + ] if not sources: return {"ok": True, "output": "no sources given; firewall unchanged"} - found = self._post("/api/firewall/filter/searchRule", {"current": 1, "rowCount": -1, "searchPhrase": "SNMP"}) + found = self._post( + "/api/firewall/filter/searchRule", + {"current": 1, "rowCount": -1, "searchPhrase": "SNMP"}, + ) for row in found.get("rows") or []: descr = str(row.get("description") or row.get("descr") or "") if descr == self._SNMP_RULE_OLD or descr.startswith(self._SNMP_RULE_PREFIX): self._post(f"/api/firewall/filter/delRule/{row['uuid']}") for source in sources: - self._post("/api/firewall/filter/addRule", { - "rule": { - "enabled": "1", - "action": "pass", - "quick": "1", - "interface": "", # floating: wherever the source comes in - "direction": "in", - "ipprotocol": "inet6" if ":" in source else "inet", - "protocol": "udp", - "source_net": source, - "source_not": "0", - "destination_net": "(self)", - "destination_not": "0", - "destination_port": "161", - "log": "0", - "floating": "yes", - "descr": f"{self._SNMP_RULE_PREFIX}{source}", - } - }) + self._post( + "/api/firewall/filter/addRule", + { + "rule": { + "enabled": "1", + "action": "pass", + "quick": "1", + "interface": "", # floating: wherever the source comes in + "direction": "in", + "ipprotocol": "inet6" if ":" in source else "inet", + "protocol": "udp", + "source_net": source, + "source_not": "0", + "destination_net": "(self)", + "destination_not": "0", + "destination_port": "161", + "log": "0", + "floating": "yes", + "descr": f"{self._SNMP_RULE_PREFIX}{source}", + } + }, + ) self._post("/api/firewall/filter/apply", {}) return {"ok": True, "output": f"udp/161 open for {', '.join(sources)} only"} @@ -2560,14 +2603,16 @@ class OPNsenseDriver( content_raw = row.get("content", "") or "" # OPNsense stores content as newline-separated values content = [v.strip() for v in content_raw.splitlines() if v.strip()] - result.append({ - "name": row.get("name", ""), - "type": row.get("type", ""), - "description": row.get("description", "") or "", - "content": content, - "enabled": str(row.get("enabled", "1")) == "1", - "proto": row.get("proto", "") or "", - }) + result.append( + { + "name": row.get("name", ""), + "type": row.get("type", ""), + "description": row.get("description", "") or "", + "content": content, + "enabled": str(row.get("enabled", "1")) == "1", + "proto": row.get("proto", "") or "", + } + ) return sorted(result, key=lambda x: (x["type"], x["name"].lower())) @@ -2596,7 +2641,7 @@ class OPNsenseDriver( group_names: set = set() try: grp = self._get("/api/ifgroups/ifgroups/searchItem?current=1&rowCount=-1") - for g in (grp.get("rows") or []): + for g in grp.get("rows") or []: n = g.get("ifname") or g.get("name") or "" if n: group_names.add(n) @@ -2612,8 +2657,7 @@ class OPNsenseDriver( iface_label = (row.get("%interface") or "").strip() or ", ".join(iface_list) explicit_floating = ( - str(row.get("floating", "0")) in ("1", "yes") - or row.get("floating") is True + str(row.get("floating", "0")) in ("1", "yes") or row.get("floating") is True ) floating = explicit_floating or len(iface_list) > 1 @@ -2621,37 +2665,45 @@ class OPNsenseDriver( category = (row.get("%categories") or row.get("category") or "").strip() # Use %-prefixed display values for source/destination - source_net = (row.get("%source_net") or row.get("source_net") or "any").strip() - destination_net = (row.get("%destination_net") or row.get("destination_net") or "any").strip() + source_net = (row.get("%source_net") or row.get("source_net") or "any").strip() + destination_net = ( + row.get("%destination_net") or row.get("destination_net") or "any" + ).strip() - result.append({ - "uuid": row.get("uuid", ""), - "sequence": int(row.get("sequence", 0) or 0), - "action": row.get("action", "pass"), - "quick": str(row.get("quick", "1")) == "1", - "interface": iface, - "interface_label": iface_label, - "floating": floating, - "is_group": len(iface_list) == 1 and iface_list[0] in group_names, - "direction": row.get("direction", "in"), - "ipprotocol": row.get("ipprotocol", "inet"), - "protocol": row.get("protocol", "any") or "any", - "source_net": source_net, - "source_port": row.get("source_port", "") or "", - "source_not": str(row.get("source_not", "0")) == "1", - "destination_net": destination_net, - "destination_port": row.get("destination_port", "") or "", - "destination_not": str(row.get("destination_not", "0")) == "1", - "description": row.get("description", "") or "", - "log": str(row.get("log", "0")) == "1", - "enabled": str(row.get("enabled", "1")) == "1", - "category": category, - "gateway": row.get("gateway", "") or "", - }) + result.append( + { + "uuid": row.get("uuid", ""), + "sequence": int(row.get("sequence", 0) or 0), + "action": row.get("action", "pass"), + "quick": str(row.get("quick", "1")) == "1", + "interface": iface, + "interface_label": iface_label, + "floating": floating, + "is_group": len(iface_list) == 1 and iface_list[0] in group_names, + "direction": row.get("direction", "in"), + "ipprotocol": row.get("ipprotocol", "inet"), + "protocol": row.get("protocol", "any") or "any", + "source_net": source_net, + "source_port": row.get("source_port", "") or "", + "source_not": str(row.get("source_not", "0")) == "1", + "destination_net": destination_net, + "destination_port": row.get("destination_port", "") or "", + "destination_not": str(row.get("destination_not", "0")) == "1", + "description": row.get("description", "") or "", + "log": str(row.get("log", "0")) == "1", + "enabled": str(row.get("enabled", "1")) == "1", + "category": category, + "gateway": row.get("gateway", "") or "", + } + ) - return sorted(result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"])) + return sorted( + result, key=lambda x: (x["floating"], x["is_group"], x["interface"], x["sequence"]) + ) - def apply_firewall_rule(self, rule: dict[str, Any], *, uuid: str | None = None) -> dict[str, Any]: + def apply_firewall_rule( + self, rule: dict[str, Any], *, uuid: str | None = None + ) -> dict[str, Any]: """Create or update a single OPNsense firewall filter rule. `rule` uses the vendor-neutral field names from @@ -2757,7 +2809,7 @@ class OPNsenseDriver( # Fetch the list of internal CAs to find the right one ca_data = self._post("/api/trust/ca/search", {}) internal_ca_uuid = None - for row in (ca_data.get("rows") or []): + for row in ca_data.get("rows") or []: if row.get("internal") == "1" or row.get("catype") == "internal": internal_ca_uuid = row.get("uuid") break @@ -2865,18 +2917,20 @@ class OPNsenseDriver( seen.add(key) # The auto-PTR flag is "addptr" since 25.x, "ptrrecord" before. ptr = row.get("addptr", row.get("ptrrecord", "1")) - result.append({ - "uuid": row.get("uuid", ""), - "hostname": host, - "domain": domain, - "fqdn": fqdn, - "ip": ip, - "record_type": rr, - "description": row.get("description", "") or "", - "enabled": str(row.get("enabled", "1")) == "1", - "ptrrecord": str(ptr) == "1", - "service": "unbound", - }) + result.append( + { + "uuid": row.get("uuid", ""), + "hostname": host, + "domain": domain, + "fqdn": fqdn, + "ip": ip, + "record_type": rr, + "description": row.get("description", "") or "", + "enabled": str(row.get("enabled", "1")) == "1", + "ptrrecord": str(ptr) == "1", + "service": "unbound", + } + ) except Exception as exc: logger.warning("Failed to fetch Unbound host overrides: %s", exc) return sorted(result, key=lambda e: e["fqdn"].lower()) @@ -2892,17 +2946,19 @@ class OPNsenseDriver( host = row.get("host", "") or "" domain = row.get("domain", "") or "" fqdn = f"{host}.{domain}" if host and domain else host or domain - result.append({ - "uuid": row.get("uuid", ""), - "hostname": host, - "domain": domain, - "fqdn": fqdn, - "ip": row.get("ip", "") or "", - "record_type": "A", - "description": row.get("description", "") or "", - "enabled": str(row.get("enabled", "1")) == "1", - "service": "dnsmasq", - }) + result.append( + { + "uuid": row.get("uuid", ""), + "hostname": host, + "domain": domain, + "fqdn": fqdn, + "ip": row.get("ip", "") or "", + "record_type": "A", + "description": row.get("description", "") or "", + "enabled": str(row.get("enabled", "1")) == "1", + "service": "dnsmasq", + } + ) except Exception as exc: logger.warning("Failed to fetch Dnsmasq host overrides: %s", exc) return sorted(result, key=lambda e: e["fqdn"].lower()) @@ -2966,19 +3022,28 @@ class OPNsenseDriver( zone_clean = zone_name.rstrip(".") # Remove existing netork-managed overrides for this zone - existing = (self._get_unbound_host_overrides() if service == "unbound" - else self._get_dnsmasq_host_overrides()) + existing = ( + self._get_unbound_host_overrides() + if service == "unbound" + else self._get_dnsmasq_host_overrides() + ) for entry in existing: - if (entry["domain"].rstrip(".") == zone_clean - and entry["description"].startswith(self._NETORK_TAG)): + if entry["domain"].rstrip(".") == zone_clean and entry["description"].startswith( + self._NETORK_TAG + ): uid = entry["uuid"] if uid: if service == "unbound": self._post(f"/api/unbound/settings/delhostoverride/{uid}") else: self._post(f"/api/dnsmasq/settings/delhostoverride/{uid}") - logger.info("Removed %s host override %s.%s (uuid %s)", - service, entry["hostname"], zone_clean, uid) + logger.info( + "Removed %s host override %s.%s (uuid %s)", + service, + entry["hostname"], + zone_clean, + uid, + ) # Re-add all enabled A/AAAA records, one override per logical record. # Two rows for the same name is a state netOrk's own database can @@ -3005,33 +3070,44 @@ class OPNsenseDriver( continue pushed.add(key) if service == "unbound": - self._post("/api/unbound/settings/addhostoverride", { - "host": { - "enabled": "1", - "hostname": rec.get("hostname", ""), - "domain": zone_clean, - "rr": rtype, - "server": rec.get("ip", ""), - "description": self._NETORK_TAG, - "addptr": "1", - "ptrrecord": "1", - "mxprio": "", - "mx": "", - } - }) + self._post( + "/api/unbound/settings/addhostoverride", + { + "host": { + "enabled": "1", + "hostname": rec.get("hostname", ""), + "domain": zone_clean, + "rr": rtype, + "server": rec.get("ip", ""), + "description": self._NETORK_TAG, + "addptr": "1", + "ptrrecord": "1", + "mxprio": "", + "mx": "", + } + }, + ) else: - self._post("/api/dnsmasq/settings/addhostoverride", { - "hostoverride": { - "enabled": "1", - "host": rec.get("hostname", ""), - "domain": zone_clean, - "ip": rec.get("ip", ""), - "description": self._NETORK_TAG, - } - }) + self._post( + "/api/dnsmasq/settings/addhostoverride", + { + "hostoverride": { + "enabled": "1", + "host": rec.get("hostname", ""), + "domain": zone_clean, + "ip": rec.get("ip", ""), + "description": self._NETORK_TAG, + } + }, + ) added += 1 - logger.info("Added %s host override %s.%s → %s", - service, rec.get("hostname"), zone_clean, rec.get("ip")) + logger.info( + "Added %s host override %s.%s → %s", + service, + rec.get("hostname"), + zone_clean, + rec.get("ip"), + ) if service == "unbound": self._post("/api/unbound/service/reconfigure") diff --git a/napalm_opnsense/ospf_mixin.py b/napalm_opnsense/ospf_mixin.py index c2f7195..be83080 100644 --- a/napalm_opnsense/ospf_mixin.py +++ b/napalm_opnsense/ospf_mixin.py @@ -33,7 +33,9 @@ class OPNsenseOspfMixin(OspfMixin): def get_ospf(self) -> dict[str, Any]: """Router ID, areas and OSPF per interface; empty when OSPF does not run.""" overview = self._get(f"{_DIAG}/ospfoverview").get("response") or {} # type: ignore[attr-defined] - interfaces = (self._get(f"{_DIAG}/ospfinterface").get("response") or {}).get("interfaces") or {} # type: ignore[attr-defined] + interfaces = (self._get(f"{_DIAG}/ospfinterface").get("response") or {}).get( + "interfaces" + ) or {} # type: ignore[attr-defined] return { "router_id": overview.get("routerId") or None, "areas": list((overview.get("areas") or {}).keys()), diff --git a/napalm_opnsense/ping_mixin.py b/napalm_opnsense/ping_mixin.py index 72bf555..74b08d3 100644 --- a/napalm_opnsense/ping_mixin.py +++ b/napalm_opnsense/ping_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 @@ -45,8 +44,9 @@ from __future__ import annotations import logging import time +from collections.abc import Callable, Iterable from ipaddress import ip_address -from typing import Any, Callable, Iterable, Optional +from typing import Any logger = logging.getLogger(__name__) @@ -78,7 +78,7 @@ class OPNsensePingMixin: PING_SWEEP_MAX_TARGETS: int = 512 # Provided by the driver this mixin is mixed into. - _ping_model_path_cache: Optional[tuple[str, ...]] = None + _ping_model_path_cache: tuple[str, ...] | None = None # ── NAPALM ─────────────────────────────────────────────────────────────── @@ -119,9 +119,9 @@ class OPNsensePingMixin: *, count: int = 1, timeout: int = 1, - max_targets: Optional[int] = None, - on_progress: Optional[Callable[[int, int], None]] = None, - should_stop: Optional[Callable[[], bool]] = None, + max_targets: int | None = None, + on_progress: Callable[[int, int], None] | None = None, + should_stop: Callable[[], bool] | None = None, ) -> dict[str, Any]: """Ping many destinations, one batch of parallel firewall jobs at a time. @@ -245,9 +245,7 @@ class OPNsensePingMixin: if attempt: time.sleep(_POLL_INTERVAL) rows = { - job_id: row - for job_id, row in self._read_ping_jobs().items() - if job_id in wanted + job_id: row for job_id, row in self._read_ping_jobs().items() if job_id in wanted } if wanted and all( _as_int(rows.get(job_id, {}).get("send")) >= count for job_id in wanted @@ -325,7 +323,7 @@ class OPNsensePingMixin: @staticmethod def _row_to_napalm_reply( - destination: str, row: Optional[dict[str, Any]], *, count: int + destination: str, row: dict[str, Any] | None, *, count: int ) -> dict[str, Any]: """Turn one ``search_jobs`` row into NAPALM's ping reply shape. diff --git a/napalm_opnsense/port_forwards.py b/napalm_opnsense/port_forwards.py index 635f41c..14d3a87 100644 --- a/napalm_opnsense/port_forwards.py +++ b/napalm_opnsense/port_forwards.py @@ -18,11 +18,12 @@ from __future__ import annotations import ipaddress import socket -from typing import Any, Dict, Iterable, List, Optional, Tuple +from collections.abc import Iterable +from typing import Any #: Port names OPNsense accepts in a rule. ``socket.getservbyname`` knows them #: too, but only where ``/etc/services`` exists -- a slim container has none. -WELL_KNOWN_PORTS: Dict[str, int] = { +WELL_KNOWN_PORTS: dict[str, int] = { "ftp": 21, "ssh": 22, "telnet": 23, @@ -49,7 +50,7 @@ WELL_KNOWN_PORTS: Dict[str, int] = { #: Alias types whose entries can be addresses. _ADDRESS_ALIASES = ("host", "network") -AliasIndex = Dict[str, Tuple[str, List[str]]] +AliasIndex = dict[str, tuple[str, list[str]]] def wan_interfaces(overview: Any) -> set: @@ -61,20 +62,23 @@ def wan_interfaces(overview: Any) -> set: def alias_index(rows: Iterable[dict]) -> AliasIndex: """``name -> (type, entries)`` from the alias list.""" return { - r["name"]: (r.get("type", ""), [e.strip() for e in str(r.get("content", "")).splitlines() if e.strip()]) + r["name"]: ( + r.get("type", ""), + [e.strip() for e in str(r.get("content", "")).splitlines() if e.strip()], + ) for r in rows if r.get("name") } -def _as_address(value: str) -> Optional[str]: +def _as_address(value: str) -> str | None: try: return str(ipaddress.ip_address(value)) except ValueError: return None -def _addresses(target: str, aliases: AliasIndex) -> List[str]: +def _addresses(target: str, aliases: AliasIndex) -> list[str]: """The addresses a rule sends to: a literal, or a host/network alias's. Anything else -- an interface address, a name resolved by DNS -- gives no @@ -89,7 +93,7 @@ def _addresses(target: str, aliases: AliasIndex) -> List[str]: return [a for a in (_as_address(e) for e in entries) if a] -def _port(value: Any, protocol: str, aliases: AliasIndex) -> Optional[int]: +def _port(value: Any, protocol: str, aliases: AliasIndex) -> int | None: """A rule's port as a number: literal, start of a range, alias or name. No port at all means every port, which the contract writes as 0. @@ -115,7 +119,7 @@ def _faces_the_wan(rule: dict, wan: set) -> bool: return bool(set(str(rule.get("interface", "")).split(",")) & wan) -def _remote_host(rule: dict) -> Optional[str]: +def _remote_host(rule: dict) -> str | None: """A source restriction; an inverted one ("all but X") restricts nothing.""" source = str(rule.get("source.network") or "").strip() if source in ("", "any") or rule.get("source.not") == "1": @@ -123,11 +127,11 @@ def _remote_host(rule: dict) -> Optional[str]: return source -def _forwards_of(rule: dict, aliases: AliasIndex) -> List[dict]: +def _forwards_of(rule: dict, aliases: AliasIndex) -> list[dict]: protocols = [p.upper() for p in str(rule.get("protocol") or "any").split("/") if p] target = str(rule.get("target", "")).strip() remote = _remote_host(rule) - result: List[dict] = [] + result: list[dict] = [] for protocol in protocols: external = _port(rule.get("destination.port", ""), protocol.lower(), aliases) if external is None: @@ -150,9 +154,9 @@ def _forwards_of(rule: dict, aliases: AliasIndex) -> List[dict]: return result -def port_forwards(rules: Iterable[dict], wan: set, aliases: AliasIndex) -> List[dict]: +def port_forwards(rules: Iterable[dict], wan: set, aliases: AliasIndex) -> list[dict]: """The destination-NAT rules on a WAN interface, as ``PortForwardDict`` entries.""" - result: List[dict] = [] + result: list[dict] = [] for rule in rules: if _faces_the_wan(rule, wan): result.extend(_forwards_of(rule, aliases)) diff --git a/napalm_opnsense/resolver_mixin.py b/napalm_opnsense/resolver_mixin.py index 0d044e8..4d59d68 100644 --- a/napalm_opnsense/resolver_mixin.py +++ b/napalm_opnsense/resolver_mixin.py @@ -45,7 +45,13 @@ class OPNsenseResolverMixin(DnsForwardMixin): def set_dns_forward(self, domain: str, server: str, port: int = 53) -> None: """Forward *domain* to *server*; an existing forward of the domain is changed.""" name = _domain(domain) - row = {"enabled": "1", "type": "forward", "domain": name, "server": server, "port": str(port)} + row = { + "enabled": "1", + "type": "forward", + "domain": name, + "server": server, + "port": str(port), + } known = [r for r in self._forward_rows() if _domain(r.get("domain", "")) == name] if known: result = self._post(f"{_SETTINGS}/setForward/{known[0]['uuid']}", {"dot": row}) # type: ignore[attr-defined] @@ -57,7 +63,11 @@ class OPNsenseResolverMixin(DnsForwardMixin): def remove_dns_forward(self, domain: str) -> None: """Delete every forward of *domain*; nothing to do when there is none.""" name = _domain(domain) - uuids = [r["uuid"] for r in self._forward_rows() if _domain(r.get("domain", "")) == name and r.get("uuid")] + uuids = [ + r["uuid"] + for r in self._forward_rows() + if _domain(r.get("domain", "")) == name and r.get("uuid") + ] if not uuids: return for uuid in uuids: diff --git a/napalm_opnsense/virtual_ips.py b/napalm_opnsense/virtual_ips.py index ea692b4..79402b7 100644 --- a/napalm_opnsense/virtual_ips.py +++ b/napalm_opnsense/virtual_ips.py @@ -18,7 +18,7 @@ shape of the reading. from __future__ import annotations import ipaddress -from typing import Any, Optional +from typing import Any from napalm_device_types.models import VirtualIpDict @@ -31,7 +31,7 @@ _CARP_STATES: dict[str, tuple[bool, bool]] = { } -def _address(text: str) -> Optional[tuple[str, Optional[int]]]: +def _address(text: str) -> tuple[str, int | None] | None: """``"10.0.0.1/24"`` -> ``("10.0.0.1", 24)``; None for what is no address.""" host, slash, prefix = text.partition("/") try: @@ -48,12 +48,12 @@ def _address(text: str) -> Optional[tuple[str, Optional[int]]]: def carp_virtual_ip( *, address: str, - prefix_length: Optional[int], + prefix_length: int | None, vhid: object, - state: Optional[str], + state: str | None, instance: str, - interface: Optional[str], -) -> Optional[VirtualIpDict]: + interface: str | None, +) -> VirtualIpDict | None: """One address bound to a CARP VHID, as a virtual IP; None for what is no address. Who holds it comes from CARP's *state* on this member, not from its diff --git a/pyproject.toml b/pyproject.toml index 3ea938a..2cc1d7a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,14 @@ include = ["napalm_opnsense*"] [tool.ruff] line-length = 100 -target-version = "py39" +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.pytest.ini_options] testpaths = ["tests"] diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index e6b6c00..2efbfc5 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1,18 +1,19 @@ """Unit tests for OPNsenseDriver — no real device required.""" import json -import pytest -import requests from unittest.mock import MagicMock, patch -from napalm_opnsense.opnsense import OPNsenseDriver -from napalm.base.exceptions import ConnectionException, ConnectionClosedException +import pytest +import requests +from napalm.base.exceptions import ConnectionClosedException, ConnectionException +from napalm_opnsense.opnsense import OPNsenseDriver # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- + @pytest.fixture def driver(): """Return a driver instance with a mocked requests.Session.""" @@ -110,6 +111,7 @@ ARP_RESPONSE = { # Helpers # --------------------------------------------------------------------------- + def _make_json_response(data): mock_resp = MagicMock() mock_resp.json.return_value = data @@ -121,6 +123,7 @@ def _make_json_response(data): # open() / close() / is_alive() # --------------------------------------------------------------------------- + class TestOpenClose: def test_open_raises_connection_exception_on_error(self): drv = OPNsenseDriver( @@ -166,6 +169,7 @@ class TestIsAlive: # _get() # --------------------------------------------------------------------------- + class TestInternalGet: def test_raises_when_no_session(self): drv = OPNsenseDriver("host", "u", "p") @@ -185,41 +189,40 @@ class TestInternalGet: # get_facts() # --------------------------------------------------------------------------- + class TestGetFacts: def test_returns_required_keys(self, driver): - driver._get = lambda path: ( - STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE - ) + driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE facts = driver.get_facts() - for key in ("vendor", "model", "hostname", "fqdn", "os_version", - "serial_number", "uptime", "interface_list"): + for key in ( + "vendor", + "model", + "hostname", + "fqdn", + "os_version", + "serial_number", + "uptime", + "interface_list", + ): assert key in facts def test_vendor_constant(self, driver): - driver._get = lambda path: ( - STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE - ) + driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE facts = driver.get_facts() assert facts["vendor"] == "OPNsense" def test_hostname_parsed(self, driver): - driver._get = lambda path: ( - STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE - ) + driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE facts = driver.get_facts() assert facts["hostname"] == "opnsense01" def test_os_version_parsed(self, driver): - driver._get = lambda path: ( - STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE - ) + driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE facts = driver.get_facts() assert facts["os_version"] == "24.7" def test_interface_list_populated(self, driver): - driver._get = lambda path: ( - STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE - ) + driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE facts = driver.get_facts() assert "em0" in facts["interface_list"] assert "em1" in facts["interface_list"] @@ -235,16 +238,12 @@ class TestGetFacts: assert facts["interface_list"] == [] def test_serial_number(self, driver): - driver._get = lambda path: ( - STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE - ) + driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE facts = driver.get_facts() assert facts["serial_number"] == "ABC123" def test_uptime(self, driver): - driver._get = lambda path: ( - STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE - ) + driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE facts = driver.get_facts() assert facts["uptime"] == 12345 @@ -253,6 +252,7 @@ class TestGetFacts: # get_interfaces() # --------------------------------------------------------------------------- + class TestGetInterfaces: def test_interface_count(self, driver): driver._get = lambda path: INTERFACES_RESPONSE @@ -311,6 +311,7 @@ class TestGetInterfaces: # get_interfaces_ip() # --------------------------------------------------------------------------- + class TestGetInterfacesIp: def test_entry_count(self, driver): driver._get = lambda path: ADDRESSES_RESPONSE @@ -340,6 +341,7 @@ class TestGetInterfacesIp: # get_arp_table() # --------------------------------------------------------------------------- + class TestGetArpTable: def test_entry_count(self, driver): driver._get = lambda path: ARP_RESPONSE @@ -419,11 +421,18 @@ class TestGetInterfacesCounters: driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE entry = driver.get_interfaces_counters()["em0"] for key in ( - "tx_errors", "rx_errors", "tx_discards", "rx_discards", - "tx_octets", "rx_octets", - "tx_unicast_packets", "rx_unicast_packets", - "tx_multicast_packets", "rx_multicast_packets", - "tx_broadcast_packets", "rx_broadcast_packets", + "tx_errors", + "rx_errors", + "tx_discards", + "rx_discards", + "tx_octets", + "rx_octets", + "tx_unicast_packets", + "rx_unicast_packets", + "tx_multicast_packets", + "rx_multicast_packets", + "tx_broadcast_packets", + "rx_broadcast_packets", ): assert key in entry @@ -505,7 +514,8 @@ class TestGetEnvironment: def test_temperature_alert_thresholds(self, driver): driver._get = lambda path: ( {"data": [{"device": "cpu0", "temperature": "85.0"}]} - if "temperature" in path else SYSTEM_RESOURCES_RESPONSE + if "temperature" in path + else SYSTEM_RESOURCES_RESPONSE ) env = driver.get_environment() assert env["temperature"]["cpu0"]["is_alert"] is True @@ -513,6 +523,7 @@ class TestGetEnvironment: def test_temperature_endpoint_failing_gracefully(self, driver): """Driver must not raise if temperature endpoint is unavailable.""" + def fake_get(path): if "temperature" in path: raise Exception("no sensor data") @@ -594,19 +605,22 @@ class TestGetRouteTo: def test_filter_by_protocol(self, driver): driver._get = lambda path: ROUTES_RESPONSE routes = driver.get_route_to(protocol="static") - assert all( - e["protocol"] == "static" - for entries in routes.values() - for e in entries - ) + assert all(e["protocol"] == "static" for entries in routes.values() for e in entries) def test_required_keys_in_entry(self, driver): driver._get = lambda path: ROUTES_RESPONSE entry = driver.get_route_to()["0.0.0.0/0"][0] for key in ( - "protocol", "current_active", "last_active", "age", - "next_hop", "outgoing_interface", "selected_next_hop", - "preference", "inactive_reason", "routing_table", + "protocol", + "current_active", + "last_active", + "age", + "next_hop", + "outgoing_interface", + "selected_next_hop", + "preference", + "inactive_reason", + "routing_table", "protocol_attributes", ): assert key in entry @@ -707,15 +721,21 @@ class TestGetLldpNeighborsDetail: detail = driver.get_lldp_neighbors_detail() entry = detail["em0"][0] for key in ( - "remote_chassis_id", "remote_system_name", "remote_port", - "remote_port_description", "remote_system_description", - "remote_system_capab", "remote_system_enable_capab", + "remote_chassis_id", + "remote_system_name", + "remote_port", + "remote_port_description", + "remote_system_description", + "remote_system_capab", + "remote_system_enable_capab", ): assert key in entry def test_chassis_id(self, driver): driver._get = lambda path: LLDP_RESPONSE - assert driver.get_lldp_neighbors_detail()["em0"][0]["remote_chassis_id"] == "aa:bb:cc:dd:ee:ff" + assert ( + driver.get_lldp_neighbors_detail()["em0"][0]["remote_chassis_id"] == "aa:bb:cc:dd:ee:ff" + ) def test_system_name(self, driver): driver._get = lambda path: LLDP_RESPONSE @@ -1010,12 +1030,19 @@ class TestGetBgpNeighbors: def test_required_peer_keys(self, driver): driver._get = self._fake_get peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"] - for key in ("local_as", "remote_as", "remote_id", "is_up", "is_enabled", - "description", "uptime", "address_family"): + for key in ( + "local_as", + "remote_as", + "remote_id", + "is_up", + "is_enabled", + "description", + "uptime", + "address_family", + ): assert key in peer - # --------------------------------------------------------------------------- # get_certificates() # @@ -1180,15 +1207,24 @@ class TestGetDdnsStatus: RADIUS_CLIENTS_RESPONSE = { "rows": [ - {"uuid": "e5483f1b-936b-47ba-8cca-24d56ad643c6", "enabled": "1", - "name": "ap-lobby", "secret": "s3cr3t", "ip": "10.0.0.5/32"}, + { + "uuid": "e5483f1b-936b-47ba-8cca-24d56ad643c6", + "enabled": "1", + "name": "ap-lobby", + "secret": "s3cr3t", + "ip": "10.0.0.5/32", + }, ] } RADIUS_USERS_RESPONSE = { "rows": [ - {"uuid": "71da23fd-03f3-4e1f-a7e1-bb25645b981e", "enabled": "1", - "username": "jdoe", "password": "hunter2"}, + { + "uuid": "71da23fd-03f3-4e1f-a7e1-bb25645b981e", + "enabled": "1", + "username": "jdoe", + "password": "hunter2", + }, ] } @@ -1237,10 +1273,13 @@ class TestCreateRadiusClient: def test_validation_failure_does_not_reconfigure(self, driver): calls = [] - driver._post = lambda path, data=None: calls.append(path) or { - "result": "failed", - "validations": {"client.name": "A value is required."}, - } + driver._post = lambda path, data=None: ( + calls.append(path) + or { + "result": "failed", + "validations": {"client.name": "A value is required."}, + } + ) result = driver.create_radius_client("", "10.0.0.5/32", "s3cr3t") assert result["success"] is False assert result["validations"] == {"client.name": "A value is required."} @@ -1320,10 +1359,13 @@ class TestCreateRadiusUser: def test_validation_failure_does_not_reconfigure(self, driver): calls = [] - driver._post = lambda path, data=None: calls.append(path) or { - "result": "failed", - "validations": {"user.username": "A value is required."}, - } + driver._post = lambda path, data=None: ( + calls.append(path) + or { + "result": "failed", + "validations": {"user.username": "A value is required."}, + } + ) result = driver.create_radius_user("", "hunter2") assert result["success"] is False assert calls == ["/api/freeradius/user/add_user"] @@ -1358,6 +1400,7 @@ class TestDeleteRadiusUser: # _post() # --------------------------------------------------------------------------- + class TestInternalPost: def test_raises_when_no_session(self): drv = OPNsenseDriver("host", "u", "p") @@ -1393,6 +1436,7 @@ class TestInternalPost: # send_wake_on_lan() # --------------------------------------------------------------------------- + class TestSendWakeOnLan: def test_raises_value_error_without_interface(self, driver): with pytest.raises(ValueError): @@ -1413,7 +1457,10 @@ class TestSendWakeOnLan: # success — the underlying `wol` CLI tool typically prints nothing. driver.session.post.return_value = _make_json_response({"status": ""}) result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan") - assert result == {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"} + assert result == { + "success": True, + "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan", + } def test_returns_failure_on_error_status(self, driver): driver.session.post.return_value = _make_json_response( @@ -1441,10 +1488,12 @@ class TestSendWakeOnLan: # load_merge_candidate() # --------------------------------------------------------------------------- -VALID_ROUTES_CONFIG = json.dumps([ - {"network": "10.0.0.0/8", "gateway": "WAN_GW", "descr": "internal"}, - {"network": "0.0.0.0/0", "gateway": "WAN_GW", "descr": "default"}, -]) +VALID_ROUTES_CONFIG = json.dumps( + [ + {"network": "10.0.0.0/8", "gateway": "WAN_GW", "descr": "internal"}, + {"network": "0.0.0.0/0", "gateway": "WAN_GW", "descr": "default"}, + ] +) class TestLoadMergeCandidate: @@ -1467,38 +1516,45 @@ class TestLoadMergeCandidate: def test_raises_if_both_args_given(self, driver): from napalm.base.exceptions import MergeConfigException + with pytest.raises(MergeConfigException): driver.load_merge_candidate(filename="f.json", config="{}") def test_raises_if_no_args_given(self, driver): from napalm.base.exceptions import MergeConfigException + with pytest.raises(MergeConfigException): driver.load_merge_candidate() def test_raises_on_invalid_json(self, driver): from napalm.base.exceptions import MergeConfigException + with pytest.raises(MergeConfigException, match="Invalid JSON"): driver.load_merge_candidate(config="not json {{{") def test_raises_if_not_a_list(self, driver): from napalm.base.exceptions import MergeConfigException + with pytest.raises(MergeConfigException, match="array"): driver.load_merge_candidate(config='{"network": "1.0.0.0/8"}') def test_raises_if_route_missing_network(self, driver): from napalm.base.exceptions import MergeConfigException + bad = json.dumps([{"gateway": "GW1"}]) with pytest.raises(MergeConfigException, match="'network'"): driver.load_merge_candidate(config=bad) def test_raises_if_route_missing_gateway(self, driver): from napalm.base.exceptions import MergeConfigException + bad = json.dumps([{"network": "10.0.0.0/8"}]) with pytest.raises(MergeConfigException, match="'gateway'"): driver.load_merge_candidate(config=bad) def test_raises_on_missing_file(self, driver): from napalm.base.exceptions import MergeConfigException + with pytest.raises(MergeConfigException, match="Cannot read"): driver.load_merge_candidate(filename="/nonexistent/path/routes.json") @@ -1532,9 +1588,9 @@ class TestCompareConfig: assert "WAN_GW" in diff def test_empty_diff_when_config_matches(self, driver): - same_config = json.dumps([ - {"network": "192.168.1.0/24", "gateway": "LAN_GW", "descr": "lan", "disabled": "0"} - ]) + same_config = json.dumps( + [{"network": "192.168.1.0/24", "gateway": "LAN_GW", "descr": "lan", "disabled": "0"}] + ) driver._get = lambda path: SEARCH_ROUTE_RESPONSE driver.load_merge_candidate(config=same_config) diff = driver.compare_config() @@ -1547,7 +1603,11 @@ class TestCompareConfig: BACKUPS_RESPONSE = { "items": [ - {"id": "config-opnsense01-1234567890.xml", "time": "1234567890", "description": "before change"}, + { + "id": "config-opnsense01-1234567890.xml", + "time": "1234567890", + "description": "before change", + }, {"id": "config-opnsense01-1234567800.xml", "time": "1234567800", "description": "initial"}, ] } @@ -1556,6 +1616,7 @@ BACKUPS_RESPONSE = { class TestCommitConfig: def test_raises_without_candidate(self, driver): from napalm.base.exceptions import MergeConfigException + with pytest.raises(MergeConfigException, match="No candidate"): driver.commit_config() @@ -1589,8 +1650,9 @@ class TestCommitConfig: assert driver._pre_commit_backup_id is None def test_raises_on_api_error(self, driver): - from napalm.base.exceptions import MergeConfigException import requests as _req + from napalm.base.exceptions import MergeConfigException + driver._get = lambda path: BACKUPS_RESPONSE driver.load_merge_candidate(config=VALID_ROUTES_CONFIG) driver.session.post.side_effect = _req.exceptions.RequestException("timeout") @@ -1602,6 +1664,7 @@ class TestCommitConfig: # discard_config() # --------------------------------------------------------------------------- + class TestDiscardConfig: def test_clears_candidate(self, driver): driver.load_merge_candidate(config=VALID_ROUTES_CONFIG) @@ -1617,6 +1680,7 @@ class TestDiscardConfig: # rollback() # --------------------------------------------------------------------------- + class TestRollback: def test_noop_when_no_backups_and_no_commit(self, driver): driver._get = lambda path: {"items": []} @@ -1655,6 +1719,7 @@ class TestRollback: # get_config() — candidate slot # --------------------------------------------------------------------------- + class TestGetConfigCandidate: def test_candidate_empty_without_staged_config(self, driver): driver._get_text = lambda path: "" @@ -1683,7 +1748,9 @@ class TestGetConfigCandidate: # the getter called .json() on XML and always returned an empty "running". # --------------------------------------------------------------------------- -CONFIG_XML = '\ngw\n' +CONFIG_XML = ( + '\ngw\n' +) def _make_xml_response(text): @@ -1767,7 +1834,9 @@ class TestCreateDhcpReservation: assert payload["description"] == "[netork]" reconfigure_call = driver.session.post.call_args_list[2] - assert reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure" + assert ( + reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure" + ) def test_updates_existing_reservation_for_same_ip(self, driver): driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE) @@ -1934,7 +2003,9 @@ class TestDeleteDhcpReservationAndLease: == "https://opnsense.example.com/api/kea/dhcpv4/delReservation/existing-uuid-456" ) reconfigure_call = driver.session.post.call_args_list[2] - assert reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure" + assert ( + reconfigure_call.args[0] == "https://opnsense.example.com/api/kea/service/reconfigure" + ) del_lease_call = driver.session.post.call_args_list[3] # IP is a URL path segment, not a JSON body field — verified live # against a real OPNsense instance; a {"ips": [ip]} body (the @@ -2006,6 +2077,7 @@ class TestDeleteDhcpReservationAndLease: # apply_firewall_rule / commit_firewall_rules # --------------------------------------------------------------------------- + class TestApplyFirewallRule: def test_add_calls_addRule_with_translated_fields(self, driver): calls = [] @@ -2356,7 +2428,9 @@ class TestGetDhcpSubnets: paths = [] driver._get = lambda path: ( paths.append(path) - or (KEA_SUBNET_SEARCH_RESPONSE if "searchSubnet" in path else KEA_SUBNET_DETAIL_RESPONSE) + or ( + KEA_SUBNET_SEARCH_RESPONSE if "searchSubnet" in path else KEA_SUBNET_DETAIL_RESPONSE + ) ) driver.get_dhcp_subnets() @@ -2491,10 +2565,7 @@ class TestApplyDhcpSubnet: } ) - assert ( - calls[0][1]["subnet"]["pools"] - == "10.10.99.10-10.10.99.20\n10.10.99.30-10.10.99.40" - ) + assert calls[0][1]["subnet"]["pools"] == "10.10.99.10-10.10.99.20\n10.10.99.30-10.10.99.40" def test_explicit_option_data_disables_autocollect(self, driver): """Kea would otherwise re-fill routers/DNS/NTP and fight the desired state.""" @@ -2553,8 +2624,15 @@ class TestKeaSubnetRecordKey: # DNS host overrides # --------------------------------------------------------------------------- -def _override_row(uuid, host="opnsense-office", domain="office.local", - ip="10.10.0.1", description="[netork]", **extra): + +def _override_row( + uuid, + host="opnsense-office", + domain="office.local", + ip="10.10.0.1", + description="[netork]", + **extra, +): row = { "uuid": uuid, "hostname": host, @@ -2640,9 +2718,17 @@ class TestSyncDnsZone: rows = [_override_row("uuid-1"), _override_row("uuid-2"), _override_row("uuid-3")] calls = self._record_calls(driver, rows) - driver.sync_dns_zone("office.local", [ - {"hostname": "opnsense-office", "ip": "10.10.0.1", "record_type": "A", "enabled": True} - ]) + driver.sync_dns_zone( + "office.local", + [ + { + "hostname": "opnsense-office", + "ip": "10.10.0.1", + "record_type": "A", + "enabled": True, + } + ], + ) deleted = [p for p, _ in calls if "delhostoverride" in p] assert sorted(p.rsplit("/", 1)[-1] for p in deleted) == ["uuid-1", "uuid-2", "uuid-3"] @@ -2665,10 +2751,23 @@ class TestSyncDnsZone: in; the device still ends up with a single override.""" calls = self._record_calls(driver, []) - driver.sync_dns_zone("office.local", [ - {"hostname": "opnsense-office", "ip": "10.10.0.1", "record_type": "A", "enabled": True}, - {"hostname": "opnsense-office", "ip": "10.10.0.1", "record_type": "A", "enabled": True}, - ]) + driver.sync_dns_zone( + "office.local", + [ + { + "hostname": "opnsense-office", + "ip": "10.10.0.1", + "record_type": "A", + "enabled": True, + }, + { + "hostname": "opnsense-office", + "ip": "10.10.0.1", + "record_type": "A", + "enabled": True, + }, + ], + ) added = [d for p, d in calls if p.endswith("addhostoverride")] assert len(added) == 1 @@ -2680,9 +2779,12 @@ class TestSyncDnsZone: rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="")] calls = self._record_calls(driver, rows) - driver.sync_dns_zone("office.local", [ - {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, - ]) + driver.sync_dns_zone( + "office.local", + [ + {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, + ], + ) assert [p for p, _ in calls if p.endswith("addhostoverride")] == [] assert [p for p, _ in calls if "delhostoverride" in p] == [] @@ -2695,10 +2797,13 @@ class TestSyncDnsZone: ] calls = self._record_calls(driver, rows) - driver.sync_dns_zone("office.local", [ - {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, - {"hostname": "NAS", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, - ]) + driver.sync_dns_zone( + "office.local", + [ + {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, + {"hostname": "NAS", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, + ], + ) assert [p.rsplit("/", 1)[-1] for p, _ in calls if "delhostoverride" in p] == ["copy"] assert [p for p, _ in calls if p.endswith("addhostoverride")] == [] @@ -2707,33 +2812,44 @@ class TestSyncDnsZone: rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="")] calls = self._record_calls(driver, rows) - driver.sync_dns_zone("office.local", [ - {"hostname": "nas", "ip": "172.22.40.9", "record_type": "A", "enabled": True}, - ]) + driver.sync_dns_zone( + "office.local", + [ + {"hostname": "nas", "ip": "172.22.40.9", "record_type": "A", "enabled": True}, + ], + ) added = [d for p, d in calls if p.endswith("addhostoverride")] assert [d["host"]["server"] for d in added] == ["172.22.40.9"] def test_a_disabled_hand_made_override_does_not_stand_in(self, driver): """It answers nothing, so the record still has to be added.""" - rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="", - enabled="0")] + rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="", enabled="0")] calls = self._record_calls(driver, rows) - driver.sync_dns_zone("office.local", [ - {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, - ]) + driver.sync_dns_zone( + "office.local", + [ + {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, + ], + ) assert len([p for p, _ in calls if p.endswith("addhostoverride")]) == 1 def test_a_hand_made_override_in_another_zone_does_not_stand_in(self, driver): - rows = [_override_row("by-hand", host="nas", domain="other.local", ip="172.22.40.8", - description="")] + rows = [ + _override_row( + "by-hand", host="nas", domain="other.local", ip="172.22.40.8", description="" + ) + ] calls = self._record_calls(driver, rows) - driver.sync_dns_zone("office.local", [ - {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, - ]) + driver.sync_dns_zone( + "office.local", + [ + {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True}, + ], + ) assert len([p for p, _ in calls if p.endswith("addhostoverride")]) == 1 diff --git a/tests/unit/test_ha.py b/tests/unit/test_ha.py index b5044a9..d8e9cad 100644 --- a/tests/unit/test_ha.py +++ b/tests/unit/test_ha.py @@ -59,7 +59,11 @@ class TestDhcpHa: def test_an_explicit_server_name_wins(self): ha = dhcp_ha( - {"dhcpv4": {"ha": {"enabled": "1", "this_server_name": "left", "%this_server_name": "x"}}}, + { + "dhcpv4": { + "ha": {"enabled": "1", "this_server_name": "left", "%this_server_name": "x"} + } + }, {"rows": []}, ) assert ha["this_server"] == "left" @@ -124,7 +128,9 @@ class TestConfigSync: def test_without_a_peer_selected_items_are_not_pushed(self): """Ticked items without a target address push nothing.""" - answer = {"hasync": {"synchronizetoip": "", "syncitems": {"kea": {"value": "Kea", "selected": 1}}}} + answer = { + "hasync": {"synchronizetoip": "", "syncitems": {"kea": {"value": "Kea", "selected": 1}}} + } assert config_sync(answer) == {"peer": None, "services": [], "unmapped": []} diff --git a/tests/unit/test_ping.py b/tests/unit/test_ping.py index 666c581..a604f99 100644 --- a/tests/unit/test_ping.py +++ b/tests/unit/test_ping.py @@ -6,11 +6,11 @@ has to be stopped and removed again. These tests drive that whole lifecycle against a fake ``_get`` / ``_post``. """ -import pytest from unittest.mock import MagicMock, patch -from napalm_opnsense.opnsense import OPNsenseDriver +import pytest +from napalm_opnsense.opnsense import OPNsenseDriver # --------------------------------------------------------------------------- # Fake API @@ -28,11 +28,11 @@ class FakePingAPI: #: to assume one, which is why nothing caught the payload going to the #: wrong node. self.model_path = tuple(model_path) - self.created = [] # payloads passed to set - self.started = [] # job ids passed to start - self.stopped = [] # job ids passed to stop - self.removed = [] # job ids passed to remove - self.jobs = {} # job id -> hostname + self.created = [] # payloads passed to set + self.started = [] # job ids passed to start + self.stopped = [] # job ids passed to stop + self.removed = [] # job ids passed to remove + self.jobs = {} # job id -> hostname self.search_calls = 0 self._next_id = 0 diff --git a/tests/unit/test_port_forwards.py b/tests/unit/test_port_forwards.py index 20e07c9..e37244b 100644 --- a/tests/unit/test_port_forwards.py +++ b/tests/unit/test_port_forwards.py @@ -112,7 +112,10 @@ class TestWhereItGoes: assert entry["internal_ip"] == "10.0.0.9" def test_an_alias_with_two_hosts_is_two_forwards(self): - assert [e["internal_ip"] for e in _read(_rule(target="web_pair"))] == ["10.0.0.5", "10.0.0.6"] + assert [e["internal_ip"] for e in _read(_rule(target="web_pair"))] == [ + "10.0.0.5", + "10.0.0.6", + ] def test_an_alias_that_names_a_host_by_dns_is_skipped(self): """No address to put the forward on; guessing one would be worse.""" @@ -129,7 +132,14 @@ class TestWhereItGoes: class TestPorts: @pytest.mark.parametrize( ("value", "expected"), - [("443", 443), ("https", 443), ("http", 80), ("postgres", 5432), ("8000-8010", 8000), ("8000:8010", 8000)], + [ + ("443", 443), + ("https", 443), + ("http", 80), + ("postgres", 5432), + ("8000-8010", 8000), + ("8000:8010", 8000), + ], ) def test_the_external_port(self, value, expected): (entry,) = _read(_rule(**{"destination.port": value})) @@ -173,7 +183,7 @@ class TestNameAndSource: assert "remote_host" not in entry def test_an_inverted_source_has_no_remote_host(self): - """"Everyone but X" is as good as anyone for whether it is reachable.""" + """ "Everyone but X" is as good as anyone for whether it is reachable.""" (entry,) = _read(_rule(**{"source.network": "203.0.113.7", "source.not": "1"})) assert "remote_host" not in entry diff --git a/tests/unit/test_public_contracts.py b/tests/unit/test_public_contracts.py index def5728..36b67e8 100644 --- a/tests/unit/test_public_contracts.py +++ b/tests/unit/test_public_contracts.py @@ -18,7 +18,9 @@ from napalm_opnsense.opnsense import OPNsenseDriver @pytest.fixture def driver(): with patch("napalm_opnsense.opnsense.requests.Session"): - drv = OPNsenseDriver(hostname="fw", username="k", password="s", optional_args={"verify": False}) + drv = OPNsenseDriver( + hostname="fw", username="k", password="s", optional_args={"verify": False} + ) drv.session = MagicMock() yield drv @@ -26,7 +28,13 @@ def driver(): class TestUpdateProgress: @pytest.mark.parametrize( "status, state", - [("running", "running"), ("done", "done"), ("error", "failed"), ("", "idle"), (None, "idle")], + [ + ("running", "running"), + ("done", "done"), + ("error", "failed"), + ("", "idle"), + (None, "idle"), + ], ) def test_the_upgrade_status_as_a_state(self, driver, status, state): st = {"status": status, "log": "***GOT REQUEST TO UPGRADE***\nfetching\n"} @@ -54,8 +62,22 @@ class TestUpdateProgress: ROWS = [ - {"uuid": "u1", "enabled": "1", "type": "forward", "domain": "lab.example", "server": "10.0.0.53", "port": "53"}, - {"uuid": "u2", "enabled": "0", "type": "forward", "domain": "old.example", "server": "10.0.0.9", "port": ""}, + { + "uuid": "u1", + "enabled": "1", + "type": "forward", + "domain": "lab.example", + "server": "10.0.0.53", + "port": "53", + }, + { + "uuid": "u2", + "enabled": "0", + "type": "forward", + "domain": "old.example", + "server": "10.0.0.9", + "port": "", + }, {"uuid": "u3", "enabled": "1", "type": "dot", "domain": "", "server": "9.9.9.9", "port": "853"}, ] @@ -86,9 +108,18 @@ class TestDnsForwards: driver.set_dns_forward("new.example.", "10.0.0.53") paths = [p for p, _ in calls] - assert paths[-2:] == ["/api/unbound/settings/addForward", "/api/unbound/service/reconfigure"] + assert paths[-2:] == [ + "/api/unbound/settings/addForward", + "/api/unbound/service/reconfigure", + ] added = calls[-2][1]["dot"] - assert added == {"enabled": "1", "type": "forward", "domain": "new.example", "server": "10.0.0.53", "port": "53"} + assert added == { + "enabled": "1", + "type": "forward", + "domain": "new.example", + "server": "10.0.0.53", + "port": "53", + } def test_a_known_domain_is_changed_not_doubled(self, driver): calls = [] @@ -105,7 +136,10 @@ class TestDnsForwards: return {"rows": []} return {"result": "failed", "validations": {"dot.server": "invalid"}} - with patch.object(driver, "_post", side_effect=post), pytest.raises(RuntimeError, match="invalid"): + with ( + patch.object(driver, "_post", side_effect=post), + pytest.raises(RuntimeError, match="invalid"), + ): driver.set_dns_forward("x.example", "not-an-ip") def test_removing_deletes_every_forward_of_the_domain(self, driver): @@ -114,7 +148,10 @@ class TestDnsForwards: driver.remove_dns_forward("lab.example.") paths = [p for p, _ in calls] - assert paths[-2:] == ["/api/unbound/settings/delForward/u1", "/api/unbound/service/reconfigure"] + assert paths[-2:] == [ + "/api/unbound/settings/delForward/u1", + "/api/unbound/service/reconfigure", + ] def test_removing_an_unknown_domain_changes_nothing(self, driver): calls = [] diff --git a/tests/unit/test_snmp_agent.py b/tests/unit/test_snmp_agent.py index 3d53c82..38f0154 100644 --- a/tests/unit/test_snmp_agent.py +++ b/tests/unit/test_snmp_agent.py @@ -83,7 +83,11 @@ class TestTheCallersValues: assert result["success"] is True general = _body(api, "/api/netsnmp/general/set")["general"] assert general["enabled"] == "1" - assert (general["community"], general["contact"], general["location"]) == ("c0mm", "ops", "rack 1") + assert (general["community"], general["contact"], general["location"]) == ( + "c0mm", + "ops", + "rack 1", + ) assert general["listen"] == {"10.0.0.1": {"selected": 1}} def test_nothing_of_an_application_is_built_in(self, driver, api): -- 2.54.0 From 8815c4a58494cad66d78f4e01c7c2521e67a1738 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 23 Oct 2026 08:33:21 +0200 Subject: [PATCH 2/4] fix: what bandit and the remaining lint rules found - The module logger and the refresh constants of opnsense.py move below the imports (E402). - Two long result messages are wrapped in parentheses (E501). - nosec for the two wildcard addresses (B104): one is the address sockstat reports for "*", the other a route's gateway compared with 0.0.0.0. Neither binds a socket; both are false positives. --- napalm_opnsense/listening.py | 3 ++- napalm_opnsense/opnsense.py | 23 ++++++++++++++--------- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/napalm_opnsense/listening.py b/napalm_opnsense/listening.py index a20c306..463ce31 100644 --- a/napalm_opnsense/listening.py +++ b/napalm_opnsense/listening.py @@ -65,7 +65,8 @@ def _service_of(command: str, user: str, services: set[str]) -> str | None: def _address(protocol: str, address: str) -> tuple: """``(address, interface)``: ``*`` is every address of the socket's family.""" if address == "*": - return ("::" if protocol.endswith("6") else "0.0.0.0"), None + # A reported wildcard address (data from sockstat), not a bind. + return ("::" if protocol.endswith("6") else "0.0.0.0"), None # nosec B104 host, _, zone = address.partition("%") return host, zone or None diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 5722eba..5d2eb0a 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -44,12 +44,6 @@ import time from ipaddress import ip_address, ip_network from typing import Any -logger = logging.getLogger(__name__) - -#: How long refresh_available_updates waits for the firewall's update check. -_REFRESH_POLLS = 20 -_REFRESH_INTERVAL = 3 - import requests from napalm.base.exceptions import ( ConnectionClosedException, @@ -76,6 +70,12 @@ from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interf from napalm_opnsense.resolver_mixin import OPNsenseResolverMixin from napalm_opnsense.virtual_ips import carp_addresses +logger = logging.getLogger(__name__) + +#: How long refresh_available_updates waits for the firewall's update check. +_REFRESH_POLLS = 20 +_REFRESH_INTERVAL = 3 + _ARCHITECTURE_SUFFIX = re.compile(r"-(amd64|aarch64|arm64|i386|armv7)$") @@ -636,7 +636,8 @@ class OPNsenseDriver( # Clean up BSD link-layer gateway references clean_gateway = ( "" - if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") + # A route's reported gateway compared with the wildcard, not a bind. + if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") # nosec B104 else gateway ) @@ -2261,7 +2262,9 @@ class OPNsenseDriver( return {"success": True, "output": status.get("status_msg", "")} return { "success": False, - "output": f"The update check did not finish within {_REFRESH_POLLS * _REFRESH_INTERVAL} s", + "output": ( + f"The update check did not finish within {_REFRESH_POLLS * _REFRESH_INTERVAL} s" + ), } def reboot_host(self) -> None: @@ -2464,7 +2467,9 @@ class OPNsenseDriver( # empty {} response at HTTP 200 — no error_msg to surface. return { "success": False, - "output": "OPNsense rejected the request (check interface identifier and MAC format)", + "output": ( + "OPNsense rejected the request (check interface identifier and MAC format)" + ), } return {"success": True, "output": f"Magic packet sent to {mac_address} via {interface}"} -- 2.54.0 From 6e0e206e067aa140071c52202100afa5afab973e Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 23 Oct 2026 08:35:15 +0200 Subject: [PATCH 3/4] chore: adopt the NAPALM driver standard - CI moves to .gitea/workflows/ci.yml and calls the shared driver workflow of NAPALM/driver-standard; .github/ is gone. - setuptools>=77 with the SPDX license string, no License classifier, Python 3.10-3.13 classifiers, napalm-device-types>=4.21.1, a dev extra with floors, links to git.netork.io, LICENSE file, py.typed. - CHANGELOG.md in Keep-a-Changelog form; README names Python 3.10, the Gitea clone URL and the new CI. - mypy strict for the package; exempt for now (52 errors): ha, listening, opnsense, ospf_mixin, ping_mixin, port_forwards, resolver_mixin. Recorded in the quality baseline with the long and complex functions, the 3100-line opnsense.py and one unannotated public method. - Coverage floor 79 % (measured 79.20 %). --- .gitea/workflows/ci.yml | 15 +++ .github/workflows/ci.yml | 48 --------- CHANGELOG.md | 43 ++++++++ LICENSE | 191 ++++++++++++++++++++++++++++++++++++ README.md | 20 ++-- napalm_opnsense/py.typed | 0 pyproject.toml | 53 ++++++++-- tests/quality_baseline.json | 65 ++++++++++++ 8 files changed, 368 insertions(+), 67 deletions(-) create mode 100644 .gitea/workflows/ci.yml delete mode 100644 .github/workflows/ci.yml create mode 100644 CHANGELOG.md create mode 100644 LICENSE create mode 100644 napalm_opnsense/py.typed create mode 100644 tests/quality_baseline.json diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..7530c79 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,15 @@ +name: CI + +on: + push: + branches: [main] + 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..22fafef --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,43 @@ +# Changelog + +All notable changes to this driver. The format follows +[Keep a Changelog](https://keepachangelog.com/), versions follow +[Semantic Versioning](https://semver.org/). + +## [Unreleased] + +### 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.10 remains the floor. Requires napalm-device-types + 4.21.1. + +### Fixed +- `get_dhcp_leases()` raises when Kea, ISC DHCP and the ARP fallback all failed, + instead of returning an empty list that looks like a server without leases (#27). +- `create_dhcp_reservation()` leaves a reservation that already holds what was asked + alone, and does not reconfigure (restart) Kea for it. + +## [0.4.0] - 2026-10-18 + +### Added +- napalm-device-types 4.17's public contracts: `apply_updates()` passes the packages it + is given and `get_update_progress()` reports the upgrade's state and log; + `get_dns_forwards()`, `set_dns_forward()` and `remove_dns_forward()` for Unbound; + `get_ospf()`. Requires napalm-device-types 4.17.0. + +## [0.3.0] - 2026-10-17 + +### Removed +- **Breaking:** `get_device_warnings()`. The pending updates are in + `get_available_updates()`. + +## [0.2.0] - 2026-10-17 + +### Changed +- **Breaking:** `run_device_action` is removed. The SNMP agent is set up through + napalm-device-types 4.13's `configure_snmp_agent` with the caller's community, + contact, location and listen address, and the firewall opens UDP/161 only to the + allowed sources (#13). 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 6496f1f..2b52992 100644 --- a/README.md +++ b/README.md @@ -14,8 +14,9 @@ NAPALM community driver for **OPNsense** firewalls (read-only, via REST API). | Dependency | Minimum version | |---|---| -| Python | 3.9 | +| Python | 3.10 | | NAPALM | 4.0 | +| napalm-device-types | 4.21.1 | | requests | 2.28 | ## Installation @@ -27,7 +28,7 @@ pip install napalm-opnsense Or from source: ```bash -git clone https://github.com/napalm-automation-community/napalm-opnsense +git clone https://git.netork.io/NAPALM/napalm-opnsense cd napalm-opnsense pip install -e . ``` @@ -201,19 +202,18 @@ d.close() ## Development +The rules are in [STANDARDS.md](https://git.netork.io/NAPALM/driver-standard/src/branch/main/STANDARDS.md). + ```bash -pip install -e ".[dev]" -pytest tests/ +python -m venv .venv && .venv/bin/pip install -e ".[dev]" +.venv/bin/pytest --cov ``` ## CI -This project includes a GitHub Actions workflow that: -- Runs unit tests across Python 3.9–3.12 -- Builds sdist and wheel -- Uploads build artifacts - -See [.github/workflows/ci.yml](.github/workflows/ci.yml). +[.gitea/workflows/ci.yml](.gitea/workflows/ci.yml) calls the shared workflow of the +driver standard: ruff, mypy, the tests on Python 3.10–3.13 and at the lowest declared +versions, bandit, pip-audit and a build check. A release tag publishes to PyPI. ## License diff --git a/napalm_opnsense/py.typed b/napalm_opnsense/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml index 2cc1d7a..77a8bb9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [build-system] -requires = ["setuptools>=68", "wheel"] +requires = ["setuptools>=77"] build-backend = "setuptools.build_meta" [project] @@ -7,45 +7,51 @@ name = "napalm-opnsense" version = "0.4.0" description = "NAPALM driver for OPNsense (read-only via REST API)." readme = "README.md" -license = { text = "Apache-2.0" } +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.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", "Operating System :: POSIX :: Linux", "Operating System :: MacOS", + "Typing :: Typed", ] dependencies = [ "napalm>=4.0.0", - "napalm_device_types>=4.17.0", + "napalm-device-types>=4.21.1", "requests>=2.28.0", ] [project.optional-dependencies] dev = [ - "pytest", - "pytest-cov", - "black", - "ruff", + "pytest>=8", + "pytest-cov>=5", + "mypy>=1.11", + "ruff>=0.6", ] [project.entry-points."napalm.drivers"] opnsense = "napalm_opnsense.opnsense:OPNsenseDriver" [project.urls] -Repository = "https://github.com/napalm-automation-community/napalm-opnsense" +Repository = "https://git.netork.io/NAPALM/napalm-opnsense" +Issues = "https://git.netork.io/NAPALM/napalm-opnsense/issues" +Changelog = "https://git.netork.io/NAPALM/napalm-opnsense/src/branch/main/CHANGELOG.md" [tool.setuptools.packages.find] where = ["."] include = ["napalm_opnsense*"] +[tool.setuptools.package-data] +napalm_opnsense = ["py.typed"] + [tool.ruff] line-length = 100 target-version = "py310" @@ -57,5 +63,34 @@ select = ["E", "F", "I", "UP", "B"] # 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.*", "requests.*"] +ignore_missing_imports = true + +[[tool.mypy.overrides]] +# Not yet under strict mypy (quality baseline); may only shrink. +module = [ + "napalm_opnsense.ha", + "napalm_opnsense.listening", + "napalm_opnsense.opnsense", + "napalm_opnsense.ospf_mixin", + "napalm_opnsense.ping_mixin", + "napalm_opnsense.port_forwards", + "napalm_opnsense.resolver_mixin", +] +ignore_errors = true + [tool.pytest.ini_options] testpaths = ["tests"] + +[tool.coverage.run] +source = ["napalm_opnsense"] + +[tool.coverage.report] +fail_under = 79 diff --git a/tests/quality_baseline.json b/tests/quality_baseline.json new file mode 100644 index 0000000..79a77f6 --- /dev/null +++ b/tests/quality_baseline.json @@ -0,0 +1,65 @@ +{ + "coverage_floor": 79.0, + "function_complexity": { + "napalm_opnsense/ha.py::config_sync": 12, + "napalm_opnsense/ha.py::dhcp_ha": 11, + "napalm_opnsense/listening.py::_service_of": 12, + "napalm_opnsense/listening.py::_socket": 19, + "napalm_opnsense/opnsense.py::OPNsenseDriver._allow_snmp_sources": 11, + "napalm_opnsense/opnsense.py::OPNsenseDriver._get_unbound_host_overrides": 17, + "napalm_opnsense/opnsense.py::OPNsenseDriver._kea_read_list": 11, + "napalm_opnsense/opnsense.py::OPNsenseDriver.apply_dhcp_subnet": 21, + "napalm_opnsense/opnsense.py::OPNsenseDriver.apply_firewall_rule": 11, + "napalm_opnsense/opnsense.py::OPNsenseDriver.create_dhcp_reservation": 16, + "napalm_opnsense/opnsense.py::OPNsenseDriver.delete_dhcp_reservation_and_lease": 14, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_bgp_neighbors": 16, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_dhcp_leases": 15, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_dhcp_leases._parse_kea": 14, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_dhcp_reservations": 15, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_dhcp_subnets": 17, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_facts": 12, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_firewall_rules": 30, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_interfaces": 13, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_interfaces_ip": 34, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_lldp_neighbors_detail": 12, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_networks": 25, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_route_to": 23, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_vpn_tunnels": 60, + "napalm_opnsense/opnsense.py::OPNsenseDriver.load_merge_candidate": 13, + "napalm_opnsense/opnsense.py::OPNsenseDriver.sync_dns_zone": 20, + "napalm_opnsense/port_forwards.py::_forwards_of": 12, + "napalm_opnsense/virtual_ips.py::carp_addresses": 14 + }, + "function_lines": { + "napalm_opnsense/opnsense.py::OPNsenseDriver._get_unbound_host_overrides": 51, + "napalm_opnsense/opnsense.py::OPNsenseDriver._regenerate_web_cert": 51, + "napalm_opnsense/opnsense.py::OPNsenseDriver.apply_dhcp_subnet": 60, + "napalm_opnsense/opnsense.py::OPNsenseDriver.create_dhcp_reservation": 85, + "napalm_opnsense/opnsense.py::OPNsenseDriver.delete_dhcp_reservation_and_lease": 79, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_bgp_neighbors": 86, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_dhcp_leases": 106, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_dhcp_subnets": 60, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_firewall_rules": 84, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_interfaces_ip": 68, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_networks": 85, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_route_to": 85, + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_vpn_tunnels": 189, + "napalm_opnsense/opnsense.py::OPNsenseDriver.load_merge_candidate": 56, + "napalm_opnsense/opnsense.py::OPNsenseDriver.sync_dns_zone": 112 + }, + "mypy_relaxed": [ + "napalm_opnsense.ha", + "napalm_opnsense.listening", + "napalm_opnsense.opnsense", + "napalm_opnsense.ospf_mixin", + "napalm_opnsense.ping_mixin", + "napalm_opnsense.port_forwards", + "napalm_opnsense.resolver_mixin" + ], + "python_file_lines": { + "napalm_opnsense/opnsense.py": 3125 + }, + "untyped_public": [ + "napalm_opnsense/opnsense.py::OPNsenseDriver.get_snmp_config" + ] +} -- 2.54.0 From ae732fd7c81c835f1baee95ed0de30ee70c8a8ba Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 23 Oct 2026 10:29:08 +0200 Subject: [PATCH 4/4] chore(release): 0.5.0 --- CHANGELOG.md | 4 ++++ pyproject.toml | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 22fafef..ed4ea8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ All notable changes to this driver. The format follows ## [Unreleased] +## [0.5.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 diff --git a/pyproject.toml b/pyproject.toml index 77a8bb9..1270e2d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "napalm-opnsense" -version = "0.4.0" +version = "0.5.0" description = "NAPALM driver for OPNsense (read-only via REST API)." readme = "README.md" license = "Apache-2.0" -- 2.54.0