README.md

@@ -128,6 +128,17 @@ A mechanism it could not read completely is `None` ("did not look"), a host with
 `[]`. The command is about 4 kB, so a driver sends it on an exec channel
 (`_run_virtual_ips_command(command, privileged=)`), not typed into a shell.
 
+`get_dhcp_ha` (declared on `DhcpServerMixin`, optional) is what a DHCP server's own HA
+configuration says: whether HA is on, which peer this server is, each peer's name, role --
+as the server words it, not a shared vocabulary -- and, where the server has one, the URL
+its HA listener answers on. It is configuration, not state: whether the pair is in hot
+standby right now is not in it. `ConfigSyncMixin` (`get_config_sync`) is where a host
+pushes its configuration (`peer`, None for no one) and for which services, by the names
+`get_services` reports; sync items the driver has no service name for stay visible under
+`unmapped`. A driver mixes it in -- no role promises it. Both getters raise when they could
+not read; neither answers "off" for "could not ask". What the facts mean for a pair (hot
+standby, which side to write to) is the caller's to work out.
+
 A firewall's CARP addresses are a third mechanism, `carp`: a firewall driver reads them from
 its own API and reports them under the reading's `carp` key (`vip_manager` and `keepalived`
 then None). What a CARP state means is protocol knowledge and stays in that driver. The

napalm_device_types/__init__.py

@@ -37,6 +37,7 @@ instead of being restated on every role that happens to need it:
 
 * :class:`~napalm_device_types.channel.CommandChannelMixin`
 * :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin`
+* :class:`~napalm_device_types.config_sync.ConfigSyncMixin`
 * :class:`~napalm_device_types.container_engine.ContainerEngineMixin`
 * :class:`~napalm_device_types.dhcp.DhcpServerMixin`
 * :class:`~napalm_device_types.firewall_rules.FirewallRuleMixin`
@@ -71,6 +72,7 @@ from napalm_device_types.channel import (
     run_on_transport,
 )
 from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
+from napalm_device_types.config_sync import ConfigSyncMixin
 from napalm_device_types.container_engine import (
     ContainerEngineConnection,
     ContainerEngineMixin,
@@ -97,6 +99,9 @@ from napalm_device_types.management_controller import ManagementControllerDriver
 from napalm_device_types.media import MediaDriver
 from napalm_device_types.models import (
     POWER_ACTIONS,
+    ConfigSyncDict,
+    DhcpHaDict,
+    DhcpHaPeerDict,
     EventLogEntryDict,
     FirmwareDict,
     HardwareInventoryDict,
@@ -159,6 +164,10 @@ __all__ = [
     "open_stream_on_transport",
     "run_on_transport",
     "ConfigLifecycleMixin",
+    "ConfigSyncDict",
+    "ConfigSyncMixin",
+    "DhcpHaDict",
+    "DhcpHaPeerDict",
     "DeviceTypeDriver",
     "DhcpServerMixin",
     "FingerprintRule",

napalm_device_types/config_sync.py

@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+"""Configuration sync between the two hosts of an HA pair.
+
+A host that pushes its configuration to a peer (OPNsense's XMLRPC sync, say)
+makes the peer's configuration of the synced services a copy of its own: a
+change written on the receiving side lasts only until the next push. Who pushes
+what is therefore a fact a caller needs before it writes to either host.
+
+Declared under ``if TYPE_CHECKING``: a contract, not a placeholder. A driver
+mixes it in; no role promises it.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from napalm_device_types.models import ConfigSyncDict
+
+
+class ConfigSyncMixin:
+    """Where a host pushes its configuration, and for which of its services."""
+
+    if TYPE_CHECKING:
+
+        def get_config_sync(self) -> ConfigSyncDict:
+            """
+            Returns the peer this host pushes its configuration to (None when it
+            pushes to no one) and the services whose configuration it pushes, by
+            the name ``get_services`` reports. Sync items the driver has no
+            service name for are listed under ``unmapped`` as the host names them.
+
+            :raises Exception: When the setting could not be read -- never
+                ``peer=None`` for "could not ask".
+            """
+            ...

napalm_device_types/dhcp.py

@@ -28,6 +28,7 @@ from typing import Any, Dict, Iterator, List, Optional, TYPE_CHECKING
 
 from napalm_device_types._matching import check_desired, index_live
 from napalm_device_types.models import (
+    DhcpHaDict,
     DhcpReservationDiffDict,
     DhcpReservationDict,
     DhcpReservationUpdateDict,
@@ -219,6 +220,20 @@ class DhcpServerMixin:
             """
             ...
 
+        def get_dhcp_ha(self) -> DhcpHaDict:
+            """
+            Returns what the server's own HA configuration declares: whether HA
+            is switched on, which peer this server is, and each peer with its
+            role as the server words it.
+
+            Optional: a driver without it has no ``get_dhcp_ha`` at all, and a
+            caller asks ``callable(getattr(driver, "get_dhcp_ha", None))``.
+
+            :raises Exception: When the configuration could not be read -- never
+                ``enabled=False`` for "could not ask".
+            """
+            ...
+
     # ------------------------------------------------------------------
     # Generic, vendor-neutral algorithms.
     # ------------------------------------------------------------------

napalm_device_types/models.py

@@ -600,6 +600,48 @@ class DhcpSubnetDiffDict(TypedDict):
     update: List[DhcpSubnetUpdateDict]
 
 
+class DhcpHaPeerDict(TypedDict):
+    """One server of a DHCP HA relationship, as the server's configuration names it."""
+
+    name: str
+    #: The peer's role as the server words it ("primary", "standby", ...) -- data,
+    #: not a vocabulary every server shares.
+    role: str
+    #: Where the peer's HA listener answers, when the server addresses it by URL.
+    url: NotRequired[str]
+
+
+class DhcpHaDict(TypedDict):
+    """What a DHCP server's own HA configuration declares.
+
+    Configured, not observed: whether the pair is in hot standby right now, or
+    one side has taken over, is the server's state and not in here.
+    """
+
+    #: The name of the service this configuration belongs to, as ``get_services``
+    #: reports it -- the caller matches it to that service.
+    service: NotRequired[str]
+    enabled: bool
+    #: Which of ``peers`` is this server.
+    this_server: NotRequired[str]
+    peers: List[DhcpHaPeerDict]
+
+
+class ConfigSyncDict(TypedDict):
+    """Where a host pushes its configuration, and whose.
+
+    ``peer`` None: the host pushes to no one (the receiving side of a pair, or a
+    host without the feature switched on).
+    """
+
+    peer: Optional[str]
+    #: The services whose configuration it pushes, by the name ``get_services``
+    #: reports.
+    services: List[str]
+    #: Sync items the driver has no service name for, as the host names them.
+    unmapped: List[str]
+
+
 class VPNTunnelDict(TypedDict):
     type: str
     local_endpoint: str

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "4.10.1"
+version = "4.11.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_dhcp_ha_config_sync.py

@@ -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"})