.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 }}

.github/workflows/ci.yml

@@ -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/*

CHANGELOG.md

@@ -0,0 +1,47 @@
+# Changelog
+
+All notable changes to this driver. The format follows
+[Keep a Changelog](https://keepachangelog.com/), versions follow
+[Semantic Versioning](https://semver.org/).
+
+## [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
+  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).

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.

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
 

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
@@ -64,14 +65,15 @@ def _service_of(command: str, user: str, services: Set[str]) -> Optional[str]:
 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
 
 
 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 +103,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 +114,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

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
@@ -45,36 +44,38 @@ import time
 from ipaddress import ip_address, ip_network
 from typing import Any
 
+import requests
+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 (
+    ConfigSyncDict,
+    DhcpHaDict,
+    FixStepOutcomeDict,
+    ListeningSocketsDict,
+    SnmpAgentSettingsDict,
+    VirtualIpsDict,
+)
+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.ospf_mixin import OPNsenseOspfMixin
+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
+
 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 requests.exceptions import RequestException
-
-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,
-    ListeningSocketsDict,
-    VirtualIpsDict,
-)
-from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException
-
-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.port_forwards import alias_index, port_forwards, wan_interfaces
-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({
+                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,12 @@ 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 = (
+                ""
+                # 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
+            )
 
             # Determine address family from network address
             family = "ipv6" if (":" in network or (gateway and ":" in gateway)) else "ipv4"
@@ -711,9 +726,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 +991,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 +1002,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 +1050,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 +1158,16 @@ class OPNsenseDriver(
         for p in info.get("plugin", []):
             if p.get("installed") != "1":
                 continue
-            result.append({
+            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 +1193,15 @@ class OPNsenseDriver(
 
         result: list[dict[str, Any]] = []
         for row in data.get("rows", []):
-            result.append({
+            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 +1351,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 +1368,17 @@ class OPNsenseDriver(
                     lease_end = int(raw_end)
                 except (ValueError, TypeError):
                     lease_end = 0
-                result.append({
+                result.append(
+                    {
                         "mac": mac,
-                    "ip": (row.get("address") or row.get("ip-address") or row.get("ip") or "").strip(),
+                        "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 +1387,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 +1678,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 +1690,16 @@ class OPNsenseDriver(
             else:
                 subnet = cidr_by_uuid.get(raw_subnet, "")
 
-            reservations.append({
+            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 +1729,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 +1847,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({
+            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 +1917,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 +1938,14 @@ class OPNsenseDriver(
         data = self._get("/api/core/service/search")
         result: list[dict[str, Any]] = []
         for row in data.get("rows", []):
-            result.append({
+            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 +1991,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 +2012,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 +2084,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",
@@ -2239,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:
@@ -2271,7 +2296,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 +2329,16 @@ class OPNsenseDriver(
             name = p.get("name", "")
             comment = p.get("comment", "")
             if q in name.lower() or q in comment.lower():
-                result.append({
+                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 +2347,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 +2377,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(
@@ -2436,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}"}
 
@@ -2464,14 +2497,16 @@ 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", {
+        result = self._post(
+            "/api/netsnmp/general/set",
+            {
                 "general": {
                     "enabled": "1",
                     "community": community,
@@ -2481,22 +2516,34 @@ class OPNsenseDriver(
                     "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", {
+            self._post(
+                "/api/firewall/filter/addRule",
+                {
                     "rule": {
                         "enabled": "1",
                         "action": "pass",
@@ -2514,7 +2561,8 @@ class OPNsenseDriver(
                         "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 +2608,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({
+            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 +2646,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 +2662,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
 
@@ -2622,9 +2671,12 @@ class OPNsenseDriver(
 
             # 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()
+            destination_net = (
+                row.get("%destination_net") or row.get("destination_net") or "any"
+            ).strip()
 
-            result.append({
+            result.append(
+                {
                     "uuid": row.get("uuid", ""),
                     "sequence": int(row.get("sequence", 0) or 0),
                     "action": row.get("action", "pass"),
@@ -2647,11 +2699,16 @@ class OPNsenseDriver(
                     "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 +2814,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,7 +2922,8 @@ 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({
+                result.append(
+                    {
                         "uuid": row.get("uuid", ""),
                         "hostname": host,
                         "domain": domain,
@@ -2876,7 +2934,8 @@ class OPNsenseDriver(
                         "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,7 +2951,8 @@ 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({
+                result.append(
+                    {
                         "uuid": row.get("uuid", ""),
                         "hostname": host,
                         "domain": domain,
@@ -2902,7 +2962,8 @@ class OPNsenseDriver(
                         "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 +3027,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,7 +3075,9 @@ class OPNsenseDriver(
                 continue
             pushed.add(key)
             if service == "unbound":
-                self._post("/api/unbound/settings/addhostoverride", {
+                self._post(
+                    "/api/unbound/settings/addhostoverride",
+                    {
                         "host": {
                             "enabled": "1",
                             "hostname": rec.get("hostname", ""),
@@ -3018,9 +3090,12 @@ class OPNsenseDriver(
                             "mxprio": "",
                             "mx": "",
                         }
-                })
+                    },
+                )
             else:
-                self._post("/api/dnsmasq/settings/addhostoverride", {
+                self._post(
+                    "/api/dnsmasq/settings/addhostoverride",
+                    {
                         "hostoverride": {
                             "enabled": "1",
                             "host": rec.get("hostname", ""),
@@ -3028,10 +3103,16 @@ class OPNsenseDriver(
                             "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")

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()),

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.
 

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

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:

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

pyproject.toml

@@ -1,54 +1,96 @@
 [build-system]
-requires = ["setuptools>=68", "wheel"]
+requires = ["setuptools>=77"]
 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 = { 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 = "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.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

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"
+ ]
+}

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 {
+        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 {
+        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([
+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: "<opnsense/>"
@@ -1683,7 +1748,9 @@ class TestGetConfigCandidate:
 # the getter called .json() on XML and always returned an empty "running".
 # ---------------------------------------------------------------------------
 
-CONFIG_XML = '<?xml version="1.0"?>\n<opnsense><system><hostname>gw</hostname></system></opnsense>\n'
+CONFIG_XML = (
+    '<?xml version="1.0"?>\n<opnsense><system><hostname>gw</hostname></system></opnsense>\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", [
+        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", [
+        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", [
+        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", [
+        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", [
+        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
 

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": []}
 

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

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

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 = []

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