diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 2a7720b..623e35a 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -1317,7 +1317,16 @@ class OPNsenseDriver( 2. **ISC DHCP** (legacy) — ``POST /api/dhcpv4/leases/searchlease`` Fields: ``mac``, ``address``, ``hostname``, ``ends`` 3. **ARP table** fallback — ``GET /api/diagnostics/interface/get_arp`` - Provides IP only, hostname will be empty. + Provides IP only, hostname will be empty, ``state`` is ``"arp"``. + + The first source with leases wins. Without any, the ARP table stands + in -- also when Kea or ISC answered empty, since a box can hand out + addresses through a service neither covers. An empty list is a real + answer only: Kea or ISC answered with no leases and the fallback added + none. When neither answered and the fallback gives no rows (it failed, + or the table is empty), there is no answer to give, and this raises + :class:`RuntimeError` naming each failure -- a caller would otherwise + take ``[]`` for a server without leases and drop the ones it knows. Each returned entry contains: @@ -1350,6 +1359,9 @@ class OPNsenseDriver( }) return result + failures: list[str] = [] + answered = False # a lease source answered, with or without leases + # 1. Kea DHCPv4 plugin try: data = self._get("/api/kea/leases4/search") @@ -1357,8 +1369,10 @@ class OPNsenseDriver( leases = _parse_kea(rows) if leases: return leases + answered = True except Exception as exc: logger.debug("Kea DHCP lease fetch failed: %s", exc) + failures.append(f"Kea: {exc}") # 2. ISC DHCP (legacy) try: @@ -1369,20 +1383,27 @@ class OPNsenseDriver( leases = _parse_kea(data.get("rows", [])) if leases: return leases + answered = True except Exception as exc: logger.debug("ISC DHCP lease fetch failed: %s", exc) + failures.append(f"ISC DHCP: {exc}") # 3. ARP table fallback (IP only, no hostname) + fallback: list[dict[str, Any]] = [] try: - arp = self.get_arp_table() - return [ + fallback = [ {"mac": e["mac"], "ip": e["ip"], "hostname": "", "lease_end": 0, "state": "arp"} - for e in arp + for e in self.get_arp_table() if e.get("mac") and e.get("ip") ] + if not fallback: + failures.append("ARP table: no entries") except Exception as exc: logger.warning("ARP table fallback failed: %s", exc) - return [] + failures.append(f"ARP table: {exc}") + if fallback or answered: + return fallback + raise RuntimeError("No DHCP lease source answered -- " + "; ".join(failures)) def create_dhcp_reservation(self, mac: str, ip: str, hostname: str = "") -> None: """Create (or update) a Kea DHCPv4 static reservation (MAC → IP). diff --git a/tests/unit/test_dhcp_leases.py b/tests/unit/test_dhcp_leases.py new file mode 100644 index 0000000..e837173 --- /dev/null +++ b/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()