feat: read the service clusters a host belongs to (0.10.0)
CI / test (3.10) (pull_request) Successful in 25s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 26s
CI / test (3.10) (push) Successful in 25s
CI / test (3.11) (push) Successful in 24s
CI / test (3.12) (push) Successful in 25s

LinuxDriver mixes in napalm-device-types' ServiceClustersMixin (4.17.0):
get_service_clusters() reads Patroni's members and roles, adguardhome-sync's
origin and replicas and a BunkerWeb worker's manager from the host's own
configuration -- the evidence that two hosts run one service, not merely
answer at one address (NetOrk/netork#1125). _run_service_clusters_command
carries the shared command on an exec channel, root first, like the virtual
IPs' one.
This commit is contained in:
Christian Manivong committed 2026-10-18 15:29:02 +02:00
1 parent c8e2b328d7
commit 427a9ac8af
3 files changed
+49 -2

No files matched your search

+38
View File
@@ -1721,6 +1721,44 @@ class TestGetVirtualIps:
assert plain.startswith("sh -c '")
# ---------------------------------------------------------------------------
# get_service_clusters (napalm-device-types ServiceClustersMixin)
# ---------------------------------------------------------------------------
_CLUSTERS = (
"SVCC_BEGIN\n[systemd]\n[patroni patroni.service]\nactive active\nscope netork-db\n"
"name db-01\nrestapi 10.7.234.12:8008\nrest-unreachable\n[no-docker]\nSVCC_END\n"
)
class TestGetServiceClusters:
"""The same template as get_virtual_ips: an exec channel, root first."""
_exec = staticmethod(TestGetVirtualIps._exec)
def test_it_reads_with_the_shared_command_on_an_exec_channel(self, driver):
from napalm_device_types import ServiceClustersMixin
assert isinstance(driver, ServiceClustersMixin)
calls = self._exec(driver, (_CLUSTERS, "", 0))
reading = driver.get_service_clusters()
assert reading["patroni"][0]["cluster"] == "netork-db"
[(command, privileged)] = calls
assert privileged is True
assert command.startswith("sh -c '")
driver._device.send_command.assert_not_called()
def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
calls = self._exec(driver, ("", "sudo: a password is required\n", 1), (_CLUSTERS, "", 0))
reading = driver.get_service_clusters()
assert reading["bunkerweb"] == []
assert [privileged for _, privileged in calls] == [True, False]
class TestHardensItsSshServer:
"""A stock Linux host runs a stock OpenSSH: the hardening netOrk offers for
ssh-audit findings comes from napalm-device-types, carried over run_command