feat: an HA pair's DHCP peers and config sync as facts (4.11.0)
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 24s
CI / test (3.12) (push) Successful in 27s
CI / test (3.13) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 26s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 27s
CI / test (3.13) (pull_request) Successful in 25s

get_dhcp_ha (declared on DhcpServerMixin, optional) is what a DHCP
server's own HA configuration says: on or off, this server's peer name,
each peer's name, role as the server words it, and HA URL.
ConfigSyncMixin.get_config_sync is where a host pushes its configuration
and for which services, by the names get_services reports; sync items
without a service name stay visible under `unmapped`. A driver mixes it
in, no role promises it. Both raise when they could not read.

What the facts mean for a pair -- hot standby, which side to write to --
is the caller's (NetOrk/netork#1125).
This commit is contained in:
christianmanivong committed 2026-10-16 15:18:25 +02:00
1 parent 2739f10ecd
commit 90367c1370
7 files changed
+183 -1

No files matched your search

+70
View File
@@ -0,0 +1,70 @@
# -*- coding: utf-8 -*-
"""Two facts an HA pair of servers reports about itself.
`get_dhcp_ha` is what a DHCP server's own HA configuration says: the peers it
pairs with and each peer's role, as the server words it. `get_config_sync` is
where a host pushes its configuration and for which of its services. Both are
declarations; what they mean for the pair -- hot standby, who to write to --
is the caller's to work out.
"""
from __future__ import annotations
import inspect
import napalm_device_types as package
from napalm_device_types import ConfigSyncMixin, DhcpServerMixin, FirewallDriver
class TestDeclaredOnly:
def test_get_dhcp_ha_is_absent_until_a_driver_implements_it(self):
assert not hasattr(DhcpServerMixin, "get_dhcp_ha")
def test_get_config_sync_is_absent_until_a_driver_implements_it(self):
assert not hasattr(ConfigSyncMixin, "get_config_sync")
def test_config_sync_mixin_owns_nothing_callable(self):
own = [
name
for name, value in vars(ConfigSyncMixin).items()
if not name.startswith("__") and inspect.isfunction(value)
]
assert own == []
def test_the_firewall_role_does_not_carry_config_sync(self):
"""A driver mixes the capability in; the role does not promise it."""
assert not issubclass(FirewallDriver, ConfigSyncMixin)
def test_a_driver_that_implements_them_is_found_by_hasattr(self):
class Pair(ConfigSyncMixin, DhcpServerMixin):
def get_dhcp_ha(self):
return {"enabled": False, "peers": []}
def get_config_sync(self):
return {"peer": None, "services": [], "unmapped": []}
assert callable(getattr(Pair(), "get_dhcp_ha", None))
assert callable(getattr(Pair(), "get_config_sync", None))
class TestTypes:
def test_are_exported(self):
for name in ("ConfigSyncMixin", "ConfigSyncDict", "DhcpHaDict", "DhcpHaPeerDict"):
assert name in package.__all__ and hasattr(package, name), name
def test_a_peer_needs_only_its_name_and_role(self):
"""A future server may not have a URL to report; Kea does."""
from napalm_device_types import DhcpHaPeerDict
assert DhcpHaPeerDict.__required_keys__ == frozenset({"name", "role"})
def test_ha_needs_only_whether_it_is_on_and_the_peers(self):
from napalm_device_types import DhcpHaDict
assert DhcpHaDict.__required_keys__ == frozenset({"enabled", "peers"})
def test_config_sync_reports_what_it_could_not_name(self):
"""A sync item the driver has no service name for is still a fact."""
from napalm_device_types import ConfigSyncDict
assert ConfigSyncDict.__required_keys__ == frozenset({"peer", "services", "unmapped"})