feat(dhcp): add generic subnet diff/apply to DhcpServerMixin

Part of netork#85. Reservations were the only DHCP desired state the mixin knew about; this adds the layer above them — the ranges a device serves and the options it publishes with them. The identity is the CIDR, matched rather than compared, the way `mac` is for a reservation. normalize_cidr deliberately does not rewrite the network address: turning 10.10.20.5/24 into 10.10.20.0/24 would make a typo silently match a real subnet and then apply that caller's pools and options to it. option_data is compared per option, and only over the options the caller named. An absent key means "not managed", not "should be empty" — without that rule a caller managing only domain_search would diff against every option the server autocollects (routers, domain_name_servers, ntp_servers) and reconfigure the DHCP daemon on every single run. Neither diff deletes. For subnets that is not merely conservative: removing one takes DHCP down for a whole VLAN, and the diff cannot tell "no longer wanted" from "was never this caller's to describe". commit_dhcp_subnets is separate from commit_dhcp_reservations even where a driver implements both with the same call — the two desired-state sets are applied independently, and a caller that changed only subnets should not have to know which reload the vendor happens to share. 19 new tests against an in-memory fake; no vendor driver needed.
2026-08-20 07:21:40 +07:00
parent b8977cdaa5
commit b53cf4d1f4
4 changed files with 479 additions and 9 deletions
@@ -0,0 +1,228 @@
+"""Tests for DhcpServerMixin's generic subnet diff/apply mechanism.
+
+Same shape as test_dhcp_diff_apply.py: diff_dhcp_subnets/apply_dhcp_subnetset
+are concrete methods that only depend on the abstract trio
+(get_dhcp_subnets/apply_dhcp_subnet/commit_dhcp_subnets), so an in-memory
+fake exercises them fully. See README.md "Design principle: generic vs.
+device-specific logic".
+
+A subnet is a heavier object than a reservation: a wrong `pools` or
+`option_data` takes a whole VLAN offline rather than one host, so the tests
+below lean on the never-delete and preserve-unmanaged-options guarantees.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pytest
+from napalm_device_types import DhcpServerMixin
+from napalm_device_types.models import DhcpSubnetDict
+
+
+def _subnet(**overrides: Any) -> DhcpSubnetDict:
+    base: DhcpSubnetDict = {
+        "uuid": "",
+        "subnet": "10.10.20.0/24",
+        "description": "Home",
+        "pools": ["10.10.20.100-10.10.20.200"],
+        "option_data": {
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+        },
+        "match_client_id": False,
+    }
+    base.update(overrides)  # type: ignore[typeddict-item]
+    return base
+
+
+class _FakeDhcpServer(DhcpServerMixin):
+    def __init__(self, live: Optional[List[DhcpSubnetDict]] = None) -> None:
+        self.live = live or []
+        self.applied: List[Dict[str, Any]] = []
+        self.commits = 0
+
+    def get_dhcp_subnets(self) -> List[DhcpSubnetDict]:
+        return self.live
+
+    def apply_dhcp_subnet(
+        self, subnet: DhcpSubnetDict, *, uuid: Optional[str] = None
+    ) -> Dict[str, Any]:
+        self.applied.append({"subnet": subnet, "uuid": uuid})
+        return {"success": True}
+
+    def commit_dhcp_subnets(self) -> Dict[str, Any]:
+        self.commits += 1
+        return {"success": True}
+
+
+# ── diff: adds ────────────────────────────────────────────────────────────────
+
+
+def test_unknown_subnet_is_an_add() -> None:
+    diff = _FakeDhcpServer().diff_dhcp_subnets([_subnet()])
+    assert len(diff["add"]) == 1
+    assert diff["add"][0]["subnet"] == "10.10.20.0/24"
+    assert diff["update"] == []
+
+
+def test_matching_subnet_with_no_changes_is_neither() -> None:
+    live = _subnet(uuid="dev-1")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert diff == {"add": [], "update": []}
+
+
+# ── diff: identity is the CIDR ────────────────────────────────────────────────
+
+
+def test_subnets_are_matched_on_cidr() -> None:
+    live = _subnet(uuid="dev-1", description="renamed on the device")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert diff["add"] == []
+    assert diff["update"][0]["uuid"] == "dev-1"
+    assert diff["update"][0]["changed_fields"] == ["description"]
+
+
+def test_a_different_cidr_is_a_new_subnet_not_an_update() -> None:
+    live = _subnet(uuid="dev-1", subnet="10.10.20.0/24")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet(subnet="10.10.30.0/24")])
+    assert len(diff["add"]) == 1
+    assert diff["update"] == []
+
+
+# ── diff: compared fields ─────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("description", "Office"),
+        ("pools", ["10.10.20.50-10.10.20.99"]),
+        ("match_client_id", True),
+    ],
+)
+def test_changed_field_is_reported(field: str, value: Any) -> None:
+    live = _subnet(uuid="dev-1")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet(**{field: value})])
+    assert diff["update"][0]["changed_fields"] == [field]
+
+
+def test_changed_option_is_reported_as_option_data() -> None:
+    live = _subnet(uuid="dev-1")
+    desired = _subnet(
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "domain_search": ["home.local", "office.local"],
+        }
+    )
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+
+def test_an_unmanaged_option_on_the_device_is_not_a_change() -> None:
+    """Absent key means "not managed" -- it must not provoke an update.
+
+    Kea autocollects routers/domain_name_servers/ntp_servers. A caller that
+    only wants to set domain_search would otherwise diff-fight the server
+    forever, reloading the DHCP daemon on every run.
+    """
+    live = _subnet(
+        uuid="dev-1",
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "ntp_servers": ["10.10.20.1"],
+        },
+    )
+    desired = _subnet(option_data={"domain_search": ["home.local"]})
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+    # ...and once it matches, it stays quiet.
+    live2 = _subnet(
+        uuid="dev-1",
+        option_data={
+            "routers": ["10.10.20.1"],
+            "domain_name_servers": ["10.10.20.1"],
+            "ntp_servers": ["10.10.20.1"],
+            "domain_search": ["home.local"],
+        },
+    )
+    assert _FakeDhcpServer([live2]).diff_dhcp_subnets([desired]) == {"add": [], "update": []}
+
+
+def test_option_order_is_significant_for_domain_search() -> None:
+    """Search order decides which zone answers an unqualified name first."""
+    live = _subnet(uuid="dev-1", option_data={"domain_search": ["a.local", "b.local"]})
+    desired = _subnet(option_data={"domain_search": ["b.local", "a.local"]})
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([desired])
+    assert diff["update"][0]["changed_fields"] == ["option_data"]
+
+
+# ── diff: never delete ────────────────────────────────────────────────────────
+
+
+def test_live_subnet_absent_from_desired_is_never_deleted() -> None:
+    """Deleting a subnet takes a whole VLAN's DHCP down. Never implicit."""
+    live = _subnet(uuid="dev-1", subnet="10.10.99.0/24")
+    diff = _FakeDhcpServer([live]).diff_dhcp_subnets([_subnet()])
+    assert len(diff["add"]) == 1
+    assert diff["update"] == []
+    assert "delete" not in diff
+
+
+# ── apply ─────────────────────────────────────────────────────────────────────
+
+
+def test_apply_creates_then_commits() -> None:
+    fake = _FakeDhcpServer()
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert fake.applied[0]["uuid"] is None
+    assert fake.commits == 1
+    assert any(line.startswith("[add]") for line in lines)
+
+
+def test_apply_updates_in_place_with_the_device_uuid() -> None:
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    list(fake.apply_dhcp_subnetset([_subnet(description="Office")]))
+    assert fake.applied[0]["uuid"] == "dev-1"
+    assert fake.commits == 1
+
+
+def test_apply_does_not_commit_when_nothing_changed() -> None:
+    """A commit reconfigures the DHCP daemon -- too costly for a no-op run."""
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert fake.commits == 0
+    assert lines == ["[commit] no changes"]
+
+
+def test_apply_names_the_subnet_in_its_progress_line() -> None:
+    fake = _FakeDhcpServer()
+    lines = list(fake.apply_dhcp_subnetset([_subnet()]))
+    assert "10.10.20.0/24" in lines[0]
+
+
+def test_apply_reports_changed_fields_on_update() -> None:
+    fake = _FakeDhcpServer([_subnet(uuid="dev-1")])
+    lines = list(fake.apply_dhcp_subnetset([_subnet(description="Office")]))
+    assert "description" in lines[0]
+
+
+# ── abstract surface ──────────────────────────────────────────────────────────
+
+
+class _BareDriver(DhcpServerMixin):
+    pass
+
+
+@pytest.mark.parametrize(
+    "call",
+    [
+        lambda d: d.get_dhcp_subnets(),
+        lambda d: d.apply_dhcp_subnet({}),  # type: ignore[typeddict-item]
+        lambda d: d.commit_dhcp_subnets(),
+    ],
+)
+def test_device_specific_methods_raise_not_implemented(call: Any) -> None:
+    with pytest.raises(NotImplementedError):
+        call(_BareDriver())