feat(opnsense): add delete_dhcp_reservation_and_lease() for Kea DHCPv4

Combined removal of a static reservation and its active lease, needed by
NetOrk's VM-deletion cleanup flow. Reservation deletion follows the same
search-then-del<X>/{uuid} + reconfigure pattern as create_dhcp_reservation
and raises on failure; lease deletion is best-effort/non-fatal since the
Kea lease-delete endpoint shape is unverified against a real box.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

napalm_opnsense/opnsense.py

@@ -1208,6 +1208,79 @@ class OPNsenseDriver(FirewallDriver):
 
         self._post("/api/kea/service/reconfigure")
 
+    def delete_dhcp_reservation_and_lease(self, mac: str, ip: str) -> dict[str, Any]:
+        """Remove a Kea DHCPv4 static reservation and any active lease for (mac, ip).
+
+        Combined per NetOrk's VM-deletion cleanup flow — reservation and
+        lease removal are always requested together, so a single driver
+        call keeps callers from having to sequence two calls themselves.
+
+        Reservation removal follows the same search-then-act pattern as
+        ``create_dhcp_reservation`` (``searchReservation`` -> ``del<X>/{uuid}``
+        -> ``service/reconfigure``) and raises on failure, mirroring that
+        method's "never silently no-op on the thing the caller explicitly
+        asked for" contract.
+
+        Lease removal is best-effort and non-fatal: the exact Kea
+        lease-delete endpoint shape is unverified in this codebase (unlike
+        reservations, ``get_dhcp_leases()`` only ever implemented the read
+        path) — a failure here just means a stale lease record lingers in
+        Kea until its own natural cleanup, which is cosmetic, not a
+        functional problem (a deleted reservation already prevents the
+        client from getting the same IP back).
+
+        :param mac: NIC MAC address of the reservation to remove.
+        :param ip: IP address of the reservation/lease to remove.
+        :returns: {"reservation_found", "reservation_deleted", "lease_found",
+            "lease_deleted"} — all bool.
+        :raises RuntimeError: Kea plugin unavailable, or Kea rejects
+            deletion of a reservation that does exist.
+        """
+        result: dict[str, Any] = {
+            "reservation_found": False,
+            "reservation_deleted": False,
+            "lease_found": False,
+            "lease_deleted": False,
+        }
+
+        # --- Reservation ---
+        try:
+            existing = self._post(
+                "/api/kea/dhcpv4/searchReservation",
+                {"current": 1, "rowCount": -1, "searchPhrase": ip},
+            )
+        except Exception as exc:
+            raise RuntimeError(f"Kea DHCPv4 plugin unavailable: {exc}") from exc
+
+        reservation_uuid = None
+        for row in existing.get("rows") or []:
+            if row.get("ip_address") == ip:
+                reservation_uuid = row.get("uuid")
+                break
+
+        if reservation_uuid:
+            result["reservation_found"] = True
+            del_result = self._post(f"/api/kea/dhcpv4/delReservation/{reservation_uuid}")
+            if del_result.get("result") != "deleted":
+                raise RuntimeError(f"Kea rejected reservation delete for {ip}: {del_result}")
+            result["reservation_deleted"] = True
+            self._post("/api/kea/service/reconfigure")
+
+        # --- Lease (best-effort) ---
+        try:
+            leases = self._get("/api/kea/leases4/search")
+            rows = leases.get("rows") or leases.get("leases") or []
+            if any((row.get("address") or row.get("ip-address")) == ip for row in rows):
+                result["lease_found"] = True
+                del_lease = self._post("/api/kea/leases4/delLease", {"ip-address": ip})
+                result["lease_deleted"] = bool(
+                    del_lease.get("result") == "deleted" or del_lease.get("status") == "ok"
+                )
+        except Exception as exc:
+            logger.warning("DHCP lease delete for %s failed (non-fatal): %s", ip, exc)
+
+        return result
+
     def get_services(self) -> list[dict[str, Any]]:
         """Return running services from OPNsense.
 

tests/unit/test_driver.py

@@ -1340,3 +1340,100 @@ class TestCreateDhcpReservation:
             driver.create_dhcp_reservation(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")
 
         driver.session.post.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# delete_dhcp_reservation_and_lease()
+# ---------------------------------------------------------------------------
+
+
+class TestDeleteDhcpReservationAndLease:
+    def test_deletes_existing_reservation_and_lease(self, driver):
+        driver.session.post.side_effect = [
+            _make_json_response(
+                {"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]}
+            ),  # searchReservation — match
+            _make_json_response({"result": "deleted"}),  # delReservation
+            _make_json_response({"status": "ok"}),  # service/reconfigure
+            _make_json_response({"result": "deleted"}),  # delLease
+        ]
+        driver.session.get.return_value = _make_json_response(
+            {"rows": [{"address": "172.22.8.253", "hwaddr": "02:aa:bb:cc:dd:ee"}]}
+        )
+
+        result = driver.delete_dhcp_reservation_and_lease(
+            mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253"
+        )
+
+        assert result == {
+            "reservation_found": True,
+            "reservation_deleted": True,
+            "lease_found": True,
+            "lease_deleted": True,
+        }
+        del_res_call = driver.session.post.call_args_list[1]
+        assert (
+            del_res_call.args[0]
+            == "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"
+        del_lease_call = driver.session.post.call_args_list[3]
+        assert del_lease_call.args[0] == "https://opnsense.example.com/api/kea/leases4/delLease"
+        assert del_lease_call.kwargs["json"] == {"ip-address": "172.22.8.253"}
+
+    def test_noop_when_no_reservation_and_no_lease_found(self, driver):
+        driver.session.post.return_value = _make_json_response({"rows": []})
+        driver.session.get.return_value = _make_json_response({"rows": []})
+
+        result = driver.delete_dhcp_reservation_and_lease(
+            mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253"
+        )
+
+        assert result == {
+            "reservation_found": False,
+            "reservation_deleted": False,
+            "lease_found": False,
+            "lease_deleted": False,
+        }
+        # Only searchReservation was called — no delReservation/reconfigure/delLease.
+        assert driver.session.post.call_count == 1
+
+    def test_reservation_delete_raises_when_kea_rejects(self, driver):
+        driver.session.post.side_effect = [
+            _make_json_response(
+                {"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]}
+            ),
+            _make_json_response({"result": "not found"}),  # delReservation rejected
+        ]
+
+        with pytest.raises(RuntimeError, match="Kea rejected reservation delete"):
+            driver.delete_dhcp_reservation_and_lease(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")
+
+        # reconfigure must NOT be called after a rejected delete
+        assert driver.session.post.call_count == 2
+
+    def test_lease_delete_failure_is_non_fatal(self, driver):
+        driver.session.post.side_effect = [
+            _make_json_response(
+                {"rows": [{"uuid": "existing-uuid-456", "ip_address": "172.22.8.253"}]}
+            ),
+            _make_json_response({"result": "deleted"}),
+            _make_json_response({"status": "ok"}),
+        ]
+        driver.session.get.side_effect = Exception("connection reset")
+
+        result = driver.delete_dhcp_reservation_and_lease(
+            mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253"
+        )
+
+        assert result["reservation_found"] is True
+        assert result["reservation_deleted"] is True
+        assert result["lease_found"] is False
+        assert result["lease_deleted"] is False
+
+    def test_raises_when_kea_plugin_unavailable(self, driver):
+        driver.session.post.side_effect = Exception("404 Not Found")
+
+        with pytest.raises(RuntimeError, match="Kea DHCPv4 plugin unavailable"):
+            driver.delete_dhcp_reservation_and_lease(mac="02:aa:bb:cc:dd:ee", ip="172.22.8.253")