From 427a9ac8afd69b04ca8d515628b3ecdf21fb2bcd Mon Sep 17 00:00:00 2001
From: Christian Manivong <christian.manivong@gmail.com>
Date: Sun, 18 Oct 2026 15:29:02 +0200
Subject: [PATCH] feat: read the service clusters a host belongs to (0.10.0)

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.
---
 napalm_linux/linux.py |  9 +++++++++
 pyproject.toml        |  4 ++--
 tests/test_linux.py   | 38 ++++++++++++++++++++++++++++++++++++++
 3 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py
index 17ada90..015b8b4 100644
--- a/napalm_linux/linux.py
+++ b/napalm_linux/linux.py
@@ -46,6 +46,7 @@ from napalm_device_types import (
     OpenSshServerMixin,
     OSDriver,
     PackageProxyMixin,
+    ServiceClustersMixin,
     SnmpAgentMixin,
     SystemIdentityMixin,
     SystemdServicesMixin,
@@ -294,6 +295,7 @@ class LinuxDriver(
     KernelFactsMixin,
     ListeningSocketsMixin,
     VirtualIpsMixin,
+    ServiceClustersMixin,
     SystemIdentityMixin,
     OpenSshServerMixin,
     SystemdServicesMixin,
@@ -1179,6 +1181,13 @@ class LinuxDriver(
         """
         return self.run_command(command, privileged=privileged, timeout=60).stdout
 
+    def _run_service_clusters_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ServiceClustersMixin.get_service_clusters``: an exec
+        channel, like the virtual IPs' command, which it resembles in size.
+        Read-only either way; root only because Patroni's configuration is the
+        postgres user's alone."""
+        return self.run_command(command, privileged=privileged, timeout=60).stdout
+
     def _run_system_identity_command(self, command: str, *, privileged: bool) -> str:
         """The transport for ``SystemIdentityMixin.get_system_identity``: an exec channel,
         as root, since ``product_uuid`` and ``product_serial`` are root's."""
diff --git a/pyproject.toml b/pyproject.toml
index 39fadae..43f6210 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.9.0"
+version = "0.10.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.14.0",
+    "napalm-device-types>=4.17.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]
diff --git a/tests/test_linux.py b/tests/test_linux.py
index 33c4876..fbd07ba 100644
--- a/tests/test_linux.py
+++ b/tests/test_linux.py
@@ -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