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):