Christian Manivong
e334dc0058
CI / test (3.10) (push) Successful in 32s
CI / test (3.11) (push) Successful in 33s
CI / test (3.12) (push) Successful in 34s
CI / test (3.10) (pull_request) Successful in 38s
CI / test (3.11) (pull_request) Successful in 37s
CI / test (3.12) (pull_request) Successful in 38s
get_dhcp_ha reads kea/dhcpv4/get (the ha section) and kea/dhcpv4/searchPeer: whether Kea's HA hook is on, this firewall's peer name -- the % twin when left at its default -- and each peer's name, role and HA URL. get_config_sync reads core/hasync/get: the address this firewall pushes its configuration to and the ticked sync items, mapped to the service names get_services reports; items that are no service stay under `unmapped`. Both raise when the request fails. Configuration, not state: Kea's control agent, which knows whether the pair is in hot standby right now, listens on localhost only. Needs napalm-device-types 4.11.0 (ConfigSyncMixin). NetOrk/netork#1125.
140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
"""Kea's HA peers and the firewall's configuration sync, as facts (netOrk #1125).
|
|
|
|
The fixture is what gw-01 and gw-02 answered on 2026-10-16: Kea in hot standby
|
|
(gw primary, gw-02 standby) and gw-01 pushing its configuration, Kea and
|
|
Unbound included, to gw-02.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from napalm_opnsense.ha import config_sync, dhcp_ha
|
|
from napalm_opnsense.opnsense import OPNsenseDriver
|
|
|
|
PAIR = json.loads((Path(__file__).parent / "fixtures" / "ha_pair.json").read_text())
|
|
|
|
|
|
@pytest.fixture
|
|
def driver() -> OPNsenseDriver:
|
|
d = OPNsenseDriver("gw-01", "user", "secret")
|
|
d._session = MagicMock()
|
|
return d
|
|
|
|
|
|
def _answers(node: str):
|
|
def _get(path: str):
|
|
return {
|
|
"/api/kea/dhcpv4/get": PAIR[node]["kea_dhcpv4_get"],
|
|
"/api/kea/dhcpv4/searchPeer": PAIR[node]["kea_dhcpv4_searchPeer"],
|
|
"/api/core/hasync/get": PAIR[node]["core_hasync_get"],
|
|
}[path]
|
|
|
|
return _get
|
|
|
|
|
|
class TestDhcpHa:
|
|
def test_reads_the_peers_with_their_roles(self, driver):
|
|
driver._get = _answers("gw-01")
|
|
|
|
ha = driver.get_dhcp_ha()
|
|
|
|
assert ha["service"] == "kea-dhcp"
|
|
assert ha["enabled"] is True
|
|
assert ha["peers"] == [
|
|
{"name": "gw", "role": "primary", "url": "http://10.5.100.2:8001/"},
|
|
{"name": "gw-02", "role": "standby", "url": "http://10.5.100.3:8001/"},
|
|
]
|
|
|
|
def test_names_this_server_from_the_effective_value(self, driver):
|
|
"""`this_server_name` is empty when left at its default -- the
|
|
hostname; the effective name is in its `%` twin."""
|
|
driver._get = _answers("gw-02")
|
|
|
|
assert driver.get_dhcp_ha()["this_server"] == "gw-02"
|
|
|
|
def test_an_explicit_server_name_wins(self):
|
|
ha = dhcp_ha(
|
|
{"dhcpv4": {"ha": {"enabled": "1", "this_server_name": "left", "%this_server_name": "x"}}},
|
|
{"rows": []},
|
|
)
|
|
assert ha["this_server"] == "left"
|
|
|
|
def test_ha_switched_off(self):
|
|
ha = dhcp_ha({"dhcpv4": {"ha": {"enabled": "0", "this_server_name": ""}}}, {"rows": []})
|
|
|
|
assert ha["enabled"] is False
|
|
assert ha["peers"] == []
|
|
assert "this_server" not in ha
|
|
|
|
def test_a_peer_without_url_has_no_url_key(self):
|
|
ha = dhcp_ha(
|
|
{"dhcpv4": {"ha": {"enabled": "1"}}},
|
|
{"rows": [{"name": "a", "role": "primary", "url": ""}]},
|
|
)
|
|
assert ha["peers"] == [{"name": "a", "role": "primary"}]
|
|
|
|
def test_a_failed_read_raises_and_never_says_off(self, driver):
|
|
driver._get = MagicMock(side_effect=RuntimeError("timeout"))
|
|
|
|
with pytest.raises(RuntimeError):
|
|
driver.get_dhcp_ha()
|
|
|
|
def test_an_answer_without_the_ha_section_raises(self):
|
|
with pytest.raises(ValueError):
|
|
dhcp_ha({"dhcpv4": {}}, {"rows": []})
|
|
|
|
|
|
class TestConfigSync:
|
|
def test_the_pushing_side_names_its_peer_and_services(self, driver):
|
|
driver._get = _answers("gw-01")
|
|
|
|
sync = driver.get_config_sync()
|
|
|
|
assert sync["peer"] == "10.5.100.3"
|
|
assert {"kea-dhcp", "unbound", "ntpd", "openssh", "wireguard"} <= set(sync["services"])
|
|
assert sync["services"] == sorted(sync["services"])
|
|
|
|
def test_items_without_a_service_stay_visible(self, driver):
|
|
driver._get = _answers("gw-01")
|
|
|
|
unmapped = driver.get_config_sync()["unmapped"]
|
|
|
|
assert {"aliases", "rules", "virtualip"} <= set(unmapped)
|
|
assert "kea" not in unmapped
|
|
|
|
def test_only_selected_items_count(self, driver):
|
|
driver._get = _answers("gw-01")
|
|
|
|
sync = driver.get_config_sync()
|
|
|
|
assert "dnsmasq" not in sync["services"] # dnsforwarder: not selected
|
|
assert "cron" not in sync["unmapped"] and "cron" not in sync["services"]
|
|
|
|
def test_the_receiving_side_pushes_to_no_one(self, driver):
|
|
driver._get = _answers("gw-02")
|
|
|
|
sync = driver.get_config_sync()
|
|
|
|
assert sync == {"peer": None, "services": [], "unmapped": []}
|
|
|
|
def test_without_a_peer_selected_items_are_not_pushed(self):
|
|
"""Ticked items without a target address push nothing."""
|
|
answer = {"hasync": {"synchronizetoip": "", "syncitems": {"kea": {"value": "Kea", "selected": 1}}}}
|
|
|
|
assert config_sync(answer) == {"peer": None, "services": [], "unmapped": []}
|
|
|
|
def test_a_failed_read_raises(self, driver):
|
|
driver._get = MagicMock(side_effect=RuntimeError("timeout"))
|
|
|
|
with pytest.raises(RuntimeError):
|
|
driver.get_config_sync()
|
|
|
|
def test_an_answer_without_the_section_raises(self):
|
|
with pytest.raises(ValueError):
|
|
config_sync({"something": {}})
|