Merge pull request 'chore: adopt the NAPALM driver standard' (#31) from chore/driver-standard into main

- CI moves to .gitea/workflows/ci.yml and calls the shared driver workflow
  of NAPALM/driver-standard; .github/ is gone.
- setuptools>=77 with the SPDX license string, no License classifier,
  Python 3.10-3.13 classifiers, napalm-device-types>=4.21.1, a dev extra
  with floors, links to git.netork.io, LICENSE file, py.typed.
- CHANGELOG.md in Keep-a-Changelog form; README names Python 3.10, the
  Gitea clone URL and the new CI.
- mypy strict for the package; exempt for now (52 errors): ha, listening,
  opnsense, ospf_mixin, ping_mixin, port_forwards, resolver_mixin.
  Recorded in the quality baseline with the long and complex functions,
  the 3100-line opnsense.py and one unannotated public method.
- Coverage floor 79 % (measured 79.20 %).

.gitea/workflows/ci.yml

@@ -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 .
 ```
@@ -57,7 +58,7 @@ API keys**.
 Pass the credentials via `optional_args`:
 
 ```python
-optional_args={
+optional_args = {
     "api_key": "your-api-key",
     "api_secret": "your-api-secret",
     "verify": False,  # set to a CA bundle path or True in production
@@ -173,9 +174,9 @@ d.open()
 routes = json.dumps([{"network": "10.0.0.0/8", "gateway": "WAN_GW"}])
 d.load_merge_candidate(config=routes)
 
-print(d.compare_config())   # unified diff
-d.commit_config()           # applies routes and calls reconfigure
-d.rollback()                # removes the routes just added
+print(d.compare_config())  # unified diff
+d.commit_config()  # applies routes and calls reconfigure
+d.rollback()  # removes the routes just added
 d.close()
 ```
 
@@ -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/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_dhcp_leases.py

@@ -0,0 +1,109 @@
+"""``get_dhcp_leases()``: an empty answer is only ever a real one (#27).
+
+The getter tries Kea, then ISC DHCP, then falls back to the ARP table (rows
+with ``state: "arp"``). A caller stores the answer as the server's lease set,
+so ``[]`` must mean "a lease source answered and has no leases". When no lease
+source answered and the ARP fallback gives nothing either, the getter raises:
+an empty list there would wipe the leases the caller knows.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from napalm_opnsense.opnsense import OPNsenseDriver
+
+KEA = "/api/kea/leases4/search"
+ISC = "/api/dhcpv4/leases/searchlease"
+ARP = "/api/diagnostics/interface/get_arp"
+
+KEA_ROW = {"hwaddr": "AA:BB:CC:DD:EE:01", "address": "10.0.0.11", "hostname": "nas", "state": "0"}
+ISC_ROW = {"mac": "aa:bb:cc:dd:ee:02", "address": "10.0.0.12", "ends": "1700000000"}
+ARP_ROW = {"intf": "igb0", "mac": "AA:BB:CC:DD:EE:03", "ip": "10.0.0.13", "expires": "1200"}
+
+
+def _driver(kea=None, isc=None, arp=None) -> OPNsenseDriver:
+    """Each source answers its value, or raises it when it is an exception."""
+
+    def answer(value):
+        if isinstance(value, Exception):
+            raise value
+        return value
+
+    driver = OPNsenseDriver("gw-01", "user", "secret")
+    driver._get = MagicMock(side_effect=lambda path: answer({KEA: kea, ARP: arp}[path]))
+    driver._post = MagicMock(side_effect=lambda path, data=None: answer({ISC: isc}[path]))
+    return driver
+
+
+def _macs(leases: list[dict]) -> list[str]:
+    return [lease["mac"] for lease in leases]
+
+
+class TestALeaseSourceAnswers:
+    def test_kea_leases_come_first(self):
+        driver = _driver(kea={"rows": [KEA_ROW]}, isc=RuntimeError("404"), arp=RuntimeError("x"))
+        assert _macs(driver.get_dhcp_leases()) == ["aa:bb:cc:dd:ee:01"]
+
+    def test_isc_leases_when_kea_has_none(self):
+        driver = _driver(kea={"rows": []}, isc={"rows": [ISC_ROW]}, arp=RuntimeError("x"))
+        assert _macs(driver.get_dhcp_leases()) == ["aa:bb:cc:dd:ee:02"]
+
+    def test_isc_leases_when_kea_fails(self):
+        driver = _driver(kea=RuntimeError("404"), isc={"rows": [ISC_ROW]}, arp=RuntimeError("x"))
+        assert _macs(driver.get_dhcp_leases()) == ["aa:bb:cc:dd:ee:02"]
+
+
+class TestTheArpFallback:
+    """No source has leases: the ARP table stands in, marked ``state: "arp"`` (#1168)."""
+
+    def test_when_both_lease_sources_fail(self):
+        driver = _driver(kea=RuntimeError("404"), isc=RuntimeError("404"), arp=[ARP_ROW])
+        leases = driver.get_dhcp_leases()
+        assert leases == [
+            {
+                "mac": "aa:bb:cc:dd:ee:03",
+                "ip": "10.0.0.13",
+                "hostname": "",
+                "lease_end": 0,
+                "state": "arp",
+            }
+        ]
+
+    def test_when_a_lease_source_answers_empty(self):
+        driver = _driver(kea={"rows": []}, isc=RuntimeError("404"), arp=[ARP_ROW])
+        assert [lease["state"] for lease in driver.get_dhcp_leases()] == ["arp"]
+
+
+class TestAnEmptyAnswer:
+    """[] only when Kea or ISC answered and has no leases."""
+
+    def test_kea_empty_and_no_arp_fallback(self):
+        driver = _driver(kea={"rows": []}, isc=RuntimeError("404"), arp=RuntimeError("timeout"))
+        assert driver.get_dhcp_leases() == []
+
+    def test_isc_empty_and_an_empty_arp_table(self):
+        driver = _driver(kea=RuntimeError("404"), isc={"rows": []}, arp=[])
+        assert driver.get_dhcp_leases() == []
+
+
+class TestNothingAnswered:
+    """No lease source answered and the fallback gave no rows: that is no answer."""
+
+    def test_all_three_fail(self):
+        driver = _driver(
+            kea=RuntimeError("kea down"), isc=RuntimeError("isc down"), arp=RuntimeError("arp down")
+        )
+        with pytest.raises(RuntimeError) as excinfo:
+            driver.get_dhcp_leases()
+        message = str(excinfo.value)
+        assert "kea down" in message
+        assert "isc down" in message
+        assert "arp down" in message
+
+    def test_lease_sources_fail_and_the_arp_table_is_empty(self):
+        driver = _driver(kea=RuntimeError("kea down"), isc=RuntimeError("isc down"), arp=[])
+        with pytest.raises(RuntimeError, match="kea down"):
+            driver.get_dhcp_leases()

tests/unit/test_driver.py

@@ -1,18 +1,19 @@
 """Unit tests for OPNsenseDriver — no real device required."""
 
 import json
-import pytest
-import requests
 from unittest.mock import MagicMock, patch
 
-from napalm_opnsense.opnsense import OPNsenseDriver
-from napalm.base.exceptions import ConnectionException, ConnectionClosedException
+import pytest
+import requests
+from napalm.base.exceptions import ConnectionClosedException, ConnectionException
 
+from napalm_opnsense.opnsense import OPNsenseDriver
 
 # ---------------------------------------------------------------------------
 # Fixtures
 # ---------------------------------------------------------------------------
 
+
 @pytest.fixture
 def driver():
     """Return a driver instance with a mocked requests.Session."""
@@ -110,6 +111,7 @@ ARP_RESPONSE = {
 # Helpers
 # ---------------------------------------------------------------------------
 
+
 def _make_json_response(data):
     mock_resp = MagicMock()
     mock_resp.json.return_value = data
@@ -121,6 +123,7 @@ def _make_json_response(data):
 # open() / close() / is_alive()
 # ---------------------------------------------------------------------------
 
+
 class TestOpenClose:
     def test_open_raises_connection_exception_on_error(self):
         drv = OPNsenseDriver(
@@ -166,6 +169,7 @@ class TestIsAlive:
 # _get()
 # ---------------------------------------------------------------------------
 
+
 class TestInternalGet:
     def test_raises_when_no_session(self):
         drv = OPNsenseDriver("host", "u", "p")
@@ -185,41 +189,40 @@ class TestInternalGet:
 # get_facts()
 # ---------------------------------------------------------------------------
 
+
 class TestGetFacts:
     def test_returns_required_keys(self, driver):
-        driver._get = lambda path: (
-            STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
-        )
+        driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
         facts = driver.get_facts()
-        for key in ("vendor", "model", "hostname", "fqdn", "os_version",
-                    "serial_number", "uptime", "interface_list"):
+        for key in (
+            "vendor",
+            "model",
+            "hostname",
+            "fqdn",
+            "os_version",
+            "serial_number",
+            "uptime",
+            "interface_list",
+        ):
             assert key in facts
 
     def test_vendor_constant(self, driver):
-        driver._get = lambda path: (
-            STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
-        )
+        driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
         facts = driver.get_facts()
         assert facts["vendor"] == "OPNsense"
 
     def test_hostname_parsed(self, driver):
-        driver._get = lambda path: (
-            STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
-        )
+        driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
         facts = driver.get_facts()
         assert facts["hostname"] == "opnsense01"
 
     def test_os_version_parsed(self, driver):
-        driver._get = lambda path: (
-            STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
-        )
+        driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
         facts = driver.get_facts()
         assert facts["os_version"] == "24.7"
 
     def test_interface_list_populated(self, driver):
-        driver._get = lambda path: (
-            STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
-        )
+        driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
         facts = driver.get_facts()
         assert "em0" in facts["interface_list"]
         assert "em1" in facts["interface_list"]
@@ -235,16 +238,12 @@ class TestGetFacts:
         assert facts["interface_list"] == []
 
     def test_serial_number(self, driver):
-        driver._get = lambda path: (
-            STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
-        )
+        driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
         facts = driver.get_facts()
         assert facts["serial_number"] == "ABC123"
 
     def test_uptime(self, driver):
-        driver._get = lambda path: (
-            STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
-        )
+        driver._get = lambda path: STATUS_RESPONSE if "status" in path else INTERFACES_RESPONSE
         facts = driver.get_facts()
         assert facts["uptime"] == 12345
 
@@ -253,6 +252,7 @@ class TestGetFacts:
 # get_interfaces()
 # ---------------------------------------------------------------------------
 
+
 class TestGetInterfaces:
     def test_interface_count(self, driver):
         driver._get = lambda path: INTERFACES_RESPONSE
@@ -311,6 +311,7 @@ class TestGetInterfaces:
 # get_interfaces_ip()
 # ---------------------------------------------------------------------------
 
+
 class TestGetInterfacesIp:
     def test_entry_count(self, driver):
         driver._get = lambda path: ADDRESSES_RESPONSE
@@ -340,6 +341,7 @@ class TestGetInterfacesIp:
 # get_arp_table()
 # ---------------------------------------------------------------------------
 
+
 class TestGetArpTable:
     def test_entry_count(self, driver):
         driver._get = lambda path: ARP_RESPONSE
@@ -419,11 +421,18 @@ class TestGetInterfacesCounters:
         driver._get = lambda path: INTERFACE_STATISTICS_RESPONSE
         entry = driver.get_interfaces_counters()["em0"]
         for key in (
-            "tx_errors", "rx_errors", "tx_discards", "rx_discards",
-            "tx_octets", "rx_octets",
-            "tx_unicast_packets", "rx_unicast_packets",
-            "tx_multicast_packets", "rx_multicast_packets",
-            "tx_broadcast_packets", "rx_broadcast_packets",
+            "tx_errors",
+            "rx_errors",
+            "tx_discards",
+            "rx_discards",
+            "tx_octets",
+            "rx_octets",
+            "tx_unicast_packets",
+            "rx_unicast_packets",
+            "tx_multicast_packets",
+            "rx_multicast_packets",
+            "tx_broadcast_packets",
+            "rx_broadcast_packets",
         ):
             assert key in entry
 
@@ -505,7 +514,8 @@ class TestGetEnvironment:
     def test_temperature_alert_thresholds(self, driver):
         driver._get = lambda path: (
             {"data": [{"device": "cpu0", "temperature": "85.0"}]}
-            if "temperature" in path else SYSTEM_RESOURCES_RESPONSE
+            if "temperature" in path
+            else SYSTEM_RESOURCES_RESPONSE
         )
         env = driver.get_environment()
         assert env["temperature"]["cpu0"]["is_alert"] is True
@@ -513,6 +523,7 @@ class TestGetEnvironment:
 
     def test_temperature_endpoint_failing_gracefully(self, driver):
         """Driver must not raise if temperature endpoint is unavailable."""
+
         def fake_get(path):
             if "temperature" in path:
                 raise Exception("no sensor data")
@@ -594,19 +605,22 @@ class TestGetRouteTo:
     def test_filter_by_protocol(self, driver):
         driver._get = lambda path: ROUTES_RESPONSE
         routes = driver.get_route_to(protocol="static")
-        assert all(
-            e["protocol"] == "static"
-            for entries in routes.values()
-            for e in entries
-        )
+        assert all(e["protocol"] == "static" for entries in routes.values() for e in entries)
 
     def test_required_keys_in_entry(self, driver):
         driver._get = lambda path: ROUTES_RESPONSE
         entry = driver.get_route_to()["0.0.0.0/0"][0]
         for key in (
-            "protocol", "current_active", "last_active", "age",
-            "next_hop", "outgoing_interface", "selected_next_hop",
-            "preference", "inactive_reason", "routing_table",
+            "protocol",
+            "current_active",
+            "last_active",
+            "age",
+            "next_hop",
+            "outgoing_interface",
+            "selected_next_hop",
+            "preference",
+            "inactive_reason",
+            "routing_table",
             "protocol_attributes",
         ):
             assert key in entry
@@ -707,15 +721,21 @@ class TestGetLldpNeighborsDetail:
         detail = driver.get_lldp_neighbors_detail()
         entry = detail["em0"][0]
         for key in (
-            "remote_chassis_id", "remote_system_name", "remote_port",
-            "remote_port_description", "remote_system_description",
-            "remote_system_capab", "remote_system_enable_capab",
+            "remote_chassis_id",
+            "remote_system_name",
+            "remote_port",
+            "remote_port_description",
+            "remote_system_description",
+            "remote_system_capab",
+            "remote_system_enable_capab",
         ):
             assert key in entry
 
     def test_chassis_id(self, driver):
         driver._get = lambda path: LLDP_RESPONSE
-        assert driver.get_lldp_neighbors_detail()["em0"][0]["remote_chassis_id"] == "aa:bb:cc:dd:ee:ff"
+        assert (
+            driver.get_lldp_neighbors_detail()["em0"][0]["remote_chassis_id"] == "aa:bb:cc:dd:ee:ff"
+        )
 
     def test_system_name(self, driver):
         driver._get = lambda path: LLDP_RESPONSE
@@ -1010,12 +1030,19 @@ class TestGetBgpNeighbors:
     def test_required_peer_keys(self, driver):
         driver._get = self._fake_get
         peer = driver.get_bgp_neighbors()["global"]["peers"]["10.0.0.1"]
-        for key in ("local_as", "remote_as", "remote_id", "is_up", "is_enabled",
-                    "description", "uptime", "address_family"):
+        for key in (
+            "local_as",
+            "remote_as",
+            "remote_id",
+            "is_up",
+            "is_enabled",
+            "description",
+            "uptime",
+            "address_family",
+        ):
             assert key in peer
 
 
-
 # ---------------------------------------------------------------------------
 # get_certificates()
 #
@@ -1180,15 +1207,24 @@ class TestGetDdnsStatus:
 
 RADIUS_CLIENTS_RESPONSE = {
     "rows": [
-        {"uuid": "e5483f1b-936b-47ba-8cca-24d56ad643c6", "enabled": "1",
-         "name": "ap-lobby", "secret": "s3cr3t", "ip": "10.0.0.5/32"},
+        {
+            "uuid": "e5483f1b-936b-47ba-8cca-24d56ad643c6",
+            "enabled": "1",
+            "name": "ap-lobby",
+            "secret": "s3cr3t",
+            "ip": "10.0.0.5/32",
+        },
     ]
 }
 
 RADIUS_USERS_RESPONSE = {
     "rows": [
-        {"uuid": "71da23fd-03f3-4e1f-a7e1-bb25645b981e", "enabled": "1",
-         "username": "jdoe", "password": "hunter2"},
+        {
+            "uuid": "71da23fd-03f3-4e1f-a7e1-bb25645b981e",
+            "enabled": "1",
+            "username": "jdoe",
+            "password": "hunter2",
+        },
     ]
 }
 
@@ -1237,10 +1273,13 @@ class TestCreateRadiusClient:
 
     def test_validation_failure_does_not_reconfigure(self, driver):
         calls = []
-        driver._post = lambda path, data=None: calls.append(path) or {
-            "result": "failed",
-            "validations": {"client.name": "A value is required."},
-        }
+        driver._post = lambda path, data=None: (
+            calls.append(path)
+            or {
+                "result": "failed",
+                "validations": {"client.name": "A value is required."},
+            }
+        )
         result = driver.create_radius_client("", "10.0.0.5/32", "s3cr3t")
         assert result["success"] is False
         assert result["validations"] == {"client.name": "A value is required."}
@@ -1320,10 +1359,13 @@ class TestCreateRadiusUser:
 
     def test_validation_failure_does_not_reconfigure(self, driver):
         calls = []
-        driver._post = lambda path, data=None: calls.append(path) or {
-            "result": "failed",
-            "validations": {"user.username": "A value is required."},
-        }
+        driver._post = lambda path, data=None: (
+            calls.append(path)
+            or {
+                "result": "failed",
+                "validations": {"user.username": "A value is required."},
+            }
+        )
         result = driver.create_radius_user("", "hunter2")
         assert result["success"] is False
         assert calls == ["/api/freeradius/user/add_user"]
@@ -1358,6 +1400,7 @@ class TestDeleteRadiusUser:
 # _post()
 # ---------------------------------------------------------------------------
 
+
 class TestInternalPost:
     def test_raises_when_no_session(self):
         drv = OPNsenseDriver("host", "u", "p")
@@ -1393,6 +1436,7 @@ class TestInternalPost:
 # send_wake_on_lan()
 # ---------------------------------------------------------------------------
 
+
 class TestSendWakeOnLan:
     def test_raises_value_error_without_interface(self, driver):
         with pytest.raises(ValueError):
@@ -1413,7 +1457,10 @@ class TestSendWakeOnLan:
         # success — the underlying `wol` CLI tool typically prints nothing.
         driver.session.post.return_value = _make_json_response({"status": ""})
         result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
-        assert result == {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"}
+        assert result == {
+            "success": True,
+            "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan",
+        }
 
     def test_returns_failure_on_error_status(self, driver):
         driver.session.post.return_value = _make_json_response(
@@ -1441,10 +1488,12 @@ class TestSendWakeOnLan:
 # load_merge_candidate()
 # ---------------------------------------------------------------------------
 
-VALID_ROUTES_CONFIG = json.dumps([
-    {"network": "10.0.0.0/8", "gateway": "WAN_GW", "descr": "internal"},
-    {"network": "0.0.0.0/0", "gateway": "WAN_GW", "descr": "default"},
-])
+VALID_ROUTES_CONFIG = json.dumps(
+    [
+        {"network": "10.0.0.0/8", "gateway": "WAN_GW", "descr": "internal"},
+        {"network": "0.0.0.0/0", "gateway": "WAN_GW", "descr": "default"},
+    ]
+)
 
 
 class TestLoadMergeCandidate:
@@ -1467,38 +1516,45 @@ class TestLoadMergeCandidate:
 
     def test_raises_if_both_args_given(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         with pytest.raises(MergeConfigException):
             driver.load_merge_candidate(filename="f.json", config="{}")
 
     def test_raises_if_no_args_given(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         with pytest.raises(MergeConfigException):
             driver.load_merge_candidate()
 
     def test_raises_on_invalid_json(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         with pytest.raises(MergeConfigException, match="Invalid JSON"):
             driver.load_merge_candidate(config="not json {{{")
 
     def test_raises_if_not_a_list(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         with pytest.raises(MergeConfigException, match="array"):
             driver.load_merge_candidate(config='{"network": "1.0.0.0/8"}')
 
     def test_raises_if_route_missing_network(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         bad = json.dumps([{"gateway": "GW1"}])
         with pytest.raises(MergeConfigException, match="'network'"):
             driver.load_merge_candidate(config=bad)
 
     def test_raises_if_route_missing_gateway(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         bad = json.dumps([{"network": "10.0.0.0/8"}])
         with pytest.raises(MergeConfigException, match="'gateway'"):
             driver.load_merge_candidate(config=bad)
 
     def test_raises_on_missing_file(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         with pytest.raises(MergeConfigException, match="Cannot read"):
             driver.load_merge_candidate(filename="/nonexistent/path/routes.json")
 
@@ -1532,9 +1588,9 @@ class TestCompareConfig:
         assert "WAN_GW" in diff
 
     def test_empty_diff_when_config_matches(self, driver):
-        same_config = json.dumps([
-            {"network": "192.168.1.0/24", "gateway": "LAN_GW", "descr": "lan", "disabled": "0"}
-        ])
+        same_config = json.dumps(
+            [{"network": "192.168.1.0/24", "gateway": "LAN_GW", "descr": "lan", "disabled": "0"}]
+        )
         driver._get = lambda path: SEARCH_ROUTE_RESPONSE
         driver.load_merge_candidate(config=same_config)
         diff = driver.compare_config()
@@ -1547,7 +1603,11 @@ class TestCompareConfig:
 
 BACKUPS_RESPONSE = {
     "items": [
-        {"id": "config-opnsense01-1234567890.xml", "time": "1234567890", "description": "before change"},
+        {
+            "id": "config-opnsense01-1234567890.xml",
+            "time": "1234567890",
+            "description": "before change",
+        },
         {"id": "config-opnsense01-1234567800.xml", "time": "1234567800", "description": "initial"},
     ]
 }
@@ -1556,6 +1616,7 @@ BACKUPS_RESPONSE = {
 class TestCommitConfig:
     def test_raises_without_candidate(self, driver):
         from napalm.base.exceptions import MergeConfigException
+
         with pytest.raises(MergeConfigException, match="No candidate"):
             driver.commit_config()
 
@@ -1589,8 +1650,9 @@ class TestCommitConfig:
         assert driver._pre_commit_backup_id is None
 
     def test_raises_on_api_error(self, driver):
-        from napalm.base.exceptions import MergeConfigException
         import requests as _req
+        from napalm.base.exceptions import MergeConfigException
+
         driver._get = lambda path: BACKUPS_RESPONSE
         driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
         driver.session.post.side_effect = _req.exceptions.RequestException("timeout")
@@ -1602,6 +1664,7 @@ class TestCommitConfig:
 # discard_config()
 # ---------------------------------------------------------------------------
 
+
 class TestDiscardConfig:
     def test_clears_candidate(self, driver):
         driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
@@ -1617,6 +1680,7 @@ class TestDiscardConfig:
 # rollback()
 # ---------------------------------------------------------------------------
 
+
 class TestRollback:
     def test_noop_when_no_backups_and_no_commit(self, driver):
         driver._get = lambda path: {"items": []}
@@ -1655,6 +1719,7 @@ class TestRollback:
 # get_config() — candidate slot
 # ---------------------------------------------------------------------------
 
+
 class TestGetConfigCandidate:
     def test_candidate_empty_without_staged_config(self, driver):
         driver._get_text = lambda path: "<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)
@@ -1787,6 +1856,83 @@ class TestCreateDhcpReservation:
             == "https://opnsense.example.com/api/kea/dhcpv4/setReservation/existing-uuid-456"
         )
 
+    # A reconfigure stops and starts Kea (ApiMutableServiceControllerBase,
+    # reconfigureForceRestart() is true for Kea). Writing a reservation that
+    # is already there, and restarting the daemon for it, once left a
+    # firewall without DHCP (#1187) -- so an unchanged one is not touched.
+
+    @pytest.mark.parametrize(
+        "subnet",
+        ["172.22.8.0/24", "6854cab3-ebb5-4031-b987-0edcc6723546"],
+        ids=["subnet-as-cidr", "subnet-as-uuid"],
+    )
+    def test_unchanged_reservation_is_neither_written_nor_reconfigured(self, driver, subnet):
+        driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE)
+        driver.session.post.side_effect = [
+            _make_json_response(
+                {
+                    "rows": [
+                        {
+                            "uuid": "existing-uuid-456",
+                            "subnet": subnet,
+                            "ip_address": "172.22.8.253",
+                            # Kea keeps the MAC as it was entered -- case and
+                            # separator do not make it a different reservation.
+                            "hw_address": "02-AA-BB-CC-DD-EE",
+                            "hostname": "new-vm",
+                            "description": "[netork]",
+                        }
+                    ]
+                }
+            ),
+        ]
+
+        driver.create_dhcp_reservation(
+            mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253", hostname="new-vm"
+        )
+
+        # searchReservation only -- no setReservation, no service/reconfigure.
+        assert driver.session.post.call_count == 1
+
+    @pytest.mark.parametrize(
+        "field, live_value",
+        [
+            ("hw_address", "02:aa:bb:cc:dd:ff"),
+            ("hostname", "old-vm"),
+            ("description", "made by hand"),
+            ("subnet", "172.22.0.0/24"),
+        ],
+    )
+    def test_reservation_differing_in_one_field_is_written_and_reconfigured(
+        self, driver, field, live_value
+    ):
+        row = {
+            "uuid": "existing-uuid-456",
+            "subnet": "172.22.8.0/24",
+            "ip_address": "172.22.8.253",
+            "hw_address": "02:aa:bb:cc:dd:ee",
+            "hostname": "new-vm",
+            "description": "[netork]",
+        }
+        row[field] = live_value
+        driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE)
+        driver.session.post.side_effect = [
+            _make_json_response({"rows": [row]}),
+            _make_json_response({"result": "saved", "uuid": "existing-uuid-456"}),
+            _make_json_response({"status": "ok"}),
+        ]
+
+        driver.create_dhcp_reservation(
+            mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253", hostname="new-vm"
+        )
+
+        paths = [c.args[0] for c in driver.session.post.call_args_list]
+        assert paths == [
+            "https://opnsense.example.com/api/kea/dhcpv4/searchReservation",
+            "https://opnsense.example.com/api/kea/dhcpv4/setReservation/existing-uuid-456",
+            "https://opnsense.example.com/api/kea/service/reconfigure",
+        ]
+
     def test_raises_when_ip_not_in_any_kea_subnet(self, driver):
         driver.session.get.return_value = _make_json_response(KEA_SUBNETS_RESPONSE)
 
@@ -1857,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
@@ -1929,6 +2077,7 @@ class TestDeleteDhcpReservationAndLease:
 # apply_firewall_rule / commit_firewall_rules
 # ---------------------------------------------------------------------------
 
+
 class TestApplyFirewallRule:
     def test_add_calls_addRule_with_translated_fields(self, driver):
         calls = []
@@ -2279,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()
@@ -2414,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."""
@@ -2476,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,
@@ -2563,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"]
@@ -2588,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
@@ -2603,9 +2779,12 @@ class TestSyncDnsZone:
         rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="")]
         calls = self._record_calls(driver, rows)
 
-        driver.sync_dns_zone("office.local", [
-            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
-        ])
+        driver.sync_dns_zone(
+            "office.local",
+            [
+                {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+            ],
+        )
 
         assert [p for p, _ in calls if p.endswith("addhostoverride")] == []
         assert [p for p, _ in calls if "delhostoverride" in p] == []
@@ -2618,10 +2797,13 @@ class TestSyncDnsZone:
         ]
         calls = self._record_calls(driver, rows)
 
-        driver.sync_dns_zone("office.local", [
-            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
-            {"hostname": "NAS", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
-        ])
+        driver.sync_dns_zone(
+            "office.local",
+            [
+                {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+                {"hostname": "NAS", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+            ],
+        )
 
         assert [p.rsplit("/", 1)[-1] for p, _ in calls if "delhostoverride" in p] == ["copy"]
         assert [p for p, _ in calls if p.endswith("addhostoverride")] == []
@@ -2630,33 +2812,44 @@ class TestSyncDnsZone:
         rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="")]
         calls = self._record_calls(driver, rows)
 
-        driver.sync_dns_zone("office.local", [
-            {"hostname": "nas", "ip": "172.22.40.9", "record_type": "A", "enabled": True},
-        ])
+        driver.sync_dns_zone(
+            "office.local",
+            [
+                {"hostname": "nas", "ip": "172.22.40.9", "record_type": "A", "enabled": True},
+            ],
+        )
 
         added = [d for p, d in calls if p.endswith("addhostoverride")]
         assert [d["host"]["server"] for d in added] == ["172.22.40.9"]
 
     def test_a_disabled_hand_made_override_does_not_stand_in(self, driver):
         """It answers nothing, so the record still has to be added."""
-        rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="",
-                              enabled="0")]
+        rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="", enabled="0")]
         calls = self._record_calls(driver, rows)
 
-        driver.sync_dns_zone("office.local", [
-            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
-        ])
+        driver.sync_dns_zone(
+            "office.local",
+            [
+                {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+            ],
+        )
 
         assert len([p for p, _ in calls if p.endswith("addhostoverride")]) == 1
 
     def test_a_hand_made_override_in_another_zone_does_not_stand_in(self, driver):
-        rows = [_override_row("by-hand", host="nas", domain="other.local", ip="172.22.40.8",
-                              description="")]
+        rows = [
+            _override_row(
+                "by-hand", host="nas", domain="other.local", ip="172.22.40.8", description=""
+            )
+        ]
         calls = self._record_calls(driver, rows)
 
-        driver.sync_dns_zone("office.local", [
-            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
-        ])
+        driver.sync_dns_zone(
+            "office.local",
+            [
+                {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+            ],
+        )
 
         assert len([p for p, _ in calls if p.endswith("addhostoverride")]) == 1
 

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
@@ -28,11 +28,11 @@ class FakePingAPI:
         #: to assume one, which is why nothing caught the payload going to the
         #: wrong node.
         self.model_path = tuple(model_path)
-        self.created = []      # payloads passed to set
-        self.started = []      # job ids passed to start
-        self.stopped = []      # job ids passed to stop
-        self.removed = []      # job ids passed to remove
-        self.jobs = {}         # job id -> hostname
+        self.created = []  # payloads passed to set
+        self.started = []  # job ids passed to start
+        self.stopped = []  # job ids passed to stop
+        self.removed = []  # job ids passed to remove
+        self.jobs = {}  # job id -> hostname
         self.search_calls = 0
         self._next_id = 0
 

tests/unit/test_port_forwards.py

@@ -112,7 +112,10 @@ class TestWhereItGoes:
         assert entry["internal_ip"] == "10.0.0.9"
 
     def test_an_alias_with_two_hosts_is_two_forwards(self):
-        assert [e["internal_ip"] for e in _read(_rule(target="web_pair"))] == ["10.0.0.5", "10.0.0.6"]
+        assert [e["internal_ip"] for e in _read(_rule(target="web_pair"))] == [
+            "10.0.0.5",
+            "10.0.0.6",
+        ]
 
     def test_an_alias_that_names_a_host_by_dns_is_skipped(self):
         """No address to put the forward on; guessing one would be worse."""
@@ -129,7 +132,14 @@ class TestWhereItGoes:
 class TestPorts:
     @pytest.mark.parametrize(
         ("value", "expected"),
-        [("443", 443), ("https", 443), ("http", 80), ("postgres", 5432), ("8000-8010", 8000), ("8000:8010", 8000)],
+        [
+            ("443", 443),
+            ("https", 443),
+            ("http", 80),
+            ("postgres", 5432),
+            ("8000-8010", 8000),
+            ("8000:8010", 8000),
+        ],
     )
     def test_the_external_port(self, value, expected):
         (entry,) = _read(_rule(**{"destination.port": value}))
@@ -173,7 +183,7 @@ class TestNameAndSource:
         assert "remote_host" not in entry
 
     def test_an_inverted_source_has_no_remote_host(self):
-        """"Everyone but X" is as good as anyone for whether it is reachable."""
+        """ "Everyone but X" is as good as anyone for whether it is reachable."""
         (entry,) = _read(_rule(**{"source.network": "203.0.113.7", "source.not": "1"}))
         assert "remote_host" not in entry
 

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