feat: report the virtual IPs vip-manager and keepalived declare
CI / test (3.10) (pull_request) Successful in 43s
CI / test (3.11) (pull_request) Successful in 53s
CI / test (3.12) (pull_request) Successful in 53s
CI / test (3.10) (push) Successful in 44s
CI / test (3.11) (push) Successful in 40s
CI / test (3.12) (push) Successful in 38s

LinuxDriver mixes in napalm-device-types' VirtualIpsMixin (4.2.0) and
carries its command on the exec channel: with the awk filter it is about
4 kB, past the line the PTY netmiko holds reliably takes. Root first,
then without it (NetOrk/netork#829).

Version 0.4.0.
This commit is contained in:
Christian Manivong
2026-10-08 12:38:03 +02:00
parent 3a76b6d47f
commit a78daccb56
4 changed files with 121 additions and 2 deletions
+98
View File
@@ -1575,3 +1575,101 @@ class TestGetListeningSockets:
refused, plain = self._sent(driver)
assert refused.startswith("sudo -n sh -c '")
assert plain.startswith("sh -c '")
# ---------------------------------------------------------------------------
# get_virtual_ips (napalm-device-types VirtualIpsMixin)
# ---------------------------------------------------------------------------
_VIPS = (
"VIPS_BEGIN\n[systemd]\n[vip-manager vip-manager.service]\nactive active\n"
"cfg ip 10.7.224.10\ncfg netmask 24\ncfg interface ens7\nVIPS_END\n"
)
class TestGetVirtualIps:
"""The command is about 4 kB -- past the line a terminal takes -- so it goes
on an exec channel, never typed into the PTY netmiko holds."""
@staticmethod
def _exec(driver, *answers):
from napalm_device_types import CommandResult
calls = []
replies = iter(answers)
def run_command(command, *, privileged=False, timeout=60, stdin=None):
calls.append((command, privileged))
return CommandResult(*next(replies))
driver.run_command = run_command
return calls
def test_it_reads_with_the_shared_command_on_an_exec_channel(self, driver):
from napalm_device_types import VirtualIpsMixin
assert isinstance(driver, VirtualIpsMixin)
calls = self._exec(driver, (_VIPS, "", 0))
reading = driver.get_virtual_ips()
assert reading["vip_manager"][0]["address"] == "10.7.224.10"
[(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), (_VIPS, "", 0))
reading = driver.get_virtual_ips()
assert reading["vip_manager"][0]["interface"] == "ens7"
assert [privileged for _, privileged in calls] == [True, False]
def test_without_a_sudo_password_the_whole_script_runs_under_sudo_n(self):
from types import SimpleNamespace
channels = []
class Channel:
def __init__(self):
channels.append(self)
def exec_command(self, command):
self.command = command
def settimeout(self, timeout):
pass
def shutdown_write(self):
pass
def close(self):
pass
def recv_ready(self):
return False
def recv_stderr_ready(self):
return False
def exit_status_ready(self):
return True
def recv_exit_status(self):
return 1
transport = SimpleNamespace(open_session=Channel)
driver = LinuxDriver("h", "u", "p", optional_args={})
driver._device = SimpleNamespace(
remote_conn_pre=SimpleNamespace(get_transport=lambda: transport)
)
driver._root = False
with pytest.raises(ValueError):
driver.get_virtual_ips()
privileged, plain = (c.command for c in channels)
assert privileged.startswith("sudo -n sh -c 'sh -c '")
assert plain.startswith("sh -c '")