diff --git a/README.md b/README.md
index e7f87f5..c29ab05 100644
--- a/README.md
+++ b/README.md
@@ -58,11 +58,11 @@ with Driver(
# OS-specific methods (from napalm-device-types OSDriver)
packages = dev.get_packages()
- updates = dev.get_pending_updates()
+ updates = dev.get_pending_updates()
services = dev.get_services()
- users = dev.get_os_users()
- procs = dev.get_processes()
- jobs = dev.get_cron_jobs()
+ users = dev.get_os_users()
+ procs = dev.get_processes()
+ jobs = dev.get_cron_jobs()
# Upgrade specific packages
result = dev.apply_updates(["openssh-server", "curl"])
@@ -121,7 +121,7 @@ apt → dnf → yum → apk → pacman
Force a specific package manager:
```python
-optional_args={"pkg_manager": "dnf"}
+optional_args = {"pkg_manager": "dnf"}
```
## SSH user permissions
diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py
index 7d5f827..26a3b65 100644
--- a/napalm_linux/linux.py
+++ b/napalm_linux/linux.py
@@ -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")
@@ -124,7 +122,7 @@ def _split_status(raw: str) -> tuple[str, int | None]:
if not matches:
return raw, None
last = matches[-1]
- return (raw[: last.start()] + raw[last.end():]).strip(), int(last.group(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:
@@ -187,54 +185,75 @@ _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] = {
- "kvm": "KVM",
- "qemu": "KVM",
- "vmware": "VMware ESXi",
- "microsoft": "Microsoft Hyper-V",
- "xen": "Xen",
- "virtualbox": "Oracle VirtualBox",
- "parallels": "Parallels",
- "docker": "Docker",
- "podman": "Podman",
- "lxc": "LXC",
- "lxc-libvirt": "LXC",
- "systemd-nspawn": "systemd-nspawn",
+ "kvm": "KVM",
+ "qemu": "KVM",
+ "vmware": "VMware ESXi",
+ "microsoft": "Microsoft Hyper-V",
+ "xen": "Xen",
+ "virtualbox": "Oracle VirtualBox",
+ "parallels": "Parallels",
+ "docker": "Docker",
+ "podman": "Podman",
+ "lxc": "LXC",
+ "lxc-libvirt": "LXC",
+ "systemd-nspawn": "systemd-nspawn",
}
# 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",
- "parallels software international inc.",
-})
+_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]] = [
- ("Raspberry Pi", "Raspberry Pi Foundation"),
- ("NVIDIA Jetson", "NVIDIA"),
- ("ODROID", "Hardkernel"),
- ("Hardkernel", "Hardkernel"),
- ("Rock Pi", "Radxa"),
- ("ROCK Pi", "Radxa"),
- ("Radxa", "Radxa"),
- ("Orange Pi", "Xunlong Software"),
- ("Banana Pi", "SinoVoip"),
- ("NanoPi", "FriendlyElec"),
- ("PINE64", "Pine64"),
- ("BeagleBone", "BeagleBoard.org"),
+ ("Raspberry Pi", "Raspberry Pi Foundation"),
+ ("NVIDIA Jetson", "NVIDIA"),
+ ("ODROID", "Hardkernel"),
+ ("Hardkernel", "Hardkernel"),
+ ("Rock Pi", "Radxa"),
+ ("ROCK Pi", "Radxa"),
+ ("Radxa", "Radxa"),
+ ("Orange Pi", "Xunlong Software"),
+ ("Banana Pi", "SinoVoip"),
+ ("NanoPi", "FriendlyElec"),
+ ("PINE64", "Pine64"),
+ ("BeagleBone", "BeagleBoard.org"),
]
@@ -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()
@@ -663,28 +690,25 @@ class LinuxDriver(
val = lines[idx].strip() if idx < len(lines) else ""
return "" if val.lower() in _BAD_DMI else val
- sys_vendor = _clean(0)
- product_name = _clean(1)
- product_ver = _clean(2)
- product_ser = _clean(3)
- product_uuid = _clean(4)
- detect_virt = lines[5].strip().lower() if len(lines) > 5 else "none"
- dt_model = _clean(6)
- cpuinfo_ser = _clean(7)
- cpuinfo_mdl = _clean(8)
+ sys_vendor = _clean(0)
+ product_name = _clean(1)
+ product_ver = _clean(2)
+ product_ser = _clean(3)
+ product_uuid = _clean(4)
+ detect_virt = lines[5].strip().lower() if len(lines) > 5 else "none"
+ dt_model = _clean(6)
+ cpuinfo_ser = _clean(7)
+ 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 {
"vendor": _VIRT_VENDOR_MAP.get(detect_virt, sys_vendor or "Container"),
- "model": "Container",
+ "model": "Container",
"serial": product_uuid,
- "is_vm": True,
+ "is_vm": True,
}
if is_vm:
@@ -705,9 +729,9 @@ class LinuxDriver(
vendor = sys_vendor
return {
"vendor": vendor,
- "model": "Virtual Machine",
+ "model": "Virtual Machine",
"serial": product_ser or product_uuid,
- "is_vm": True,
+ "is_vm": True,
}
# Bare-metal: prefer product_version when it reads like a marketing name
@@ -715,9 +739,9 @@ class LinuxDriver(
pv_usable = product_ver and product_ver != product_name and " " in product_ver
return {
"vendor": sys_vendor,
- "model": product_ver if pv_usable else product_name,
+ "model": product_ver if pv_usable else product_name,
"serial": product_ser,
- "is_vm": False,
+ "is_vm": False,
}
# ARM/embedded fallback: no DMI, try device tree and /proc/cpuinfo
@@ -725,9 +749,9 @@ class LinuxDriver(
if arm_model:
return {
"vendor": _arm_vendor_from_model(arm_model),
- "model": arm_model,
+ "model": arm_model,
"serial": cpuinfo_ser,
- "is_vm": False,
+ "is_vm": False,
}
return {"vendor": "", "model": "", "serial": "", "is_vm": False}
@@ -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({
- "network": str(net),
- "interface": iface_name,
- "gateway": str(iface_obj.ip),
- "family": family,
- "prefix_length": net.prefixlen,
- "vlan_id": None,
- })
+ 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({
- "interface": m.group(2),
- "mac": m.group(3),
- "ip": m.group(1),
- "age": 0.0,
- })
+ 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({
- "ip_address": icmp_m.group(1),
- "rtt": float(icmp_m.group(3)),
- })
+ 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,16 +1308,18 @@ 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({
- "name": name,
- "version": version,
- "installed": True,
- "description": description,
- "size": size,
- "source": "apt",
- "source_package": source_package,
- "source_version": source_version,
- })
+ packages.append(
+ {
+ "name": name,
+ "version": version,
+ "installed": True,
+ "description": description,
+ "size": size,
+ "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({
- "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,
- "source": self._pkg_manager or "rpm",
- })
+ 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,
+ "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({
- "name": m.group(1),
- "version": m.group(2),
- "installed": True,
- "description": m.group(3).strip(),
- "size": 0,
- "source": "apk",
- })
+ 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({
- "name": parts[0],
- "version": parts[1],
- "installed": True,
- "description": "",
- "size": 0,
- "source": "pacman",
- })
+ 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({
- "name": name,
- "version": versions.get(name, ""),
- "installed": name in installed,
- "description": description.strip(),
- "size": 0,
- "source": "apt",
- })
+ 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({
- "name": name,
- "version": version,
- "installed": name in installed,
- "description": description.strip(),
- "size": 0,
- "source": self._pkg_manager or "rpm",
- })
+ 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({
- "name": name,
- "version": version,
- "installed": name in installed,
- "description": description,
- "size": 0,
- "source": "apk",
- })
+ 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({
- "name": name,
- "version": version,
- "installed": name in installed,
- "description": description,
- "size": 0,
- "source": "pacman",
- })
+ 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({
- "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,
- })
+ 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({
- "name": m.group(1),
- "current_version": m.group(2),
- "new_version": m.group(3),
- })
+ 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({
- "name": m.group(1),
- "current_version": m.group(2),
- "new_version": m.group(3),
- })
+ 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({
- "name": m.group(2),
- "running": m.group(1) == "+",
- "enabled": False,
- "pid": 0,
- })
+ 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({
- "username": username,
- "uid": uid,
- "gid": gid,
- "home": home,
- "shell": shell,
- "groups": username_to_groups.get(username, []),
- })
+ users.append(
+ {
+ "username": username,
+ "uid": uid,
+ "gid": gid,
+ "home": home,
+ "shell": shell,
+ "groups": username_to_groups.get(username, []),
+ }
+ )
return users
# ------------------------------------------------------------------
@@ -1856,19 +1941,21 @@ class LinuxDriver(
command = parts[10]
except (ValueError, IndexError):
continue
- processes.append({
- "pid": pid,
- "ppid": ppid,
- "user": user,
- "cpu": cpu,
- "memory": mem,
- "vsz": vsz,
- "rss": rss,
- "tty": tty,
- "state": state,
- "started": started,
- "command": command,
- })
+ processes.append(
+ {
+ "pid": pid,
+ "ppid": ppid,
+ "user": user,
+ "cpu": cpu,
+ "memory": mem,
+ "vsz": vsz,
+ "rss": rss,
+ "tty": tty,
+ "state": state,
+ "started": started,
+ "command": command,
+ }
+ )
return processes
# ------------------------------------------------------------------
@@ -1909,14 +1996,14 @@ 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
comment = ""
if "#" in stripped:
idx = stripped.index("#")
- comment = stripped[idx + 1:].strip()
+ comment = stripped[idx + 1 :].strip()
stripped = stripped[:idx].strip()
if not stripped or stripped.startswith("@") or stripped.startswith("MAILTO"):
@@ -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:
diff --git a/pyproject.toml b/pyproject.toml
index ea47f3c..88679cb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -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"]
diff --git a/tests/test_command_channel.py b/tests/test_command_channel.py
index 66ef3bb..4072792 100644
--- a/tests/test_command_channel.py
+++ b/tests/test_command_channel.py
@@ -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
diff --git a/tests/test_fix_templates.py b/tests/test_fix_templates.py
index 51e820c..18c3a4c 100644
--- a/tests/test_fix_templates.py
+++ b/tests/test_fix_templates.py
@@ -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"
diff --git a/tests/test_linux.py b/tests/test_linux.py
index 7df3abe..30d0757 100644
--- a/tests/test_linux.py
+++ b/tests/test_linux.py
@@ -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([
- "DMIBEGIN",
- sys_vendor, product_name, product_version, product_serial,
- product_uuid, detect_virt, dt_model, cpuinfo_serial, cpuinfo_model,
- ])
+ return "\n".join(
+ [
+ "DMIBEGIN",
+ 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,20 +800,23 @@ class TestCollectPlatformInfo:
# ---------------------------------------------------------------------------
-@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"),
- ("NVIDIA Jetson Nano Developer Kit", "NVIDIA"),
- ("Rock Pi 4C", "Radxa"),
- ("Orange Pi 5 Plus", "Xunlong Software"),
- ("Banana Pi BPI-R3", "SinoVoip"),
- ("NanoPi R4S", "FriendlyElec"),
- ("PINE64 RockPro64", "Pine64"),
- ("BeagleBone Black", "BeagleBoard.org"),
- ("Unknown Board 1.0", "Unknown Board"),
- ("SomeSingleWordBoard", "SomeSingleWordBoard"),
-])
+@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"),
+ ("NVIDIA Jetson Nano Developer Kit", "NVIDIA"),
+ ("Rock Pi 4C", "Radxa"),
+ ("Orange Pi 5 Plus", "Xunlong Software"),
+ ("Banana Pi BPI-R3", "SinoVoip"),
+ ("NanoPi R4S", "FriendlyElec"),
+ ("PINE64 RockPro64", "Pine64"),
+ ("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 \n", f"{marker}0\nchris@host:~$ "])
+ channel = _Channel(["x \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()
diff --git a/tests/test_ssh_host_key.py b/tests/test_ssh_host_key.py
index bd1cdf0..35683a9 100644
--- a/tests/test_ssh_host_key.py
+++ b/tests/test_ssh_host_key.py
@@ -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
diff --git a/tests/test_sudo_off_the_command_line.py b/tests/test_sudo_off_the_command_line.py
index 98b07d3..321c72d 100644
--- a/tests/test_sudo_off_the_command_line.py
+++ b/tests/test_sudo_off_the_command_line.py
@@ -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""