Author SHA1 Message Date
Christian Manivong 60b56c37e0 feat: restart the host (reboot_host)
netOrk asks a driver for reboot_host before it offers a restart, and the
Linux driver had none, so a Linux host -- and OpenMediaVault and QTS, which
inherit this driver -- could not be restarted from netOrk at all (netOrk
#637). reboot_host runs /sbin/reboot detached and two seconds late through
_run_privileged: the launcher's exit status comes back before the host goes
down, and closing the session cannot take the restart with it. A refusal
(sudo without a password) raises with what the host said.
2026-10-06 13:05:27 +02:00
christianmanivong e82f99df7b Merge pull request 'fix: read a command to its exit status, not to the first line that looks like a prompt' (#12) from fix/read-to-the-marker into master 2026-10-06 05:29:25 +00:00
Christian Manivong 2fed2f73e2 fix: read a command to its exit status, not to the first line that looks like a prompt
_send waited for any output ending in "#", "$" or ">". netmiko matches that
against everything read so far, so a chunk that happened to end in such a
line ended the read while the command still ran. apt's bad-signature line,
"... <ftpmaster@ubuntu.com>", did exactly that: refresh_available_updates
returned half its output, and the rest -- exit status and prompt included --
arrived as the next command's output, so the update read after it failed
with "no exit status" (netOrk #615, seen on nine hosts on 2026-10-06).

A command that ends in "echo __NAME=$?" (__NETORK_RC=, and device-types'
__APT_RC= and __SVC_RC=) is now read until that marker, with a number, and
the prompt line after it. The echoed command line carries a literal $? and
cannot match. Every other command keeps the prompt pattern.
2026-10-06 07:18:37 +02:00
christianmanivong b49acb8ed7 Merge pull request 'feat: report where an update comes from and whether it is a security fix, refresh the index, read the host status' (#11) from feat/update-origin-host-status into master 2026-10-05 22:20:08 +00:00
Christian Manivong a6f9a17858 feat: report where an update comes from and whether it is a security fix, refresh the index, read the host status
For netOrk MVP 5, on napalm-device-types 2.3.0:

- get_available_updates (apt) runs the shared APT_UPGRADABLE_COMMAND and
  parse_apt_upgradable: origin and security from apt's suites, and a
  ValueError instead of [] when apt failed or its output was cut short.
- dnf/yum: check-update's exit status decides (0 none, 100 updates, anything
  else raises); security comes from `updateinfo list --security`, and is None
  when dnf cannot say. The repository column becomes the origin.
- refresh_available_updates(): apt-get update, dnf/yum makecache, apk update;
  pacman is left out (-Sy without -u invites a partial upgrade).
- HostStatusMixin: reboot required and self-patching, read over SSH.
- _run_privileged(): root runs directly, a sudo password goes through _sudo,
  otherwise sudo -n. Shared by service control and the refresh.
- _split_status() drops terminal codes before it looks for the exit status;
  a pseudo-terminal left keypad codes in front of the marker.

OpenMediaVault inherits all of it.
2026-10-06 00:20:07 +02:00
christianmanivong e31bc2a3bf Merge pull request 'feat: start, stop, restart, enable and disable services, and list them in one round trip' (#9) from feat/manage-service into master 2026-10-05 11:12:14 +00:00
3 changed files with 400 additions and 40 deletions
+126 -39
View File
@@ -32,11 +32,17 @@ from netmiko.exceptions import (
from napalm.base.exceptions import ConnectionException, ConnectionClosedException from napalm.base.exceptions import ConnectionException, ConnectionClosedException
from napalm.base.netmiko_helpers import netmiko_args from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import ( from napalm_device_types import (
APT_UPGRADABLE_COMMAND,
DNF_SECURITY_COMMAND,
FingerprintRule, FingerprintRule,
HostStatusMixin,
KernelFactsMixin, KernelFactsMixin,
OSDriver, OSDriver,
SystemdServicesMixin, SystemdServicesMixin,
SystemdUnavailable, SystemdUnavailable,
parse_apt_upgradable,
parse_dnf_security,
strip_terminal_codes,
) )
from napalm_device_types.models import ( from napalm_device_types.models import (
ApplyUpdatesResultDict, ApplyUpdatesResultDict,
@@ -62,6 +68,63 @@ _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"]
_RC_MARKER = "__NETORK_RC=" _RC_MARKER = "__NETORK_RC="
_RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) _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
``<ftpmaster@ubuntu.com>`` 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}$?``.
The status is ``None`` when the marker never arrived (output cut short), so
a caller can tell "unknown" from "succeeded".
"""
raw = strip_terminal_codes(raw)
matches = list(_RC_MARKER_RE.finditer(raw))
if not matches:
return raw, None
last = matches[-1]
return (raw[: last.start()] + raw[last.end():]).strip(), int(last.group(1))
#: How each package manager refreshes its index. pacman is left out on purpose:
#: ``pacman -Sy`` without ``-u`` invites a partial upgrade on the next install.
_REFRESH = {
# No LC_ALL=C here: under sudo it is an environment variable sudoers may refuse
# to set. Only the exit status decides, so the language is merely what is shown.
"apt": "apt-get update -q 2>&1",
"dnf": "dnf makecache -q 2>&1",
"yum": "yum makecache -q 2>&1",
"apk": "apk update -q 2>&1",
}
_YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null"
#: Restart the host two seconds later, detached from this session: the launcher's
#: exit status comes back before the host goes down, and closing the session
#: cannot take the restart with it.
_REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; /sbin/reboot) </dev/null >/dev/null 2>&1 &'"
#: What to do when sudo wants a password netOrk does not have. #: What to do when sudo wants a password netOrk does not have.
_SUDO_PASSWORD_HINT = ( _SUDO_PASSWORD_HINT = (
"sudo requires a password on this device but none is configured in netOrk. " "sudo requires a password on this device but none is configured in netOrk. "
@@ -151,7 +214,7 @@ def _short_image_id(raw: str) -> str:
return raw.strip().removeprefix("sha256:")[:12] return raw.strip().removeprefix("sha256:")[:12]
class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver): class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, HostStatusMixin, OSDriver):
"""NAPALM driver for generic Linux systems. """NAPALM driver for generic Linux systems.
Connects via SSH (netmiko ``linux`` device type) and auto-detects the Connects via SSH (netmiko ``linux`` device type) and auto-detects the
@@ -273,7 +336,7 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
command, command,
read_timeout=read_timeout, read_timeout=read_timeout,
cmd_verify=False, cmd_verify=False,
expect_string=r'[#$\>]\s*$', expect_string=_expect_for(command),
).strip() ).strip()
def _sudo(self, command: str, read_timeout: float = 100) -> str: def _sudo(self, command: str, read_timeout: float = 100) -> str:
@@ -297,13 +360,9 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
The status is ``None`` when the marker never arrived (output cut short), The status is ``None`` when the marker never arrived (output cut short),
so a caller can tell "unknown" from "succeeded". so a caller can tell "unknown" from "succeeded".
""" """
raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout) return _split_status(
matches = list(_RC_MARKER_RE.finditer(raw)) self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
if not matches: )
return raw, None
last = matches[-1]
output = (raw[: last.start()] + raw[last.end():]).strip()
return output, int(last.group(1))
def _is_root(self) -> bool: def _is_root(self) -> bool:
"""Whether the SSH user is root, asked once per session. """Whether the SSH user is root, asked once per session.
@@ -321,12 +380,39 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
Without a sudo password, ``sudo -n`` fails at once where a prompt would Without a sudo password, ``sudo -n`` fails at once where a prompt would
otherwise hang the session until the read timeout. otherwise hang the session until the read timeout.
""" """
if not privileged or self._is_root(): if not privileged:
return self._send(command, read_timeout=timeout)
return self._run_privileged(command, timeout)
def _run_privileged(self, command: str, timeout: float = 100) -> str:
"""Run *command* as root: directly for a root login, through ``_sudo``
with a sudo password, and through ``sudo -n`` without one -- which fails at
once where a password prompt would hang the session until the timeout."""
if self._is_root():
return self._send(command, read_timeout=timeout) return self._send(command, read_timeout=timeout)
if self._sudo_password: if self._sudo_password:
return self._sudo(command, read_timeout=timeout) return self._sudo(command, read_timeout=timeout)
return self._send(f"sudo -n {command}", read_timeout=timeout) return self._send(f"sudo -n {command}", read_timeout=timeout)
def reboot_host(self) -> None:
"""Restart the host (``HostRebootMixin``); returns once the restart is under way.
:raises RuntimeError: when the host refuses -- sudo without a password,
no ``reboot`` -- or its answer carried no exit status.
"""
output, status = _split_status(
self._run_privileged(f"{_REBOOT_DETACHED}; echo {_RC_MARKER}$?", 30)
)
if status != 0:
reason = output or f"the reboot command exited with status {status}"
if "password is required" in output:
reason = f"{reason}\n{_SUDO_PASSWORD_HINT}"
raise RuntimeError(reason)
def _run_host_status_command(self, command: str) -> str:
"""The transport for ``HostStatusMixin.get_host_status``: read-only, no sudo."""
return self._send(command, read_timeout=60)
def _detect_pkg_manager(self) -> str | None: def _detect_pkg_manager(self) -> str | None:
"""Return the first package manager binary found on PATH.""" """Return the first package manager binary found on PATH."""
for pm in _PKG_MANAGERS: for pm in _PKG_MANAGERS:
@@ -1243,48 +1329,49 @@ class LinuxDriver(KernelFactsMixin, SystemdServicesMixin, OSDriver):
def _get_updates_apt(self) -> list[UpdateDict]: def _get_updates_apt(self) -> list[UpdateDict]:
# apt list --upgradable does not need root; avoid sudo so it works even # apt list --upgradable does not need root; avoid sudo so it works even
# without a configured sudo password. # without a configured sudo password.
out = self._send( # Raises ValueError when apt failed or the output was cut short.
"LC_ALL=C apt list --upgradable 2>/dev/null | grep -v '^Listing'", return parse_apt_upgradable(self._send(APT_UPGRADABLE_COMMAND, read_timeout=60))
read_timeout=60,
)
# Join wrapped lines: netmiko's 80-col pseudo-TTY causes long apt lines to
# break; continuation lines start with a space.
raw_lines: List[str] = []
for line in out.splitlines():
if line.startswith(" ") and raw_lines:
raw_lines[-1] += line.strip()
else:
raw_lines.append(line)
updates: list[UpdateDict] = []
for line in raw_lines:
# openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]
m = re.match(
r"^(\S+)/\S+\s+(\S+)\s+\S+\s+\[upgradable from:\s+(\S+)\]", line
)
if m:
updates.append({
"name": m.group(1),
"current_version": m.group(3),
"new_version": m.group(2),
})
return updates
def _get_updates_rpm(self) -> list[UpdateDict]: def _get_updates_rpm(self) -> list[UpdateDict]:
"""dnf/yum check-update: exit 100 means updates, 0 none, anything else failed."""
cmd = "dnf check-update --quiet 2>/dev/null" if self._pkg_manager == "dnf" else "yum check-update -q 2>/dev/null" cmd = "dnf check-update --quiet 2>/dev/null" if self._pkg_manager == "dnf" else "yum check-update -q 2>/dev/null"
out = self._sudo(cmd) output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
if status not in (0, 100):
raise RuntimeError(f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}")
security = self._rpm_security_names()
updates: list[UpdateDict] = [] updates: list[UpdateDict] = []
for line in out.splitlines(): for line in output.splitlines():
parts = line.split() parts = line.split()
if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]: if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]:
name_arch = parts[0] name = parts[0].rsplit(".", 1)[0]
name = name_arch.rsplit(".", 1)[0] if "." in name_arch else name_arch
updates.append({ updates.append({
"name": name, "name": name,
"current_version": "", "current_version": "",
"new_version": parts[1], "new_version": parts[1],
"origin": parts[2] if len(parts) >= 3 else None,
"security": None if security is None else name in security,
}) })
return updates return updates
def _rpm_security_names(self) -> set[str] | None:
"""Packages a pending security advisory covers; None when dnf/yum cannot say."""
cmd = DNF_SECURITY_COMMAND if self._pkg_manager == "dnf" else _YUM_SECURITY_COMMAND
output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
return parse_dnf_security(output) if status == 0 else None
def refresh_available_updates(self) -> dict[str, Any]:
"""Refresh the package index (apt-get update, dnf makecache, apk update)."""
cmd = _REFRESH.get(self._pkg_manager or "")
if cmd is None:
return {
"success": False,
"output": f"Refreshing the index is not supported for {self._pkg_manager!r}",
}
output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 180))
if status != 0 and "password is required" in output:
output = f"{output}\n{_SUDO_PASSWORD_HINT}"
return {"success": status == 0, "output": output}
def _get_updates_apk(self) -> list[UpdateDict]: def _get_updates_apk(self) -> list[UpdateDict]:
out = self._send("apk version -l '<' 2>/dev/null") out = self._send("apk version -l '<' 2>/dev/null")
updates: list[UpdateDict] = [] updates: list[UpdateDict] = []
+1 -1
View File
@@ -37,7 +37,7 @@ classifiers = [
] ]
dependencies = [ dependencies = [
"napalm>=4.0", "napalm>=4.0",
"napalm-device-types>=2.2.0", "napalm-device-types>=2.3.0",
"netmiko>=4.0.0", "netmiko>=4.0.0",
"paramiko>=5.0.0", # CVE-2026-44405 "paramiko>=5.0.0", # CVE-2026-44405
] ]
+273
View File
@@ -1,5 +1,7 @@
"""Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed).""" """Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
import re
import pytest import pytest
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model
@@ -226,6 +228,7 @@ APT_UPGRADABLE = (
"Listing... Done\n" "Listing... Done\n"
"openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n" "openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n"
"curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n" "curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n"
"__APT_RC=0\n"
) )
@@ -1049,6 +1052,106 @@ class TestSudoStatus:
assert "__NETORK_RC=1" not in output 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) <ftpmaster@ubuntu.com>\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
``<ftpmaster@ubuntu.com>`` 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 <a@b>\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: class TestUninstallExitStatus:
"""Whether a removal worked is what the package manager's exit status says. """Whether a removal worked is what the package manager's exit status says.
@@ -1282,3 +1385,173 @@ class TestManageService:
driver.manage_service("cron; reboot", "stop") driver.manage_service("cron; reboot", "stop")
assert driver._device.send_command.call_count == 0 assert driver._device.send_command.call_count == 0
# ---------------------------------------------------------------------------
# Updates: origin and security, refresh, host status (netOrk MVP 5)
# ---------------------------------------------------------------------------
APT_WITH_SECURITY = (
"openssl/noble-updates,noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable from: 3.0.13-0ubuntu3.5]\n"
"docker-compose-plugin/noble 5.6.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.5.1-1~ubuntu.24.04~noble]\n"
"__APT_RC=0\n"
)
class TestAvailableUpdates:
def test_apt_reports_origin_and_security(self, driver):
driver._pkg_manager = "apt"
_mock_send(driver, APT_WITH_SECURITY)
updates = {u["name"]: u for u in driver.get_available_updates()}
assert updates["openssl"]["security"] is True
assert updates["openssl"]["origin"] == "noble-updates,noble-security"
assert updates["docker-compose-plugin"]["security"] is False
def test_apt_that_could_not_read_raises_instead_of_reporting_nothing(self, driver):
driver._pkg_manager = "apt"
_mock_send(driver, "E: Could not open lock file\n__APT_RC=100\n")
with pytest.raises(ValueError):
driver.get_available_updates()
def test_apt_without_an_exit_status_raises(self, driver):
driver._pkg_manager = "apt"
_mock_send(driver, "openssl/noble-security 3.0.13-0ubuntu3.6 amd64 [upgradable fro")
with pytest.raises(ValueError):
driver.get_available_updates()
def test_dnf_marks_what_a_security_advisory_covers(self, driver):
driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = [
"0", # id -u
"openssl-libs.x86_64 1:3.1.4-2.fc40 updates\n"
"vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
"FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n__NETORK_RC=0",
]
updates = {u["name"]: u for u in driver.get_available_updates()}
assert updates["openssl-libs"]["security"] is True
assert updates["vim-enhanced"]["security"] is False
def test_dnf_without_advisories_leaves_security_unknown(self, driver):
driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = [
"0",
"vim-enhanced.x86_64 2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
"Error: updateinfo metadata missing\n__NETORK_RC=1",
]
assert driver.get_available_updates()[0]["security"] is None
def test_dnf_that_failed_raises(self, driver):
driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = ["0", "Error: Failed to download metadata\n__NETORK_RC=1"]
with pytest.raises(RuntimeError):
driver.get_available_updates()
class TestRefreshAvailableUpdates:
def _sent(self, driver) -> list:
return [c[0][0] for c in driver._device.send_command.call_args_list]
def test_apt_refreshes_its_index_as_root(self, driver):
driver._pkg_manager = "apt"
driver._device.send_command.side_effect = ["0", "Hit:1 http://archive.ubuntu.com noble InRelease\n__NETORK_RC=0"]
result = driver.refresh_available_updates()
assert result["success"] is True
assert "apt-get update" in self._sent(driver)[1]
def test_without_a_sudo_password_it_never_waits_for_one(self, driver):
driver._pkg_manager = "apt"
driver._device.send_command.side_effect = ["1000", "sudo: a password is required\n__NETORK_RC=1"]
result = driver.refresh_available_updates()
assert result["success"] is False
assert self._sent(driver)[1].startswith("sudo -n apt-get update")
def test_dnf_refreshes_its_metadata(self, driver):
driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = ["0", "Metadata cache created.\n__NETORK_RC=0"]
assert driver.refresh_available_updates()["success"] is True
assert "dnf makecache" in self._sent(driver)[1]
def test_pacman_is_not_refreshed_on_its_own(self, driver):
"""pacman -Sy without -u invites a partial upgrade on the next install."""
driver._pkg_manager = "pacman"
result = driver.refresh_available_updates()
assert result["success"] is False
driver._device.send_command.assert_not_called()
class TestHostStatus:
def test_the_driver_carries_the_shared_command(self, driver):
from napalm_device_types.host_status import HOST_STATUS_COMMAND
_mock_send(
driver,
"HSTAT_BEGIN\n[reboot-required]\n[kernel]\n6.8.0-142-generic\n[modules]\n"
"6.8.0-142-generic\n[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n",
)
status = driver.get_host_status()
assert driver._device.send_command.call_args[0][0] == HOST_STATUS_COMMAND
assert status["reboot_required"] is True
class TestTerminalCodes:
def test_a_status_marker_behind_a_terminal_code_is_still_read(self, driver):
"""apt-get on a pseudo-terminal leaves keypad codes in front of the marker."""
driver._pkg_manager = "apt"
driver._device.send_command.side_effect = ["0", "Hit:1 noble InRelease\n\x1b>__NETORK_RC=0"]
assert driver.refresh_available_updates()["success"] is True
class TestRebootHost:
"""``reboot_host`` (napalm-device-types' ``HostRebootMixin``) restarts the host.
Without it netOrk could not restart a Linux host at all: its capability check
looks for ``reboot_host`` and found nothing (netOrk #637). The restart is
detached and a moment late, so the launcher's exit status comes back before
the host goes down, and closing the session cannot take it along.
"""
def test_the_driver_can_restart_its_host(self):
assert callable(getattr(LinuxDriver, "reboot_host", None))
def test_the_restart_is_detached_and_privileged(self, driver):
driver._root = False
driver._device.send_command.return_value = "\n__NETORK_RC=0"
driver.reboot_host()
sent = driver._device.send_command.call_args[0][0]
assert sent.startswith("sudo -n sh -c ")
assert "/sbin/reboot" in sent and "trap" in sent and "&" in sent
assert sent.endswith("echo __NETORK_RC=$?")
def test_a_refusal_is_raised_with_what_the_host_said(self, driver):
driver._root = False
driver._device.send_command.return_value = "sudo: a password is required\n__NETORK_RC=1"
with pytest.raises(RuntimeError, match="password is required"):
driver.reboot_host()
def test_no_exit_status_is_no_success(self, driver):
driver._root = True
driver._device.send_command.return_value = "something else"
with pytest.raises(RuntimeError):
driver.reboot_host()