NAPALM/napalm-linux
1
0 Fork 0
Code Issues 12 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

chore: adopt the NAPALM driver standard #38

Merged
christianmanivong merged 6 commits from chore/driver-standard into main 2026-10-23 12:55:55 +00:00
Conversation 0 Commits 6 Files Changed 14
+438 -205
7 changed files with 438 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 38391891ec - Show all commits

No files matched your search

napalm_linux/linux.py
+239 -135
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0
"""NAPALM driver for generic Linux systems.
@@ -21,16 +20,10 @@ from __future__ import annotations
import ipaddress
import logging
import re
import socket
from shlex import quote as _shlex_quote
from typing import Any, Dict, List, Optional
from typing import Any
from netmiko import ConnectHandler
from netmiko.exceptions import (
NetmikoAuthenticationException,
NetmikoTimeoutException,
)
from napalm.base.exceptions import ConnectionException, ConnectionClosedException
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import (
APT_UPGRADABLE_COMMAND,
@@ -50,18 +43,17 @@ from napalm_device_types import (
ServiceClustersMixin,
ServicePackagesMixin,
SnmpAgentMixin,
SystemIdentityMixin,
SystemdServicesMixin,
SystemdUnavailable,
SystemIdentityMixin,
VirtualIpsMixin,
parse_apt_upgradable,
open_stream_on_transport,
parse_apt_upgradable,
parse_dnf_security,
run_fix_steps,
run_on_transport,
strip_terminal_codes,
)
from napalm_device_types.ssh_host_keys import HostKeyCheck
from napalm_device_types.models import (
ApplyUpdatesResultDict,
CronJobDict,
@@ -70,11 +62,17 @@ from napalm_device_types.models import (
PackageDict,
ProcessDict,
ServiceDict,
SNMPConfigDict,
SnmpAgentSettingsDict,
SNMPConfigDict,
UpdateDict,
UserDict,
)
from napalm_device_types.ssh_host_keys import HostKeyCheck
from netmiko import ConnectHandler
from netmiko.exceptions import (
NetmikoAuthenticationException,
NetmikoTimeoutException,
)
logger = logging.getLogger("napalm_linux")
@@ -187,11 +185,20 @@ _SUDO_PASSWORD_HINT = (
)
# DMI field values that carry no useful information (OEM defaults, blanks)
_BAD_DMI: frozenset[str] = frozenset({
"", "none", "n/a", "not specified", "not applicable",
"to be filled by o.e.m.", "default string", "unknown",
"no asset tag", "not present",
})
_BAD_DMI: frozenset[str] = frozenset(
{
"",
"none",
"n/a",
"not specified",
"not applicable",
"to be filled by o.e.m.",
"default string",
"unknown",
"no asset tag",
"not present",
}
)
# systemd-detect-virt output → human-readable vendor name
_VIRT_VENDOR_MAP: dict[str, str] = {
@@ -210,16 +217,28 @@ _VIRT_VENDOR_MAP: dict[str, str] = {
}
# Container technologies reported by systemd-detect-virt
_CONTAINER_VIRT: frozenset[str] = frozenset({
"docker", "podman", "lxc", "lxc-libvirt", "systemd-nspawn",
})
_CONTAINER_VIRT: frozenset[str] = frozenset(
{
"docker",
"podman",
"lxc",
"lxc-libvirt",
"systemd-nspawn",
}
)
# DMI sys_vendor strings that indicate a VM when detect-virt is unavailable
_VM_DMI_VENDORS: frozenset[str] = frozenset({
"qemu", "vmware, inc.", "microsoft corporation",
"innotek gmbh", "xen", "bochs",
_VM_DMI_VENDORS: frozenset[str] = frozenset(
{
"qemu",
"vmware, inc.",
"microsoft corporation",
"innotek gmbh",
"xen",
"bochs",
"parallels software international inc.",
})
}
)
# Known ARM board model prefixes → canonical vendor name
_ARM_VENDOR_PREFIXES: list[tuple[str, str]] = [
@@ -396,11 +415,15 @@ class LinuxDriver(
try:
self._device.send_command(_HISTORY_OFF, expect_string=_PROMPT_RE)
except Exception:
logger.debug("Could not switch the shell history off on %s", self.hostname, exc_info=True)
logger.debug(
"Could not switch the shell history off on %s", self.hostname, exc_info=True
)
# Prevent PTY from wrapping long output lines (e.g. docker JSON).
try:
self._device.send_command("stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$")
self._device.send_command(
"stty cols 10000 2>/dev/null || true", expect_string=r"[#$>]\s*$"
)
except Exception:
pass
@@ -421,7 +444,7 @@ class LinuxDriver(
if self._device:
try:
return {"is_alive": self._device.remote_conn.transport.is_active()}
except (AttributeError, socket.error, EOFError):
except (OSError, AttributeError, EOFError):
return {"is_alive": False}
return {"is_alive": False}
@@ -473,7 +496,11 @@ class LinuxDriver(
staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
lines = staged.stdout.strip().splitlines()
tmp = lines[-1] if lines else ""
if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]:
if (
staged.exit_code != 0
or not tmp.startswith(_STAGE_PREFIX)
or "/" in tmp[len(_STAGE_PREFIX) :]
):
reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
return CommandResult(reason, "", staged.exit_code or 1)
tmp_q, part_q, path_q = (_shlex_quote(p) for p in (tmp, f"{path}.netork-tmp", path))
@@ -647,7 +674,7 @@ class LinuxDriver(
# files, so fields 0-4 are empty). The sentinel anchors the output so
# splitlines()[start+N] always maps to the correct field index.
"printf 'DMIBEGIN\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n%s\\n' "
"\"$v\" \"$n\" \"$r\" \"$s\" \"$u\" \"$d\" \"$dt\" \"$cs\" \"$cm\""
'"$v" "$n" "$r" "$s" "$u" "$d" "$dt" "$cs" "$cm"'
)
try:
raw_lines = self._send(dmi_cmd).splitlines()
@@ -674,10 +701,7 @@ class LinuxDriver(
cpuinfo_mdl = _clean(8)
is_container = detect_virt in _CONTAINER_VIRT
is_vm = (
detect_virt not in ("none", "")
or sys_vendor.lower() in _VM_DMI_VENDORS
)
is_vm = detect_virt not in ("none", "") or sys_vendor.lower() in _VM_DMI_VENDORS
if is_container:
return {
@@ -743,7 +767,9 @@ class LinuxDriver(
platform = self._collect_platform_info()
iface_out = self._send("ip -o link show | awk -F': ' '{print $2}' | cut -d@ -f1")
interface_list = [i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"]
interface_list = [
i.strip() for i in iface_out.splitlines() if i.strip() and i.strip() != "lo"
]
# Currently-booted kernel release, distinct from an installed-but-not-yet-
# booted newer kernel (used for kernel CVE relevance).
@@ -769,7 +795,7 @@ class LinuxDriver(
except (IndexError, ValueError):
return 0
def get_lldp_neighbors(self) -> Dict[str, List[dict[str, Any]]]:
def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
"""Return LLDP neighbors if lldpd is installed and currently running.
Uses ``lldpctl -f keyvalue``. Returns an empty dict when lldpd is
@@ -789,8 +815,8 @@ class LinuxDriver(
return {}
output = self._send("lldpctl -f keyvalue 2>/dev/null || true")
neighbors: Dict[str, List[dict[str, Any]]] = {}
entries: Dict[str, Dict[str, str]] = {}
neighbors: dict[str, list[dict[str, Any]]] = {}
entries: dict[str, dict[str, str]] = {}
for line in output.splitlines():
line = line.strip()
@@ -899,14 +925,16 @@ class LinuxDriver(
net.version == 6 and net.prefixlen >= 128
):
continue
networks.append({
networks.append(
{
"network": str(net),
"interface": iface_name,
"gateway": str(iface_obj.ip),
"family": family,
"prefix_length": net.prefixlen,
"vlan_id": None,
})
}
)
except ValueError:
pass
@@ -917,13 +945,13 @@ class LinuxDriver(
destination: str = "",
protocol: str = "",
longer: bool = False,
) -> Dict[str, List[dict[str, Any]]]:
) -> dict[str, list[dict[str, Any]]]:
"""Return the routing table via ``ip route show``.
OSPF routes (from FRR/Quagga) are included via ``ip route show proto ospf``
if any are present. The result is keyed by network prefix.
"""
routes: Dict[str, List[dict[str, Any]]] = {}
routes: dict[str, list[dict[str, Any]]] = {}
proto_map = {
"kernel": "connected",
@@ -937,7 +965,9 @@ class LinuxDriver(
"zebra": "zebra",
}
def _make_entry(proto: str, nexthop: str, iface: str, metric: int, network: str) -> dict[str, Any]:
def _make_entry(
proto: str, nexthop: str, iface: str, metric: int, network: str
) -> dict[str, Any]:
family = "ipv6" if (":" in network or (nexthop and ":" in nexthop)) else "ipv4"
return {
"protocol": proto,
@@ -972,7 +1002,11 @@ class LinuxDriver(
if not dest_m:
continue
raw_dest = dest_m.group(1)
network = "0.0.0.0/0" if raw_dest == "default" else ("::/0" if raw_dest == "default6" else raw_dest)
network = (
"0.0.0.0/0"
if raw_dest == "default"
else ("::/0" if raw_dest == "default6" else raw_dest)
)
if "/" not in network:
network += "/32"
@@ -1000,9 +1034,14 @@ class LinuxDriver(
# FRR/Zebra enrichment via vtysh — properly attributes OSPF/BGP/RIP protocols.
# FRR installs routes into the kernel as "proto zebra"; vtysh gives the real source.
_frr_code: Dict[str, str] = {
"O": "ospf", "B": "bgp", "R": "rip", "I": "isis",
"S": "static", "K": "connected", "C": "connected",
_frr_code: dict[str, str] = {
"O": "ospf",
"B": "bgp",
"R": "rip",
"I": "isis",
"S": "static",
"K": "connected",
"C": "connected",
}
if self._send("command -v vtysh 2>/dev/null").strip():
try:
@@ -1040,23 +1079,23 @@ class LinuxDriver(
return routes
def get_arp_table(self, vrf: str = "") -> List[dict[str, Any]]:
def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
"""Return the ARP/neighbour table."""
entries = []
neigh_out = self._send("ip -4 neigh show")
for line in neigh_out.splitlines():
# 192.168.1.1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE
m = re.match(
r"^([\d.]+)\s+dev\s+(\S+)\s+lladdr\s+([\da-f:]+)\s+(\S+)", line
)
m = re.match(r"^([\d.]+)\s+dev\s+(\S+)\s+lladdr\s+([\da-f:]+)\s+(\S+)", line)
if not m:
continue
entries.append({
entries.append(
{
"interface": m.group(2),
"mac": m.group(3),
"ip": m.group(1),
"age": 0.0,
})
}
)
return entries
def get_config(
@@ -1133,9 +1172,7 @@ class LinuxDriver(
sent, received = int(m.group(1)), int(m.group(2))
# Parse rtt line: "rtt min/avg/max/mdev = 0.123/0.456/0.789/0.100 ms"
rtt_m = re.search(
r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output
)
rtt_m = re.search(r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output)
results = []
for line in output.splitlines():
@@ -1143,10 +1180,12 @@ class LinuxDriver(
r"bytes from ([\d.]+).*icmp_seq=\d+ ttl=(\d+) time=([\d.]+) ms", line
)
if icmp_m:
results.append({
results.append(
{
"ip_address": icmp_m.group(1),
"rtt": float(icmp_m.group(3)),
})
}
)
return {
"success": {
@@ -1233,9 +1272,7 @@ class LinuxDriver(
return self._get_packages_apk()
if self._pkg_manager == "pacman":
return self._get_packages_pacman()
raise NotImplementedError(
f"Package manager '{self._pkg_manager}' is not supported"
)
raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
def _get_packages_apt(self) -> list[PackageDict]:
out = self._send(
@@ -1251,7 +1288,9 @@ class LinuxDriver(
continue
name = parts[0].strip()
version = parts[1].strip()
size = int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
size = (
int(parts[2].strip()) * 1024 if len(parts) > 2 and parts[2].strip().isdigit() else 0
)
# Debian source package (e.g. openssh-server → openssh) for OSV matching.
source_package = parts[3].strip() if len(parts) > 3 and parts[3].strip() else name
# And its version, which is a different number from this package's.
@@ -1269,7 +1308,8 @@ class LinuxDriver(
# dpkg that does not know the field at all.
source_version = parts[4].strip() if len(parts) > 4 and parts[4].strip() else version
description = parts[5].strip() if len(parts) > 5 else ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": True,
@@ -1278,7 +1318,8 @@ class LinuxDriver(
"source": "apt",
"source_package": source_package,
"source_version": source_version,
})
}
)
return packages
def _get_packages_rpm(self) -> list[PackageDict]:
@@ -1290,14 +1331,18 @@ class LinuxDriver(
parts = line.split("\t", 3)
if len(parts) < 2:
continue
packages.append({
packages.append(
{
"name": parts[0].strip(),
"version": parts[1].strip(),
"installed": True,
"description": parts[3].strip() if len(parts) > 3 else "",
"size": int(parts[2].strip()) if len(parts) > 2 and parts[2].strip().isdigit() else 0,
"size": int(parts[2].strip())
if len(parts) > 2 and parts[2].strip().isdigit()
else 0,
"source": self._pkg_manager or "rpm",
})
}
)
return packages
def _get_packages_apk(self) -> list[PackageDict]:
@@ -1308,14 +1353,16 @@ class LinuxDriver(
m = re.match(r"^(\S+)-(\d[\S]*)\s*(.*)", line)
if not m:
continue
packages.append({
packages.append(
{
"name": m.group(1),
"version": m.group(2),
"installed": True,
"description": m.group(3).strip(),
"size": 0,
"source": "apk",
})
}
)
return packages
def _get_packages_pacman(self) -> list[PackageDict]:
@@ -1325,27 +1372,32 @@ class LinuxDriver(
parts = line.split(None, 1)
if len(parts) < 2:
continue
packages.append({
packages.append(
{
"name": parts[0],
"version": parts[1],
"installed": True,
"description": "",
"size": 0,
"source": "pacman",
})
}
)
return packages
def search_packages(self, query: str) -> List[dict[str, Any]]:
def search_packages(self, query: str) -> list[dict[str, Any]]:
"""Search available (installable) packages matching *query*."""
from shlex import quote as _q
safe_q = _q(query)
installed = {p["name"] for p in self.get_packages()}
packages: List[dict[str, Any]] = []
packages: list[dict[str, Any]] = []
if self._pkg_manager == "apt":
out = self._send(f"apt-cache search {safe_q} 2>/dev/null")
versions: Dict[str, str] = {}
ver_out = self._send(f"apt-cache show {safe_q} 2>/dev/null | grep -E '^(Package|Version):' || true")
versions: dict[str, str] = {}
ver_out = self._send(
f"apt-cache show {safe_q} 2>/dev/null | grep -E '^(Package|Version):' || true"
)
cur_pkg = ""
for line in ver_out.splitlines():
if line.startswith("Package:"):
@@ -1357,14 +1409,16 @@ class LinuxDriver(
continue
name, _, description = line.partition(" - ")
name = name.strip()
packages.append({
packages.append(
{
"name": name,
"version": versions.get(name, ""),
"installed": name in installed,
"description": description.strip(),
"size": 0,
"source": "apt",
})
}
)
elif self._pkg_manager in ("dnf", "yum"):
cmd = "dnf" if self._pkg_manager == "dnf" else "yum"
@@ -1375,14 +1429,16 @@ class LinuxDriver(
pkg_ver, _, description = line.partition(" : ")
name = pkg_ver.split(".")[0].strip()
version = ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": name in installed,
"description": description.strip(),
"size": 0,
"source": self._pkg_manager or "rpm",
})
}
)
elif self._pkg_manager == "apk":
out = self._send(f"apk search {safe_q} 2>/dev/null")
@@ -1395,14 +1451,16 @@ class LinuxDriver(
name, version, description = m.group(1), m.group(2), (m.group(3) or "")
else:
name, version, description = line, "", ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": name in installed,
"description": description,
"size": 0,
"source": "apk",
})
}
)
elif self._pkg_manager == "pacman":
out = self._send(f"pacman -Ss {safe_q} 2>/dev/null || true")
@@ -1417,14 +1475,16 @@ class LinuxDriver(
name = name_parts[0]
version = parts[1] if len(parts) > 1 else ""
description = lines[i + 1].strip() if i + 1 < len(lines) else ""
packages.append({
packages.append(
{
"name": name,
"version": version,
"installed": name in installed,
"description": description,
"size": 0,
"source": "pacman",
})
}
)
i += 2
continue
i += 1
@@ -1434,10 +1494,13 @@ class LinuxDriver(
def install_package(self, name: str) -> dict[str, Any]:
"""Install a package by name. Returns ``{"success": bool, "output": str}``."""
from shlex import quote as _q
safe = _q(name)
pm = self._pkg_manager
if pm == "apt":
raw = self._sudo(f"DEBIAN_FRONTEND=noninteractive apt-get install -y {safe} 2>&1 || true")
raw = self._sudo(
f"DEBIAN_FRONTEND=noninteractive apt-get install -y {safe} 2>&1 || true"
)
elif pm in ("dnf", "yum"):
raw = self._sudo(f"{pm} install -y {safe} 2>&1 || true")
elif pm == "apk":
@@ -1447,7 +1510,17 @@ class LinuxDriver(
else:
return {"success": False, "output": f"Unsupported package manager: {pm}"}
low = raw.lower()
success = not any(kw in low for kw in ("error:", "failed", "no packages", "not found", "unable to locate", "no match"))
success = not any(
kw in low
for kw in (
"error:",
"failed",
"no packages",
"not found",
"unable to locate",
"no match",
)
)
return {"success": success, "output": raw.strip()}
#: Words in a package manager's output that mean it did not do the job.
@@ -1479,6 +1552,7 @@ class LinuxDriver(
eventually break something apt would have refused to.
"""
from shlex import quote as _q
safe = _q(name)
pm = self._pkg_manager
if pm == "apt":
@@ -1540,9 +1614,7 @@ class LinuxDriver(
return self._get_updates_apk()
if self._pkg_manager == "pacman":
return self._get_updates_pacman()
raise NotImplementedError(
f"Package manager '{self._pkg_manager}' is not supported"
)
raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
def get_package_proxy(self) -> str | None:
"""The proxy apt fetches through (``00proxy``); None for none or a non-apt host."""
@@ -1560,23 +1632,31 @@ class LinuxDriver(
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"
)
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:]}")
raise RuntimeError(
f"{self._pkg_manager} check-update failed (exit {status}): {output[-200:]}"
)
security = self._rpm_security_names()
updates: list[UpdateDict] = []
for line in output.splitlines():
parts = line.split()
if len(parts) >= 2 and not line.startswith(" ") and "." in parts[0]:
name = parts[0].rsplit(".", 1)[0]
updates.append({
updates.append(
{
"name": name,
"current_version": "",
"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
def _rpm_security_names(self) -> set[str] | None:
@@ -1614,11 +1694,13 @@ class LinuxDriver(
# openssh-9.3_p2-r3 < 9.3_p2-r4
m = re.match(r"^(\S+)-(\S+)\s+<\s+(\S+)", line)
if m:
updates.append({
updates.append(
{
"name": m.group(1),
"current_version": m.group(2),
"new_version": m.group(3),
})
}
)
return updates
def _get_updates_pacman(self) -> list[UpdateDict]:
@@ -1628,11 +1710,13 @@ class LinuxDriver(
# openssh 9.3p2-1 -> 9.4p1-1
m = re.match(r"^(\S+)\s+(\S+)\s+->\s+(\S+)", line)
if m:
updates.append({
updates.append(
{
"name": m.group(1),
"current_version": m.group(2),
"new_version": m.group(3),
})
}
)
return updates
# ------------------------------------------------------------------
@@ -1640,9 +1724,9 @@ class LinuxDriver(
# ------------------------------------------------------------------
# Allowlist for package names – same pattern used by napalm-proxmox
_PKG_NAME_RE = re.compile(r'^[a-zA-Z0-9_\-\+\.]+$')
_PKG_NAME_RE = re.compile(r"^[a-zA-Z0-9_\-\+\.]+$")
def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
def apply_updates(self, packages: list[str]) -> ApplyUpdatesResultDict:
"""Upgrade *packages* (or all pending updates when the list is empty).
Package names are validated against ``^[a-zA-Z0-9_\\-\\+\\.]+$`` before
@@ -1660,30 +1744,27 @@ class LinuxDriver(
return self._apply_updates_apk(packages)
if self._pkg_manager == "pacman":
return self._apply_updates_pacman(packages)
raise NotImplementedError(
f"Package manager '{self._pkg_manager}' is not supported"
)
raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
def _apply_updates_apt(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_apt(self, packages: list[str]) -> ApplyUpdatesResultDict:
pkg_args = " ".join(packages) if packages else "--with-new-pkgs"
cmd = (
"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y "
f"{pkg_args} 2>&1"
if packages else
"DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1"
f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1"
if packages
else "DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1"
)
try:
output = self._sudo(cmd, read_timeout=600)
success = not re.search(r'^E:', output, re.MULTILINE)
success = not re.search(r"^E:", output, re.MULTILINE)
result: ApplyUpdatesResultDict = {"success": success, "output": output}
if not success:
m = re.search(r'^E:.*', output, re.MULTILINE)
m = re.search(r"^E:.*", output, re.MULTILINE)
result["error"] = m.group(0) if m else "apt-get exited with errors"
return result
except Exception as exc:
return {"success": False, "output": "", "error": str(exc)}
def _apply_updates_rpm(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_rpm(self, packages: list[str]) -> ApplyUpdatesResultDict:
bin_ = self._pkg_manager # "dnf" or "yum"
if packages:
pkg_args = " ".join(packages)
@@ -1694,16 +1775,16 @@ class LinuxDriver(
output = self._sudo(cmd, read_timeout=600)
# dnf/yum signal failure via "Error:" lines or non-zero exit;
# since we can't check the exit code directly, look for error markers.
success = not re.search(r'^Error:', output, re.MULTILINE | re.IGNORECASE)
success = not re.search(r"^Error:", output, re.MULTILINE | re.IGNORECASE)
result: ApplyUpdatesResultDict = {"success": success, "output": output}
if not success:
m = re.search(r'^Error:.*', output, re.MULTILINE | re.IGNORECASE)
m = re.search(r"^Error:.*", output, re.MULTILINE | re.IGNORECASE)
result["error"] = m.group(0) if m else f"{bin_} exited with errors"
return result
except Exception as exc:
return {"success": False, "output": "", "error": str(exc)}
def _apply_updates_apk(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_apk(self, packages: list[str]) -> ApplyUpdatesResultDict:
if packages:
pkg_args = " ".join(packages)
cmd = f"apk upgrade {pkg_args} 2>&1"
@@ -1719,7 +1800,7 @@ class LinuxDriver(
except Exception as exc:
return {"success": False, "output": "", "error": str(exc)}
def _apply_updates_pacman(self, packages: List[str]) -> ApplyUpdatesResultDict:
def _apply_updates_pacman(self, packages: list[str]) -> ApplyUpdatesResultDict:
if packages:
pkg_args = " ".join(packages)
cmd = f"pacman --noconfirm -S {pkg_args} 2>&1"
@@ -1763,12 +1844,14 @@ class LinuxDriver(
m = re.match(r"^\s*\[\s*([+\-?])\s*\]\s+(\S+)", line)
if not m:
continue
services.append({
services.append(
{
"name": m.group(2),
"running": m.group(1) == "+",
"enabled": False,
"pid": 0,
})
}
)
return services
# ------------------------------------------------------------------
@@ -1781,7 +1864,7 @@ class LinuxDriver(
groups_out = self._send("getent group 2>/dev/null || cat /etc/group")
# Build uid→[group] map from /etc/group
uid_to_groups: Dict[int, List[str]] = {}
uid_to_groups: dict[int, list[str]] = {}
for line in groups_out.splitlines():
parts = line.split(":")
if len(parts) < 4:
@@ -1793,7 +1876,7 @@ class LinuxDriver(
uid_to_groups.setdefault(-1, []) # placeholder
# Simpler: collect username→groups, then join with passwd
username_to_groups: Dict[str, List[str]] = {}
username_to_groups: dict[str, list[str]] = {}
for line in groups_out.splitlines():
parts = line.split(":")
if len(parts) < 4:
@@ -1813,14 +1896,16 @@ class LinuxDriver(
uid, gid = int(uid_s), int(gid_s)
except ValueError:
continue
users.append({
users.append(
{
"username": username,
"uid": uid,
"gid": gid,
"home": home,
"shell": shell,
"groups": username_to_groups.get(username, []),
})
}
)
return users
# ------------------------------------------------------------------
@@ -1856,7 +1941,8 @@ class LinuxDriver(
command = parts[10]
except (ValueError, IndexError):
continue
processes.append({
processes.append(
{
"pid": pid,
"ppid": ppid,
"user": user,
@@ -1868,7 +1954,8 @@ class LinuxDriver(
"state": state,
"started": started,
"command": command,
})
}
)
return processes
# ------------------------------------------------------------------
@@ -1909,7 +1996,7 @@ class LinuxDriver(
@staticmethod
def _parse_cron_line(
line: str, source_user: str, has_user_field: bool
) -> Optional[CronJobDict]:
) -> CronJobDict | None:
"""Parse a single crontab line; returns ``None`` for comments/blanks."""
stripped = line.strip()
# Remove trailing comment
@@ -1946,12 +2033,11 @@ class LinuxDriver(
# ── Device actions ────────────────────────────────────────────────────────
def get_snmp_config(self) -> Optional[SNMPConfigDict]:
def get_snmp_config(self) -> SNMPConfigDict | None:
"""Return SNMP agent config if snmpd is installed and running."""
try:
running = (
self._send("systemctl is-active snmpd 2>/dev/null || true").strip()
== "active"
self._send("systemctl is-active snmpd 2>/dev/null || true").strip() == "active"
)
if not running:
return None
@@ -1986,7 +2072,7 @@ class LinuxDriver(
"grep -E '^agentAddress' /etc/snmp/snmpd.conf 2>/dev/null | head -1"
).strip()
if port_line:
m = re.search(r':(\d+)', port_line)
m = re.search(r":(\d+)", port_line)
if m:
port = int(m.group(1))
except Exception:
@@ -2062,7 +2148,9 @@ class LinuxDriver(
return self._root_outcome(f"rm -f {_APT_PROXY_CONF}", 30, "proxy removed")
if not _PROXY_URL.match(url):
raise ValueError(f"not a proxy URL: {url!r}")
written = self._install_file_as_root(_APT_PROXY_CONF, f'Acquire::http::Proxy "{url}";\n')
written = self._install_file_as_root(
_APT_PROXY_CONF, f'Acquire::http::Proxy "{url}";\n'
)
if written.exit_code:
return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
return {"ok": True, "output": f"apt fetches through {url}"}
@@ -2076,7 +2164,9 @@ class LinuxDriver(
if not _USER_NAME.match(user):
raise ValueError(f"not a user name: {user!r}")
return self._root_outcome(
f"usermod -aG docker {user}", 30, f"{user} is in the docker group from the next login"
f"usermod -aG docker {user}",
30,
f"{user} is in the docker group from the next login",
)
return run_fix_steps([("grant", grant)])
@@ -2101,15 +2191,19 @@ class LinuxDriver(
raw = self._sudo("wg show all dump 2>&1 || true")
_perm_errors = (
"operation not permitted", "permission denied",
"a password is required", "a terminal is required",
"command not found", "not found", "no such file",
"operation not permitted",
"permission denied",
"a password is required",
"a terminal is required",
"command not found",
"not found",
"no such file",
)
has_peer_data = raw and not any(e in raw.lower() for e in _perm_errors)
if has_peer_data:
# Build listen-port map from interface lines (5 tab-separated fields)
listen_ports: Dict[str, str] = {}
listen_ports: dict[str, str] = {}
for line in raw.splitlines():
parts = line.split("\t")
if len(parts) == 5:
@@ -2121,8 +2215,17 @@ class LinuxDriver(
parts = line.split("\t")
if len(parts) != 9:
continue
(iface, pubkey, _psk, endpoint, allowed_ips,
latest_hs, rx_bytes, tx_bytes, _keepalive) = parts
(
iface,
pubkey,
_psk,
endpoint,
allowed_ips,
latest_hs,
rx_bytes,
tx_bytes,
_keepalive,
) = parts
iface = iface.strip()
pubkey = pubkey.strip()
@@ -2181,13 +2284,14 @@ class LinuxDriver(
try:
import json as _json
iface_list = _json.loads(ip_raw)
except Exception:
return tunnels
# /proc/net/dev for total RX/TX bytes per interface
proc_dev = self._send("cat /proc/net/dev 2>/dev/null || true")
proc_bytes: Dict[str, tuple] = {}
proc_bytes: dict[str, tuple] = {}
for line in proc_dev.splitlines()[2:]:
line = line.strip()
if ":" not in line:
pyproject.toml
+8 -1
View File
@@ -63,4 +63,11 @@ include = ["napalm_linux*"]
[tool.ruff]
line-length = 100
target-version = "py39"
target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]
[tool.ruff.lint.per-file-ignores]
# Recorded device output keeps its lines.
"tests/**" = ["E501"]
tests/test_command_channel.py
+3 -1
View File
@@ -70,7 +70,9 @@ class FakeTransport:
def _driver(*, root=False, sudo_password=None):
driver = LinuxDriver("h", "u", "p", optional_args={"sudo_password": sudo_password})
transport = FakeTransport()
driver._device = SimpleNamespace(remote_conn_pre=SimpleNamespace(get_transport=lambda: transport))
driver._device = SimpleNamespace(
remote_conn_pre=SimpleNamespace(get_transport=lambda: transport)
)
driver._root = root
return driver, transport
tests/test_fix_templates.py
+3 -1
View File
@@ -203,7 +203,9 @@ class TestManagementPeer:
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"
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"
tests/test_linux.py
+181 -65
View File
@@ -1,9 +1,10 @@
"""Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
import re
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import MagicMock, patch
from napalm_linux.linux import (
_YUM_SECURITY_COMMAND,
LinuxDriver,
@@ -11,7 +12,6 @@ from napalm_linux.linux import (
_split_status,
)
# ---------------------------------------------------------------------------
# Fixture – driver without a real connection
# ---------------------------------------------------------------------------
@@ -327,11 +327,7 @@ PASSWD_OUT = (
"daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n"
)
GROUP_OUT = (
"sudo:x:27:admin\n"
"docker:x:999:admin\n"
"adm:x:4:admin\n"
)
GROUP_OUT = "sudo:x:27:admin\ndocker:x:999:admin\nadm:x:4:admin\n"
def test_get_os_users(driver):
@@ -477,8 +473,7 @@ DNF_UPGRADE_SUCCESS = (
)
DNF_UPGRADE_ERROR = (
"Last metadata expiration check: 0:01:23 ago.\n"
"Error: No match for argument: nonexistent-pkg\n"
"Last metadata expiration check: 0:01:23 ago.\nError: No match for argument: nonexistent-pkg\n"
)
@@ -532,6 +527,7 @@ def test_type_label_is_linux():
def test_type_label_overrides_os_driver():
from napalm_device_types import OSDriver
assert OSDriver.TYPE_LABEL == "OS"
assert LinuxDriver.TYPE_LABEL != OSDriver.TYPE_LABEL
@@ -555,18 +551,31 @@ def _dmi_output(
# Mirrors the real shell output: DMIBEGIN sentinel followed by 9 fields.
# The sentinel prevents _send()'s .strip() from eating leading blank lines
# (which occur on ARM where all DMI files are absent).
return "\n".join([
return "\n".join(
[
"DMIBEGIN",
sys_vendor, product_name, product_version, product_serial,
product_uuid, detect_virt, dt_model, cpuinfo_serial, cpuinfo_model,
])
sys_vendor,
product_name,
product_version,
product_serial,
product_uuid,
detect_virt,
dt_model,
cpuinfo_serial,
cpuinfo_model,
]
)
class TestCollectPlatformInfo:
def test_baremetal_dell(self, driver):
raw = _dmi_output(
"Dell Inc.", "PowerEdge R720", "Not Specified", "ABC123",
"8a2e3f00-dead-beef-0000-123456789abc", "none",
"Dell Inc.",
"PowerEdge R720",
"Not Specified",
"ABC123",
"8a2e3f00-dead-beef-0000-123456789abc",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -577,8 +586,12 @@ class TestCollectPlatformInfo:
def test_baremetal_lenovo_product_version_preferred(self, driver):
raw = _dmi_output(
"LENOVO", "10M8000VUS", "ThinkCentre M910x", "MP1234",
"8a2e3f00-dead-beef-0000-123456789abc", "none",
"LENOVO",
"10M8000VUS",
"ThinkCentre M910x",
"MP1234",
"8a2e3f00-dead-beef-0000-123456789abc",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -589,8 +602,12 @@ class TestCollectPlatformInfo:
def test_vm_kvm(self, driver):
raw = _dmi_output(
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "pc-i440fx-9.1", "",
"4c4c4544-0000-2010-8020-b4c04f534a31", "kvm",
"QEMU",
"Standard PC (i440FX + PIIX, 1996)",
"pc-i440fx-9.1",
"",
"4c4c4544-0000-2010-8020-b4c04f534a31",
"kvm",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -601,8 +618,12 @@ class TestCollectPlatformInfo:
def test_vm_vmware(self, driver):
raw = _dmi_output(
"VMware, Inc.", "VMware Virtual Platform", "None", "VMware-42 12 34 56",
"4244560c-dead-beef-0000-abcdef123456", "vmware",
"VMware, Inc.",
"VMware Virtual Platform",
"None",
"VMware-42 12 34 56",
"4244560c-dead-beef-0000-abcdef123456",
"vmware",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -613,8 +634,12 @@ class TestCollectPlatformInfo:
def test_vm_hyperv(self, driver):
raw = _dmi_output(
"Microsoft Corporation", "Virtual Machine", "Hyper-V UEFI Release v4.1", "",
"7C5B4B1F-1234-5678-ABCD-000000000001", "microsoft",
"Microsoft Corporation",
"Virtual Machine",
"Hyper-V UEFI Release v4.1",
"",
"7C5B4B1F-1234-5678-ABCD-000000000001",
"microsoft",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -625,8 +650,12 @@ class TestCollectPlatformInfo:
def test_junk_dmi_values_filtered(self, driver):
raw = _dmi_output(
"To Be Filled By O.E.M.", "To Be Filled By O.E.M.", "Not Specified",
"To Be Filled By O.E.M.", "", "none",
"To Be Filled By O.E.M.",
"To Be Filled By O.E.M.",
"Not Specified",
"To Be Filled By O.E.M.",
"",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -637,8 +666,12 @@ class TestCollectPlatformInfo:
def test_vm_kvm_fallback_via_dmi_when_detect_virt_unavailable(self, driver):
# systemd-detect-virt returns "none" (not installed), sys_vendor reveals QEMU
raw = _dmi_output(
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "", "",
"4c4c4544-0000-2010-8020-b4c04f534a31", "none",
"QEMU",
"Standard PC (i440FX + PIIX, 1996)",
"",
"",
"4c4c4544-0000-2010-8020-b4c04f534a31",
"none",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -648,8 +681,12 @@ class TestCollectPlatformInfo:
def test_container_docker(self, driver):
raw = _dmi_output(
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "", "",
"4c4c4544-0000-2010-8020-b4c04f534a31", "docker",
"QEMU",
"Standard PC (i440FX + PIIX, 1996)",
"",
"",
"4c4c4544-0000-2010-8020-b4c04f534a31",
"docker",
)
with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info()
@@ -674,7 +711,12 @@ class TestCollectPlatformInfo:
def test_arm_device_tree_fallback(self, driver):
raw = _dmi_output(
"", "", "", "", "", "none",
"",
"",
"",
"",
"",
"none",
dt_model="Raspberry Pi 4 Model B Rev 1.4",
cpuinfo_serial="100000002a6d96dc",
cpuinfo_model="Raspberry Pi 4 Model B Rev 1.4",
@@ -688,7 +730,12 @@ class TestCollectPlatformInfo:
def test_arm_cpuinfo_model_when_no_dt(self, driver):
raw = _dmi_output(
"", "", "", "", "", "none",
"",
"",
"",
"",
"",
"none",
dt_model="",
cpuinfo_serial="0000000012345678",
cpuinfo_model="Raspberry Pi 3 Model B Plus Rev 1.3",
@@ -713,7 +760,12 @@ class TestCollectPlatformInfo:
# d="none\nnone" which shifted subsequent lines off by one.
# With the ${d:-none} fix this no longer happens.
raw = _dmi_output(
"", "", "", "", "", "none",
"",
"",
"",
"",
"",
"none",
dt_model="",
cpuinfo_serial="0000000012345678",
cpuinfo_model="Raspberry Pi 3 Model B Rev 1.2",
@@ -726,8 +778,12 @@ class TestCollectPlatformInfo:
def test_arm_fallback_ignored_when_dmi_present(self, driver):
raw = _dmi_output(
"Dell Inc.", "PowerEdge R720", "Not Specified", "XYZ999",
"8a2e3f00-dead-beef-0000-123456789abc", "none",
"Dell Inc.",
"PowerEdge R720",
"Not Specified",
"XYZ999",
"8a2e3f00-dead-beef-0000-123456789abc",
"none",
dt_model="some-dt-model",
cpuinfo_serial="deadbeef",
cpuinfo_model="some cpuinfo model",
@@ -744,7 +800,9 @@ class TestCollectPlatformInfo:
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("model,expected", [
@pytest.mark.parametrize(
"model,expected",
[
("Raspberry Pi 4 Model B Rev 1.4", "Raspberry Pi Foundation"),
("Raspberry Pi 3 Model B Rev 1.2", "Raspberry Pi Foundation"),
("ODROID-N2L", "Hardkernel"),
@@ -757,7 +815,8 @@ class TestCollectPlatformInfo:
("BeagleBone Black", "BeagleBoard.org"),
("Unknown Board 1.0", "Unknown Board"),
("SomeSingleWordBoard", "SomeSingleWordBoard"),
])
],
)
def test_arm_vendor_from_model(model, expected):
assert _arm_vendor_from_model(model) == expected
@@ -768,11 +827,27 @@ def test_arm_vendor_from_model(model, expected):
def test_get_facts_baremetal_vendor_model_serial(driver):
platform = {"vendor": "Dell Inc.", "model": "PowerEdge R720", "serial": "ABC123", "is_vm": False}
with patch.object(driver, "_collect_platform_info", return_value=platform), \
patch.object(driver, "_parse_uptime", return_value=86400), \
patch.object(driver, "_send", side_effect=["myhost", "myhost.example.com", "Debian GNU/Linux 12", "eth0\neth1",
"6.1.0-18-amd64"]):
platform = {
"vendor": "Dell Inc.",
"model": "PowerEdge R720",
"serial": "ABC123",
"is_vm": False,
}
with (
patch.object(driver, "_collect_platform_info", return_value=platform),
patch.object(driver, "_parse_uptime", return_value=86400),
patch.object(
driver,
"_send",
side_effect=[
"myhost",
"myhost.example.com",
"Debian GNU/Linux 12",
"eth0\neth1",
"6.1.0-18-amd64",
],
),
):
facts = driver.get_facts()
assert facts["vendor"] == "Dell Inc."
assert facts["model"] == "PowerEdge R720"
@@ -785,13 +860,20 @@ def test_get_facts_baremetal_vendor_model_serial(driver):
def test_get_facts_vm_kvm(driver):
platform = {
"vendor": "KVM", "model": "Virtual Machine",
"serial": "4c4c4544-0000-2010-8020-b4c04f534a31", "is_vm": True,
"vendor": "KVM",
"model": "Virtual Machine",
"serial": "4c4c4544-0000-2010-8020-b4c04f534a31",
"is_vm": True,
}
with patch.object(driver, "_collect_platform_info", return_value=platform), \
patch.object(driver, "_parse_uptime", return_value=3600), \
patch.object(driver, "_send", side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0",
"5.15.0-91-generic"]):
with (
patch.object(driver, "_collect_platform_info", return_value=platform),
patch.object(driver, "_parse_uptime", return_value=3600),
patch.object(
driver,
"_send",
side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0", "5.15.0-91-generic"],
),
):
facts = driver.get_facts()
assert facts["vendor"] == "KVM"
assert facts["model"] == "Virtual Machine"
@@ -800,9 +882,15 @@ def test_get_facts_vm_kvm(driver):
def test_get_facts_fallback_vendor_when_dmi_empty(driver):
platform = {"vendor": "", "model": "", "serial": "", "is_vm": False}
with patch.object(driver, "_collect_platform_info", return_value=platform), \
patch.object(driver, "_parse_uptime", return_value=0), \
patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0", "6.6.7-0-lts"]):
with (
patch.object(driver, "_collect_platform_info", return_value=platform),
patch.object(driver, "_parse_uptime", return_value=0),
patch.object(
driver,
"_send",
side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0", "6.6.7-0-lts"],
),
):
facts = driver.get_facts()
assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute
@@ -813,7 +901,12 @@ class TestNoDockerOfItsOwn:
only the fix that lets the login user use Docker stays."""
def test_the_cli_methods_are_gone(self):
for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"):
for name in (
"get_docker_info",
"get_docker_outdated",
"reconstruct_docker_run",
"_docker_bin",
):
assert not hasattr(LinuxDriver, name), name
def test_the_engine_access_and_the_permission_fix_stay(self):
@@ -1019,7 +1112,9 @@ class TestReadToTheEnd:
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:~$ "])
channel = _Channel(
["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "]
)
driver._device = channel
assert _split_status(driver._send(_TRUE_WITH_STATUS))[1] == 0
@@ -1043,7 +1138,7 @@ class TestReadToTheEnd:
)
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:~$ "])
channel = _Channel(["x <a@b>\n", f"{marker}0\nchris@host:~$ "])
driver._device = channel
output = driver._send(command)
@@ -1167,7 +1262,11 @@ def _kernel_wire(report: str) -> str:
import base64
import gzip
return "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(report.encode())).decode() + "KFACTS_END"
return (
"KFACTS_BEGIN\n"
+ base64.encodebytes(gzip.compress(report.encode())).decode()
+ "KFACTS_END"
)
def test_get_kernel_facts_carries_the_shared_command_across(driver):
@@ -1369,12 +1468,16 @@ class TestAvailableUpdates:
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"),
(
"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"
),
# DNF_SECURITY_COMMAND pipes through cat: the outer status is cat's, dnf's own
# is in the __DNF_RC marker.
("FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"),
(
"FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"
),
]
updates = {u["name"]: u for u in driver.get_available_updates()}
@@ -1402,10 +1505,14 @@ class TestAvailableUpdates:
driver._pkg_manager = "yum"
driver._device.send_command.side_effect = [
"0", # id -u
("openssl-libs.x86_64 1:1.0.2k-26.el7_9 updates\n"
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100"),
("RHSA-2023:3556 Important/Sec. openssl-libs-1:1.0.2k-26.el7_9.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"),
(
"openssl-libs.x86_64 1:1.0.2k-26.el7_9 updates\n"
"vim-enhanced.x86_64 2:7.4.629-8.el7_9 updates\n__NETORK_RC=100"
),
(
"RHSA-2023:3556 Important/Sec. openssl-libs-1:1.0.2k-26.el7_9.x86_64\n"
"__DNF_RC=0\n__NETORK_RC=0"
),
]
updates = {u["name"]: u for u in driver.get_available_updates()}
@@ -1449,7 +1556,10 @@ class TestAvailableUpdates:
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"]
driver._device.send_command.side_effect = [
"0",
"Error: Failed to download metadata\n__NETORK_RC=1",
]
with pytest.raises(RuntimeError):
driver.get_available_updates()
@@ -1461,7 +1571,10 @@ class TestRefreshAvailableUpdates:
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"]
driver._device.send_command.side_effect = [
"0",
"Hit:1 http://archive.ubuntu.com noble InRelease\n__NETORK_RC=0",
]
result = driver.refresh_available_updates()
@@ -1470,7 +1583,10 @@ class TestRefreshAvailableUpdates:
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"]
driver._device.send_command.side_effect = [
"1000",
"sudo: a password is required\n__NETORK_RC=1",
]
result = driver.refresh_available_updates()
tests/test_ssh_host_key.py
+3 -1
View File
@@ -27,7 +27,9 @@ OTHER_KEY = paramiko.ECDSAKey.generate()
def _device() -> MagicMock:
device = MagicMock()
device.send_command.return_value = "/usr/bin/apt"
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = (
SERVER_KEY
)
return device
tests/test_sudo_off_the_command_line.py
+1 -1
View File
@@ -30,7 +30,7 @@ class Channel:
"""An exec channel answering with *output* and *status*; it refuses a line
that holds the password, which is what ``ps`` would show."""
def __init__(self, host: "Transport"):
def __init__(self, host: Transport):
self.host = host
self.command = ""
self.stdin = b""