diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index c6248bf..92c38c8 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -69,6 +69,31 @@ _RC_MARKER = "__NETORK_RC=" _RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) +#: The end of a command's output when nothing better is known: a line that looks +#: like a shell prompt. +_PROMPT_RE = r"[#$\>]\s*$" +#: A command ending in ``echo __NAME=$?`` reports its exit status on a line of its +#: own: ``_RC_MARKER`` here, ``__APT_RC=`` and ``__SVC_RC=`` in napalm-device-types. +_STATUS_ECHO_RE = re.compile(r"echo\s+(__[A-Z_]+=)\$\?") + + +def _expect_for(command: str) -> str: + """The pattern that ends *command*'s output. + + netmiko stops reading as soon as the pattern matches what it has read so far. + A line the command prints can end in ``#``, ``$`` or ``>`` -- apt's + ```` after a bad signature -- and was taken for the + prompt: half the output came back, and the rest started the next command's + (#615). A command that echoes its exit status is read until that marker, with + a number, and the prompt line after it. The echoed command line carries a + literal ``$?`` and cannot match. + """ + markers = _STATUS_ECHO_RE.findall(command) + if not markers: + return _PROMPT_RE + return re.escape(markers[-1]) + r"\d+\s*\n.*" + _PROMPT_RE + + def _split_status(raw: str) -> tuple[str, int | None]: """``(output, exit_status)`` of a command followed by ``echo {_RC_MARKER}$?``. @@ -306,7 +331,7 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDri command, read_timeout=read_timeout, cmd_verify=False, - expect_string=r'[#$\>]\s*$', + expect_string=_expect_for(command), ).strip() def _sudo(self, command: str, read_timeout: float = 100) -> str: diff --git a/tests/test_linux.py b/tests/test_linux.py index d0b074e..66cdd66 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -1,5 +1,7 @@ """Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed).""" +import re + import pytest from unittest.mock import MagicMock, patch from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model @@ -1050,6 +1052,106 @@ class TestSudoStatus: assert "__NETORK_RC=1" not in output +class _Channel: + """A netmiko connection that hands out its output in chunks and stops where + netmiko does: at the first chunk after which ``expect_string`` matches all + that was read so far.""" + + def __init__(self, chunks): + self.chunks = list(chunks) + self.patterns: list = [] + + def send_command(self, command, *, expect_string, **_kwargs): + self.patterns.append(expect_string) + output = "" + while self.chunks: + output += self.chunks.pop(0) + if re.search(expect_string, output): + return output + raise TimeoutError(f"pattern not detected: {expect_string!r}") + + +#: What vault-01 sent on 2026-10-06 while its apt proxy served a corrupted +#: InRelease: the signature line ends in ">", which looks like a prompt (#615). +_BADSIG_CHUNKS = [ + "sudo -n apt-get update -q 2>&1; echo __NETORK_RC=$?\n", + "Fehl:2 http://archive.ubuntu.com/ubuntu noble-updates InRelease\n" + " Die folgenden Signaturen waren ungültig: BADSIG 871920D1991BC93C " + "Ubuntu Archive Automatic Signing Key (2018) \n", + "W: Fehler beim Holen von http://archive.ubuntu.com/ubuntu/dists/noble-updates/InRelease\n" + "E: Das Depot ist nicht signiert.\n__NETORK_RC=100\n", + "chris@vault-01:~$ ", +] + + +class TestReadToTheEnd: + """A line the command prints can end in ``>``, ``#`` or ``$`` -- apt's + ```` after a bad signature. Taken for the prompt, it + ended the read while the command still ran, and the rest arrived as the next + command's output (#615). A command that echoes its exit status is read until + that marker and the prompt after it.""" + + def test_a_signature_line_does_not_end_the_refresh(self, driver): + driver._root = False + driver._device = _Channel(_BADSIG_CHUNKS) + + result = driver.refresh_available_updates() + + assert result["success"] is False + assert "E: Das Depot ist nicht signiert." in result["output"] + + def test_the_session_stays_in_step(self, driver): + """Everything up to the prompt is consumed, so the next command reads its own output.""" + driver._root = False + driver._device = _Channel(_BADSIG_CHUNKS + ["true\n", "__NETORK_RC=0\nchris@vault-01:~$ "]) + + driver.refresh_available_updates() + output, status = driver._sudo_status("true") + + assert status == 0 + assert "BADSIG" not in output + + def test_the_echoed_command_does_not_count_as_the_marker(self, driver): + """Its literal ``$?`` is no number.""" + channel = _Channel(["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "]) + driver._device = channel + + assert driver._sudo_status("true")[1] == 0 + assert channel.chunks == [] + + def test_the_marker_alone_is_not_the_end(self, driver): + """The prompt after it has to be read too, or it would start the next output.""" + channel = _Channel(["out\n__NETORK_RC=0\n", "chris@vault-01:~$ "]) + driver._device = channel + + driver._sudo_status("true") + + assert channel.chunks == [] + + @pytest.mark.parametrize( + "command", + [ + "{ LC_ALL=C apt list --upgradable 2>/dev/null; echo __APT_RC=$?; } | cat", + "timeout 45 systemctl restart -- cron.service; echo __SVC_RC=$?", + ], + ) + def test_every_status_marker_is_waited_for(self, driver, command): + marker = re.search(r"echo (__[A-Z_]+=)", command).group(1) + channel = _Channel([f"x \n", f"{marker}0\nchris@host:~$ "]) + driver._device = channel + + output = driver._send(command) + + assert f"{marker}0" in output + + def test_a_command_without_a_marker_still_ends_at_the_prompt(self, driver): + channel = _Channel(["6.8.0-142-generic\nchris@host:~$ "]) + driver._device = channel + + assert driver._send("uname -r").startswith("6.8.0-142-generic") + assert channel.patterns == [r"[#$\>]\s*$"] + + class TestUninstallExitStatus: """Whether a removal worked is what the package manager's exit status says.