Merge pull request 'chore: adopt the NAPALM driver standard' (#38) from chore/driver-standard into main

2026-10-23 12:55:55 +00:00
commit ca580fa394
@@ -0,0 +1,17 @@
 name: CI
 on:
  push:
    branches: [main, master]
    tags: ["v*"]
  pull_request:
 jobs:
  driver:
    uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
    with:
      standard-ref: main
      # Until napalm-device-types 4.22.0 (AptProxyMixin, #53) is on PyPI.
      pre-install: "git+https://git.netork.io/NAPALM/napalm-device-types.git"
    secrets:
      PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
@@ -1,48 +0,0 @@
 name: CI
 on:
  push:
    branches: ["**"]
  pull_request:
    branches: ["**"]
 jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      fail-fast: false
      matrix:
        python-version: ["3.10", "3.11", "3.12"]
    steps:
      - name: Checkout
        uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
          cache: pip
      - name: Install package with dev extras
        run: |
          python -m pip install --upgrade pip
          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
          python -m pip install -e ".[dev]"
      - name: Run unit tests
        run: |
          python -m pytest -q --tb=short
      - name: Build wheel and sdist
        run: |
          python -m pip install build
          python -m build
      - name: Upload dist artifacts
        # v4 refuses to run on Gitea ("not currently supported on GHES").
        uses: actions/upload-artifact@v3
        with:
          name: dist-${{ matrix.python-version }}
          path: dist/*
@@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 ## [0.15.0] - 2026-10-23
 The first release on PyPI. It collects everything since 0.4.0, which was the last version recorded here.
 ### Changed
 - The driver follows the NAPALM driver standard
  (https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
  tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
  a build check, and a release tag publishes to PyPI. The package now ships `py.typed`
  and a LICENSE file. **Python 3.9 is no longer supported**; napalm-device-types never
  ran on it.
 - Exceptions the driver deliberately ignores (the PTY width at login, closing a dead
  session, unreadable route lines, the snmpd.conf community and port) are now logged
  at debug level.
 ### Added
 - apt's proxy through napalm-device-types' `AptProxyMixin` (NetOrk/netork#1397). `get_package_proxy()` is what `apt-config dump` says, not `00proxy` alone: a proxy in `90curtin-aptproxy` or `/etc/apt/apt.conf` wins over it, and was missed. New: `get_package_proxy_report()` (every line that sets it, per-host overrides, the https sources) and `check_package_proxy(url, expect)` (a fetch with `apt-helper`, past any proxy). `configure_package_proxy(url)` now also comments out other proxy lines, reads the result back and fetches one `InRelease` through the proxy; it fails when apt still fetches elsewhere. Reads run on the exec channel. Requires napalm-device-types 4.22.0; version 0.14.0.
 - `get_host_firewall()` (napalm-device-types' `HostFirewallMixin`, NetOrk/netork#1344): the host's own netfilter rules — `nft -j list ruleset`, with the tables iptables manages taken from `iptables-save` — as chains and rules in one shape. Read-only, as root, on an exec channel (the nft JSON is one long line). Requires napalm-device-types 4.21.0; version 0.13.0.
@@ -0,0 +1,191 @@
                                 Apache License
                           Version 2.0, January 2004
                        http://www.apache.org/licenses/
   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
   1. Definitions.
      "License" shall mean the terms and conditions for use, reproduction,
      and distribution as defined by Sections 1 through 9 of this document.
      "Licensor" shall mean the copyright owner or entity authorized by
      the copyright owner that is granting the License.
      "Legal Entity" shall mean the union of the acting entity and all
      other entities that control, are controlled by, or are under common
      control with that entity. For the purposes of this definition,
      "control" means (i) the power, direct or indirect, to cause the
      direction or management of such entity, whether by contract or
      otherwise, or (ii) ownership of fifty percent (50%) or more of the
      outstanding shares, or (iii) beneficial ownership of such entity.
      "You" (or "Your") shall mean an individual or Legal Entity
      exercising permissions granted by this License.
      "Source" form shall mean the preferred form for making modifications,
      including but not limited to software source code, documentation
      source, and configuration files.
      "Object" form shall mean any form resulting from mechanical
      transformation or translation of a Source form, including but
      not limited to compiled object code, generated documentation,
      and conversions to other media types.
      "Work" shall mean the work of authorship made available under
      the License, as indicated by a copyright notice that is included in
      or attached to the work (an example is provided in the Appendix below).
      "Derivative Works" shall mean any work, whether in Source or Object
      form, that is based on (or derived from) the Work and for which the
      editorial revisions, annotations, elaborations, or other transformations
      represent, as a whole, an original work of authorship. For the purposes
      of this License, Derivative Works shall not include works that remain
      separable from, or merely link (or bind by name) to the interfaces of,
      the Work and Derivative Works thereof.
      "Contribution" shall mean, as submitted to the Licensor for inclusion
      in the Work by the copyright owner or by an individual or Legal Entity
      authorized to submit on behalf of the copyright owner. For the purposes
      of this definition, "submitted" means any form of electronic, verbal,
      or written communication sent to the Licensor or its representatives,
      including but not limited to communication on electronic mailing lists,
      source code control systems, and issue tracking systems that are managed
      by, or on behalf of, the Licensor for the purpose of discussing and
      improving the Work, but excluding communication that is conspicuously
      marked or designated in writing by the copyright owner as "Not a
      Contribution."
      "Contributor" shall mean Licensor and any Legal Entity on behalf of
      whom a Contribution has been received by the Licensor and incorporated
      within the Work.
   2. Grant of Copyright License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      copyright license to reproduce, prepare Derivative Works of,
      publicly display, publicly perform, sublicense, and distribute the
      Work and such Derivative Works in Source or Object form.
   3. Grant of Patent License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      (except as stated in this section) patent license to make, have made,
      use, offer to sell, sell, import, and otherwise transfer the Work,
      where such license applies only to those patent claims licensable
      by such Contributor that are necessarily infringed by their
      Contribution(s) alone or by the combined Contribution(s) and the Work
      to which such Contribution(s) was submitted. If You institute patent
      litigation against any entity (including a cross-claim or counterclaim
      in a lawsuit) alleging that the Work or any Contribution embodied
      within the Work constitutes direct or contributory patent infringement,
      then any patent licenses granted to You under this License for that
      Work shall terminate as of the date such litigation is filed.
   4. Redistribution. You may reproduce and distribute copies of the
      Work or Derivative Works thereof in any medium, with or without
      modifications, and in Source or Object form, provided that You
      meet the following conditions:
      (a) You must give any other recipients of the Work or Derivative
          Works a copy of this License; and
      (b) You must cause any modified files to carry prominent notices
          stating that You changed the files; and
      (c) You must retain, in the Source form of any Derivative Works
          that You distribute, all copyright, patent, trademark, and
          attribution notices from the Source form of the Work,
          excluding those notices that do not pertain to any part of
          the Derivative Works; and
      (d) If the Work includes a "NOTICE" text file as part of its
          distribution, You must include a readable copy of the
          attribution notices contained within such NOTICE file, in
          at least one of the following places: within a NOTICE text file
          distributed as part of the Derivative Works; within the Source
          form or documentation, if provided along with the Derivative
          Works; or, within a display generated by the Derivative Works,
          if and wherever such third-party notices normally appear. The
          contents of the NOTICE file are for informational purposes only
          and do not modify the License. You may add Your own attribution
          notices within Derivative Works that You distribute, alongside
          or as an addendum to the NOTICE text from the Work, provided
          that such additional attribution notices cannot be construed
          as modifying the License.
      You may add Your own license statement for Your modifications and
      may provide additional grant of rights to use, copy, modify, merge,
      publish, distribute, sublicense, and/or sell copies of the Work,
      and to permit persons to whom the Work is furnished to do so,
      subject to the following conditions:
   5. Submission of Contributions. Unless You explicitly state otherwise,
      any Contribution intentionally submitted for inclusion in the Work
      by You to the Licensor shall be under the terms and conditions of
      this License, without any additional terms or conditions.
      Notwithstanding the above, nothing herein shall supersede or modify
      the terms of any separate license agreement you may have executed
      with Licensor regarding such Contributions.
   6. Trademarks. This License does not grant permission to use the trade
      names, trademarks, service marks, or product names of the Licensor,
      except as required for reasonable and customary use in describing the
      origin of the Work and reproducing the content of the NOTICE file.
   7. Disclaimer of Warranty. Unless required by applicable law or
      agreed to in writing, Licensor provides the Work (and each
      Contributor provides its Contributions) on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
      implied, including, without limitation, any conditions of TITLE,
      MERCHANTIBILITY, or FITNESS FOR A PARTICULAR PURPOSE.
      See the License for the specific language governing permissions and
      limitations under the License.
   8. Limitation of Liability. In no event and under no legal theory,
      whether in tort (including negligence), contract, or otherwise,
      unless required by applicable law (such as deliberate and grossly
      negligent acts) or agreed to in writing, shall any Contributor be
      liable to You for damages, including any direct, indirect, special,
      incidental, or exemplary damages of any character arising as a
      result of this License or out of the use or inability to use the
      Work (even if such Contributor has been advised of the possibility
      of such damages).
   9. Accepting Warranty or Additional Liability. While redistributing
      the Work or Derivative Works thereof, You may choose to offer,
      and charge a fee for, acceptance of support, warranty, indemnity,
      or other liability obligations and/or rights consistent with this
      License. However, in accepting such obligations, You may offer only
      conditions consistent with this License, subject to the following
      additional conditions:
      (a) You may offer additional warranty, indemnity, or other liability
          terms and conditions; and
      (b) You may offer support for the Work in return for a fee.
   END OF TERMS AND CONDITIONS
   APPENDIX: How to apply the Apache License to your work.
      To apply the Apache License to your work, attach the following
      boilerplate notice, with the fields enclosed by brackets "[]"
      replaced with your own identifying information. (Don't include
      the brackets!)  The text should be enclosed in the appropriate
      comment syntax for the file format in question. It may also be
      possible to make it available with a creative commons license.
   Copyright 2026 Christian Manivong
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at
       http://www.apache.org/licenses/LICENSE-2.0
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
@@ -24,7 +24,7 @@ pip install napalm napalm-linux
 Or from source:
 ```bash
-git clone https://github.com/chrismanivong/napalm-linux
+git clone https://git.netork.io/NAPALM/napalm-linux
 pip install -e napalm-linux/
 ```
@@ -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
@@ -1,4 +1,3 @@
 # -*- coding: utf-8 -*-
 # Licensed under the Apache License, Version 2.0
 """NAPALM driver for generic Linux systems.
@@ -21,20 +20,15 @@ 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 napalm.base.exceptions import ConnectionClosedException, ConnectionException
 from netmiko.exceptions import (
    NetmikoAuthenticationException,
    NetmikoTimeoutException,
 )
 from napalm.base.exceptions import ConnectionException, ConnectionClosedException
 from napalm.base.netmiko_helpers import netmiko_args
 from napalm_device_types import (
    APT_UPGRADABLE_COMMAND,
    DNF_SECURITY_COMMAND,
    AptProxyMixin,
    ByteStream,
    CommandResult,
    ContainerEngineMixin,
@@ -46,22 +40,20 @@ from napalm_device_types import (
    ListeningSocketsMixin,
    OpenSshServerMixin,
    OSDriver,
    AptProxyMixin,
    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:
@@ -173,25 +171,35 @@ _NO_STDIN = "exec </dev/null; "
 #: Where ``_install_file_as_root`` stages content: a fresh file from ``mktemp``
 #: under ``umask 077``, never a fixed name another user could create first.
-_STAGE_TEMPLATE = "/tmp/netork-stage.XXXXXX"
+#: The path is on the device, not a local temp file.
 _STAGE_TEMPLATE = "/tmp/netork-stage.XXXXXX"  # nosec B108
 _STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
 _STAGE = (
    f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
    '{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
 )
-#: What to do when sudo wants a password netOrk does not have.
+#: What to do when sudo wants a password netOrk does not have (a message, not a password).
-_SUDO_PASSWORD_HINT = (
+_SUDO_PASSWORD_HINT = (  # nosec B105
    "sudo requires a password on this host and no sudo password was given: pass one, "
    "or allow this user passwordless sudo (NOPASSWD)."
 )
 # DMI field values that carry no useful information (OEM defaults, blanks)
-_BAD_DMI: frozenset[str] = frozenset({
+_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",
+        "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 +218,28 @@ _VIRT_VENDOR_MAP: dict[str, str] = {
 }
 # Container technologies reported by systemd-detect-virt
-_CONTAINER_VIRT: frozenset[str] = frozenset({
+_CONTAINER_VIRT: frozenset[str] = frozenset(
-    "docker", "podman", "lxc", "lxc-libvirt", "systemd-nspawn",
+    {
-})
+        "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({
+_VM_DMI_VENDORS: frozenset[str] = frozenset(
-    "qemu", "vmware, inc.", "microsoft corporation",
+    {
-    "innotek gmbh", "xen", "bochs",
+        "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]] = [
@@ -255,7 +275,10 @@ def _arm_vendor_from_model(model: str) -> str:
 #: Where net-snmp reads its configuration, and how each package manager installs it.
 _SNMPD_CONF = "/etc/snmp/snmpd.conf"
 _SNMP_INSTALL = {
-    "apt": "apt-get update -y >/dev/null 2>&1; DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
+    "apt": (
        "apt-get update -y >/dev/null 2>&1; "
        "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1"
    ),
    "dnf": "dnf install -y net-snmp net-snmp-utils 2>&1",
    "yum": "yum install -y net-snmp net-snmp-utils 2>&1",
    "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1",
@@ -394,13 +417,17 @@ 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
+            logger.debug("%s: ignored", self.hostname, exc_info=True)
        self._pkg_manager = self._forced_pkg_manager or self._detect_pkg_manager()
        logger.debug("Connected to %s, pkg_manager=%s", self.hostname, self._pkg_manager)
@@ -411,7 +438,7 @@ class LinuxDriver(
            try:
                self._device.disconnect()
            except Exception:
-                pass
+                logger.debug("%s: ignored", self.hostname, exc_info=True)
            self._device = None
        self._pkg_manager = None
@@ -419,7 +446,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}
@@ -471,7 +498,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))
@@ -639,13 +670,15 @@ class LinuxDriver(
            "u=$(cat /sys/class/dmi/id/product_uuid 2>/dev/null); "
            "d=$(systemd-detect-virt 2>/dev/null); d=${d:-none}; "
            "dt=$(tr -d '\\0' </sys/firmware/devicetree/base/model 2>/dev/null); "
-            "cs=$(grep '^Serial' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 | xargs 2>/dev/null); "
+            "cs=$(grep '^Serial' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 "
-            "cm=$(grep '^Model' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 | xargs 2>/dev/null); "
+            "| xargs 2>/dev/null); "
            "cm=$(grep '^Model' /proc/cpuinfo 2>/dev/null | head -1 | cut -d: -f2 "
            "| xargs 2>/dev/null); "
            # DMIBEGIN sentinel: _send() strips leading blank lines (ARM has no DMI
            # 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()
@@ -672,10 +705,7 @@ class LinuxDriver(
        cpuinfo_mdl = _clean(8)
        is_container = detect_virt in _CONTAINER_VIRT
-        is_vm = (
+        is_vm = detect_virt not in ("none", "") or sys_vendor.lower() in _VM_DMI_VENDORS
            detect_virt not in ("none", "")
            or sys_vendor.lower() in _VM_DMI_VENDORS
        )
        if is_container:
            return {
@@ -741,7 +771,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).
@@ -767,7 +799,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
@@ -787,8 +819,8 @@ class LinuxDriver(
            return {}
        output = self._send("lldpctl -f keyvalue 2>/dev/null || true")
-        neighbors: Dict[str, List[dict[str, Any]]] = {}
+        neighbors: dict[str, list[dict[str, Any]]] = {}
-        entries: Dict[str, Dict[str, str]] = {}
+        entries: dict[str, dict[str, str]] = {}
        for line in output.splitlines():
            line = line.strip()
@@ -897,14 +929,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
@@ -915,13 +949,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",
@@ -935,7 +969,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,
@@ -970,7 +1006,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"
@@ -998,9 +1038,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] = {
+        _frr_code: dict[str, str] = {
-            "O": "ospf", "B": "bgp", "R": "rip", "I": "isis",
+            "O": "ospf",
-            "S": "static", "K": "connected", "C": "connected",
+            "B": "bgp",
            "R": "rip",
            "I": "isis",
            "S": "static",
            "K": "connected",
            "C": "connected",
        }
        if self._send("command -v vtysh 2>/dev/null").strip():
            try:
@@ -1034,27 +1079,27 @@ class LinuxDriver(
                                _make_entry(mapped_proto, nexthop, iface, metric, prefix)
                            )
            except Exception:
-                pass
+                logger.debug("%s: ignored", self.hostname, exc_info=True)
        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(
+            m = re.match(r"^([\d.]+)\s+dev\s+(\S+)\s+lladdr\s+([\da-f:]+)\s+(\S+)", line)
                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(
@@ -1131,9 +1176,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(
+        rtt_m = re.search(r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output)
            r"rtt .* = ([\d.]+)/([\d.]+)/([\d.]+)/([\d.]+) ms", output
        )
        results = []
        for line in output.splitlines():
@@ -1141,10 +1184,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": {
@@ -1231,9 +1276,7 @@ class LinuxDriver(
            return self._get_packages_apk()
        if self._pkg_manager == "pacman":
            return self._get_packages_pacman()
-        raise NotImplementedError(
+        raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
            f"Package manager '{self._pkg_manager}' is not supported"
        )
    def _get_packages_apt(self) -> list[PackageDict]:
        out = self._send(
@@ -1249,7 +1292,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.
@@ -1267,7 +1312,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,
@@ -1276,26 +1322,32 @@ class LinuxDriver(
                    "source": "apt",
                    "source_package": source_package,
                    "source_version": source_version,
-            })
+                }
            )
        return packages
    def _get_packages_rpm(self) -> list[PackageDict]:
        out = self._send(
-            "rpm -qa --queryformat '%{NAME}\\t%{VERSION}-%{RELEASE}\\t%{SIZE}\\t%{SUMMARY}\\n' 2>/dev/null"
+            "rpm -qa --queryformat "
            "'%{NAME}\\t%{VERSION}-%{RELEASE}\\t%{SIZE}\\t%{SUMMARY}\\n' 2>/dev/null"
        )
        packages: list[PackageDict] = []
        for line in out.splitlines():
            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]:
@@ -1306,14 +1358,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]:
@@ -1323,27 +1377,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] = {}
+            versions: dict[str, str] = {}
-            ver_out = self._send(f"apt-cache show {safe_q} 2>/dev/null | grep -E '^(Package|Version):' || true")
+            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:"):
@@ -1355,14 +1414,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"
@@ -1373,14 +1434,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")
@@ -1393,14 +1456,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")
@@ -1415,14 +1480,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
@@ -1432,10 +1499,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":
@@ -1445,7 +1515,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.
@@ -1477,6 +1557,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":
@@ -1538,9 +1619,7 @@ class LinuxDriver(
            return self._get_updates_apk()
        if self._pkg_manager == "pacman":
            return self._get_updates_pacman()
-        raise NotImplementedError(
+        raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
            f"Package manager '{self._pkg_manager}' is not supported"
        )
    def _get_updates_apt(self) -> list[UpdateDict]:
        # apt list --upgradable does not need root; avoid sudo so it works even
@@ -1550,23 +1629,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:
@@ -1604,11 +1691,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]:
@@ -1618,11 +1707,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
    # ------------------------------------------------------------------
@@ -1630,9 +1721,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
@@ -1650,30 +1741,27 @@ class LinuxDriver(
            return self._apply_updates_apk(packages)
        if self._pkg_manager == "pacman":
            return self._apply_updates_pacman(packages)
-        raise NotImplementedError(
+        raise NotImplementedError(f"Package manager '{self._pkg_manager}' is not supported")
            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"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1"
-            f"{pkg_args} 2>&1"
+            if packages
-            if packages else
+            else "DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1"
            "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)
@@ -1684,16 +1772,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"
@@ -1709,7 +1797,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"
@@ -1753,12 +1841,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
    # ------------------------------------------------------------------
@@ -1770,20 +1860,8 @@ class LinuxDriver(
        passwd_out = self._send("getent passwd 2>/dev/null || cat /etc/passwd")
        groups_out = self._send("getent group 2>/dev/null || cat /etc/group")
-        # Build uid→[group] map from /etc/group
+        # username → supplementary groups, joined with passwd below
-        uid_to_groups: Dict[int, List[str]] = {}
+        username_to_groups: dict[str, list[str]] = {}
        for line in groups_out.splitlines():
            parts = line.split(":")
            if len(parts) < 4:
                continue
            gname = parts[0]
            members = [m.strip() for m in parts[3].split(",") if m.strip()]
            for member in members:
                # We'll convert username→uid below; collect by username first
                uid_to_groups.setdefault(-1, [])  # placeholder
        # Simpler: collect username→groups, then join with passwd
        username_to_groups: Dict[str, List[str]] = {}
        for line in groups_out.splitlines():
            parts = line.split(":")
            if len(parts) < 4:
@@ -1803,14 +1881,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
    # ------------------------------------------------------------------
@@ -1846,7 +1926,8 @@ class LinuxDriver(
                command = parts[10]
            except (ValueError, IndexError):
                continue
-            processes.append({
+            processes.append(
                {
                    "pid": pid,
                    "ppid": ppid,
                    "user": user,
@@ -1858,7 +1939,8 @@ class LinuxDriver(
                    "state": state,
                    "started": started,
                    "command": command,
-            })
+                }
            )
        return processes
    # ------------------------------------------------------------------
@@ -1897,16 +1979,14 @@ class LinuxDriver(
        return jobs
    @staticmethod
-    def _parse_cron_line(
+    def _parse_cron_line(line: str, source_user: str, has_user_field: bool) -> CronJobDict | None:
        line: str, source_user: str, has_user_field: bool
    ) -> Optional[CronJobDict]:
        """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"):
@@ -1936,12 +2016,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()
+                self._send("systemctl is-active snmpd 2>/dev/null || true").strip() == "active"
                == "active"
            )
            if not running:
                return None
@@ -1968,7 +2047,7 @@ class LinuxDriver(
                        community = parts[3]
                        break
            except Exception:
-                pass
+                logger.debug("%s: ignored", self.hostname, exc_info=True)
            # Detect port override
            try:
@@ -1976,11 +2055,11 @@ 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:
-                pass
+                logger.debug("%s: ignored", self.hostname, exc_info=True)
            return SNMPConfigDict(running=True, community=community, port=port, version="2c")
        except Exception as exc:
@@ -2067,7 +2146,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)])
@@ -2092,15 +2173,19 @@ class LinuxDriver(
        raw = self._sudo("wg show all dump 2>&1 || true")
        _perm_errors = (
-            "operation not permitted", "permission denied",
+            "operation not permitted",
-            "a password is required", "a terminal is required",
+            "permission denied",
-            "command not found", "not found", "no such file",
+            "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:
@@ -2112,8 +2197,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()
@@ -2172,13 +2266,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:
@@ -1,14 +1,14 @@
 [build-system]
-requires = ["setuptools>=68", "wheel"]
+requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"
 [project]
 name = "napalm-linux"
-version = "0.14.0"
+version = "0.15.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
-license = { text = "Apache-2.0" }
+license = "Apache-2.0"
 authors = [
    { name = "Christian Manivong", email = "christian@manivong.de" },
 ]
@@ -24,13 +24,12 @@ classifiers = [
    "Development Status :: 3 - Alpha",
    "Intended Audience :: Developers",
    "Intended Audience :: System Administrators",
    "License :: OSI Approved :: Apache Software License",
    "Operating System :: OS Independent",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.9",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Programming Language :: Python :: 3.13",
    "Topic :: System :: Networking",
    "Topic :: System :: Systems Administration",
    "Typing :: Typed",
@@ -44,23 +43,58 @@ dependencies = [
 [project.optional-dependencies]
 dev = [
-    "pytest",
+    "pytest>=8",
-    "pytest-cov",
+    "pytest-cov>=5",
-    "black",
+    "ruff>=0.6",
-    "ruff",
+    "mypy>=1.11",
    "mypy",
 ]
 [project.entry-points."napalm.drivers"]
 linux = "napalm_linux:LinuxDriver"
 [project.urls]
-Repository = "https://github.com/chrismanivong/napalm-linux"
+Repository = "https://git.netork.io/NAPALM/napalm-linux"
 Issues = "https://git.netork.io/NAPALM/napalm-linux/issues"
 Changelog = "https://git.netork.io/NAPALM/napalm-linux/src/branch/main/CHANGELOG.md"
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["napalm_linux*"]
 [tool.setuptools.package-data]
 napalm_linux = ["py.typed"]
 [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"]
 [tool.mypy]
 python_version = "3.10"
 strict = true
 # napalm itself is untyped, so every driver subclasses Any.
 disallow_subclassing_any = false
 [[tool.mypy.overrides]]
 module = ["napalm.*", "netmiko.*", "paramiko.*"]
 ignore_missing_imports = true
 [[tool.mypy.overrides]]
 # Not yet under strict mypy (quality baseline); may only shrink.
 module = ["napalm_linux.linux"]
 ignore_errors = true
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 [tool.coverage.run]
 source = ["napalm_linux"]
 [tool.coverage.report]
 fail_under = 57
@@ -0,0 +1,28 @@
 {
 "coverage_floor": 57.0,
 "function_complexity": {
  "napalm_linux/linux.py::LinuxDriver._collect_platform_info": 24,
  "napalm_linux/linux.py::LinuxDriver._parse_cron_line": 12,
  "napalm_linux/linux.py::LinuxDriver.get_networks": 13,
  "napalm_linux/linux.py::LinuxDriver.get_route_to": 23,
  "napalm_linux/linux.py::LinuxDriver.get_snmp_config": 13,
  "napalm_linux/linux.py::LinuxDriver.get_vpn_tunnels": 28,
  "napalm_linux/linux.py::LinuxDriver.search_packages": 26
 },
 "function_lines": {
  "napalm_linux/linux.py::LinuxDriver._collect_platform_info": 106,
  "napalm_linux/linux.py::LinuxDriver.get_networks": 52,
  "napalm_linux/linux.py::LinuxDriver.get_route_to": 138,
  "napalm_linux/linux.py::LinuxDriver.get_vpn_tunnels": 152,
  "napalm_linux/linux.py::LinuxDriver.ping": 52,
  "napalm_linux/linux.py::LinuxDriver.search_packages": 106,
  "napalm_linux/linux.py::LinuxDriver.uninstall_package": 51
 },
 "mypy_relaxed": [
  "napalm_linux.linux"
 ],
 "python_file_lines": {
  "napalm_linux/linux.py": 2307
 },
 "untyped_public": []
 }
@@ -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
@@ -248,7 +248,7 @@ def test_every_annotation_resolves():
    import inspect
    import typing
-    for name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
+    for _name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
        if member.__module__ == "napalm_linux.linux":
            typing.get_type_hints(member)
@@ -263,7 +263,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"
@@ -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 = (
+GROUP_OUT = "sudo:x:27:admin\ndocker:x:999:admin\nadm:x:4:admin\n"
    "sudo:x:27:admin\n"
    "docker:x:999:admin\n"
    "adm: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"
+    "Last metadata expiration check: 0:01:23 ago.\nError: No match for argument: nonexistent-pkg\n"
    "Error: 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,
+            sys_vendor,
-        product_uuid, detect_virt, dt_model, cpuinfo_serial, cpuinfo_model,
+            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",
+            "Dell Inc.",
-            "8a2e3f00-dead-beef-0000-123456789abc", "none",
+            "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",
+            "LENOVO",
-            "8a2e3f00-dead-beef-0000-123456789abc", "none",
+            "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", "",
+            "QEMU",
-            "4c4c4544-0000-2010-8020-b4c04f534a31", "kvm",
+            "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",
+            "VMware, Inc.",
-            "4244560c-dead-beef-0000-abcdef123456", "vmware",
+            "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", "",
+            "Microsoft Corporation",
-            "7C5B4B1F-1234-5678-ABCD-000000000001", "microsoft",
+            "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.",
-            "To Be Filled By O.E.M.", "", "none",
+            "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)", "", "",
+            "QEMU",
-            "4c4c4544-0000-2010-8020-b4c04f534a31", "none",
+            "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)", "", "",
+            "QEMU",
-            "4c4c4544-0000-2010-8020-b4c04f534a31", "docker",
+            "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",
+            "Dell Inc.",
-            "8a2e3f00-dead-beef-0000-123456789abc", "none",
+            "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}
+    platform = {
-    with patch.object(driver, "_collect_platform_info", return_value=platform), \
+        "vendor": "Dell Inc.",
-         patch.object(driver, "_parse_uptime", return_value=86400), \
+        "model": "PowerEdge R720",
-         patch.object(driver, "_send", side_effect=["myhost", "myhost.example.com", "Debian GNU/Linux 12", "eth0\neth1",
+        "serial": "ABC123",
-                                                   "6.1.0-18-amd64"]):
+        "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",
+        "vendor": "KVM",
-        "serial": "4c4c4544-0000-2010-8020-b4c04f534a31", "is_vm": True,
+        "model": "Virtual Machine",
        "serial": "4c4c4544-0000-2010-8020-b4c04f534a31",
        "is_vm": True,
    }
-    with patch.object(driver, "_collect_platform_info", return_value=platform), \
+    with (
-         patch.object(driver, "_parse_uptime", return_value=3600), \
+        patch.object(driver, "_collect_platform_info", return_value=platform),
-         patch.object(driver, "_send", side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0",
+        patch.object(driver, "_parse_uptime", return_value=3600),
-                                                   "5.15.0-91-generic"]):
+        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), \
+    with (
-         patch.object(driver, "_parse_uptime", return_value=0), \
+        patch.object(driver, "_collect_platform_info", return_value=platform),
-         patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0", "6.6.7-0-lts"]):
+        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"),
+                "openssl-libs.x86_64   1:1.0.2k-26.el7_9   updates\n"
-            ("RHSA-2023:3556 Important/Sec. openssl-libs-1:1.0.2k-26.el7_9.x86_64\n"
+                "vim-enhanced.x86_64   2:7.4.629-8.el7_9   updates\n__NETORK_RC=100"
-            "__DNF_RC=0\n__NETORK_RC=0"),
+            ),
            (
                "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()
@@ -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
@@ -76,7 +78,7 @@ def test_a_changed_key_is_refused():
        try:
            raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
        except paramiko.SSHException as exc:  # what netmiko does with it
-            raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
+            raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc
    driver = _driver(seen, SshHostKey.of(OTHER_KEY))
    with patch("napalm_linux.linux.ConnectHandler", side_effect=refuse):
@@ -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""