feat: get_management_peer, the address the host sees the SSH connection come from
CI / test (3.10) (push) Failing after 16s
CI / test (3.11) (push) Failing after 14s
CI / test (3.12) (push) Failing after 16s
CI / test (3.10) (pull_request) Failing after 15s
CI / test (3.11) (pull_request) Failing after 14s
CI / test (3.12) (pull_request) Failing after 15s

From SSH_CONNECTION's client side; None when it is empty or no address.
The caller uses it to choose the sources an SNMP agent answers.
This commit is contained in:
Christian Manivong committed 2026-10-16 15:19:04 +02:00
1 parent 1bc54b66e8
commit ec8668476a
2 files changed
+28

No files matched your search

+8
View File
@@ -1999,6 +1999,14 @@ class LinuxDriver(
output = f"{output}\n{_SUDO_PASSWORD_HINT}" output = f"{output}\n{_SUDO_PASSWORD_HINT}"
return {"ok": status == 0, "output": (output.strip()[-300:] if status != 0 else done)} return {"ok": status == 0, "output": (output.strip()[-300:] if status != 0 else done)}
def get_management_peer(self) -> str | None:
"""The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
out = self.run_command('printf "%s\\n" "$SSH_CONNECTION"', timeout=15).stdout.split()
try:
return str(ipaddress.ip_address(out[0])) if out else None
except ValueError:
return None
def _install_snmp_agent(self) -> FixStepOutcomeDict: def _install_snmp_agent(self) -> FixStepOutcomeDict:
pm = self._pkg_manager or self._detect_pkg_manager() pm = self._pkg_manager or self._detect_pkg_manager()
command = _SNMP_INSTALL.get(pm or "") command = _SNMP_INSTALL.get(pm or "")
+20
View File
@@ -191,3 +191,23 @@ def test_every_annotation_resolves():
for name, member in inspect.getmembers(LinuxDriver, inspect.isfunction): for name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
if member.__module__ == "napalm_linux.linux": if member.__module__ == "napalm_linux.linux":
typing.get_type_hints(member) typing.get_type_hints(member)
class TestManagementPeer:
"""The address the host sees this connection come from (``SSH_CONNECTION``)."""
def _driver(self, answer):
d = LinuxDriver("h", "admin", "p")
d._device = MagicMock()
d.run_command = MagicMock(return_value=MagicMock(stdout=answer, exit_code=0))
return d
def test_the_client_side_of_ssh_connection(self):
assert self._driver("10.7.224.20 51234 10.0.0.5 22\n").get_management_peer() == "10.7.224.20"
def test_an_ipv6_client(self):
assert self._driver("fd00::20 51234 fd00::5 22\n").get_management_peer() == "fd00::20"
@pytest.mark.parametrize("answer", ["", "\n", "garbage", "not-an-ip 1 2 3"])
def test_nothing_usable_is_none(self, answer):
assert self._driver(answer).get_management_peer() is None