chore: adopt the NAPALM driver standard #38

Merged
christianmanivong merged 6 commits from chore/driver-standard into main 2026-10-23 12:55:55 +00:00
14 changed files with 903 additions and 449 deletions

No files matched your search

+17
View File
@@ -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 }}
-48
View File
@@ -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/*
+15
View File
@@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased] ## [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 ### 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. - 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. - `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.
+191
View File
@@ -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.
+6 -6
View File
@@ -24,7 +24,7 @@ pip install napalm napalm-linux
Or from source: Or from source:
```bash ```bash
git clone https://github.com/chrismanivong/napalm-linux git clone https://git.netork.io/NAPALM/napalm-linux
pip install -e napalm-linux/ pip install -e napalm-linux/
``` ```
@@ -58,11 +58,11 @@ with Driver(
# OS-specific methods (from napalm-device-types OSDriver) # OS-specific methods (from napalm-device-types OSDriver)
packages = dev.get_packages() packages = dev.get_packages()
updates = dev.get_pending_updates() updates = dev.get_pending_updates()
services = dev.get_services() services = dev.get_services()
users = dev.get_os_users() users = dev.get_os_users()
procs = dev.get_processes() procs = dev.get_processes()
jobs = dev.get_cron_jobs() jobs = dev.get_cron_jobs()
# Upgrade specific packages # Upgrade specific packages
result = dev.apply_updates(["openssh-server", "curl"]) result = dev.apply_updates(["openssh-server", "curl"])
@@ -121,7 +121,7 @@ apt → dnf → yum → apk → pacman
Force a specific package manager: Force a specific package manager:
```python ```python
optional_args={"pkg_manager": "dnf"} optional_args = {"pkg_manager": "dnf"}
``` ```
## SSH user permissions ## SSH user permissions
+393 -298
View File
File diff suppressed because it is too large. Load diff
View File
Whitespace-only changes.
+47 -13
View File
@@ -1,14 +1,14 @@
[build-system] [build-system]
requires = ["setuptools>=68", "wheel"] requires = ["setuptools>=77"]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[project] [project]
name = "napalm-linux" name = "napalm-linux"
version = "0.14.0" version = "0.15.0"
description = "NAPALM driver for generic Linux systems via SSH" description = "NAPALM driver for generic Linux systems via SSH"
readme = "README.md" readme = "README.md"
requires-python = ">=3.9" requires-python = ">=3.10"
license = { text = "Apache-2.0" } license = "Apache-2.0"
authors = [ authors = [
{ name = "Christian Manivong", email = "christian@manivong.de" }, { name = "Christian Manivong", email = "christian@manivong.de" },
] ]
@@ -24,13 +24,12 @@ classifiers = [
"Development Status :: 3 - Alpha", "Development Status :: 3 - Alpha",
"Intended Audience :: Developers", "Intended Audience :: Developers",
"Intended Audience :: System Administrators", "Intended Audience :: System Administrators",
"License :: OSI Approved :: Apache Software License",
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: System :: Networking", "Topic :: System :: Networking",
"Topic :: System :: Systems Administration", "Topic :: System :: Systems Administration",
"Typing :: Typed", "Typing :: Typed",
@@ -44,23 +43,58 @@ dependencies = [
[project.optional-dependencies] [project.optional-dependencies]
dev = [ dev = [
"pytest", "pytest>=8",
"pytest-cov", "pytest-cov>=5",
"black", "ruff>=0.6",
"ruff", "mypy>=1.11",
"mypy",
] ]
[project.entry-points."napalm.drivers"] [project.entry-points."napalm.drivers"]
linux = "napalm_linux:LinuxDriver" linux = "napalm_linux:LinuxDriver"
[project.urls] [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] [tool.setuptools.packages.find]
where = ["."] where = ["."]
include = ["napalm_linux*"] include = ["napalm_linux*"]
[tool.setuptools.package-data]
napalm_linux = ["py.typed"]
[tool.ruff] [tool.ruff]
line-length = 100 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
+28
View File
@@ -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": []
}
+3 -1
View File
@@ -70,7 +70,9 @@ class FakeTransport:
def _driver(*, root=False, sudo_password=None): def _driver(*, root=False, sudo_password=None):
driver = LinuxDriver("h", "u", "p", optional_args={"sudo_password": sudo_password}) driver = LinuxDriver("h", "u", "p", optional_args={"sudo_password": sudo_password})
transport = FakeTransport() 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 driver._root = root
return driver, transport return driver, transport
+4 -2
View File
@@ -248,7 +248,7 @@ def test_every_annotation_resolves():
import inspect import inspect
import typing 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": if member.__module__ == "napalm_linux.linux":
typing.get_type_hints(member) typing.get_type_hints(member)
@@ -263,7 +263,9 @@ class TestManagementPeer:
return d return d
def test_the_client_side_of_ssh_connection(self): 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): def test_an_ipv6_client(self):
assert self._driver("fd00::20 51234 fd00::5 22\n").get_management_peer() == "fd00::20" assert self._driver("fd00::20 51234 fd00::5 22\n").get_management_peer() == "fd00::20"
+194 -78
View File
@@ -1,9 +1,10 @@
"""Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed).""" """Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
import re import re
from unittest.mock import MagicMock, patch
import pytest import pytest
from unittest.mock import MagicMock, patch
from napalm_linux.linux import ( from napalm_linux.linux import (
_YUM_SECURITY_COMMAND, _YUM_SECURITY_COMMAND,
LinuxDriver, LinuxDriver,
@@ -11,7 +12,6 @@ from napalm_linux.linux import (
_split_status, _split_status,
) )
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Fixture – driver without a real connection # Fixture – driver without a real connection
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -327,11 +327,7 @@ PASSWD_OUT = (
"daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n" "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): def test_get_os_users(driver):
@@ -477,8 +473,7 @@ DNF_UPGRADE_SUCCESS = (
) )
DNF_UPGRADE_ERROR = ( 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(): def test_type_label_overrides_os_driver():
from napalm_device_types import OSDriver from napalm_device_types import OSDriver
assert OSDriver.TYPE_LABEL == "OS" assert OSDriver.TYPE_LABEL == "OS"
assert LinuxDriver.TYPE_LABEL != OSDriver.TYPE_LABEL 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. # Mirrors the real shell output: DMIBEGIN sentinel followed by 9 fields.
# The sentinel prevents _send()'s .strip() from eating leading blank lines # The sentinel prevents _send()'s .strip() from eating leading blank lines
# (which occur on ARM where all DMI files are absent). # (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, "DMIBEGIN",
product_uuid, detect_virt, dt_model, cpuinfo_serial, cpuinfo_model, sys_vendor,
]) product_name,
product_version,
product_serial,
product_uuid,
detect_virt,
dt_model,
cpuinfo_serial,
cpuinfo_model,
]
)
class TestCollectPlatformInfo: class TestCollectPlatformInfo:
def test_baremetal_dell(self, driver): def test_baremetal_dell(self, driver):
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -577,8 +586,12 @@ class TestCollectPlatformInfo:
def test_baremetal_lenovo_product_version_preferred(self, driver): def test_baremetal_lenovo_product_version_preferred(self, driver):
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -589,8 +602,12 @@ class TestCollectPlatformInfo:
def test_vm_kvm(self, driver): def test_vm_kvm(self, driver):
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -601,8 +618,12 @@ class TestCollectPlatformInfo:
def test_vm_vmware(self, driver): def test_vm_vmware(self, driver):
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -613,8 +634,12 @@ class TestCollectPlatformInfo:
def test_vm_hyperv(self, driver): def test_vm_hyperv(self, driver):
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -625,8 +650,12 @@ class TestCollectPlatformInfo:
def test_junk_dmi_values_filtered(self, driver): def test_junk_dmi_values_filtered(self, driver):
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -637,8 +666,12 @@ class TestCollectPlatformInfo:
def test_vm_kvm_fallback_via_dmi_when_detect_virt_unavailable(self, driver): def test_vm_kvm_fallback_via_dmi_when_detect_virt_unavailable(self, driver):
# systemd-detect-virt returns "none" (not installed), sys_vendor reveals QEMU # systemd-detect-virt returns "none" (not installed), sys_vendor reveals QEMU
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -648,8 +681,12 @@ class TestCollectPlatformInfo:
def test_container_docker(self, driver): def test_container_docker(self, driver):
raw = _dmi_output( 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): with patch.object(driver, "_send", return_value=raw):
info = driver._collect_platform_info() info = driver._collect_platform_info()
@@ -674,7 +711,12 @@ class TestCollectPlatformInfo:
def test_arm_device_tree_fallback(self, driver): def test_arm_device_tree_fallback(self, driver):
raw = _dmi_output( raw = _dmi_output(
"", "", "", "", "", "none", "",
"",
"",
"",
"",
"none",
dt_model="Raspberry Pi 4 Model B Rev 1.4", dt_model="Raspberry Pi 4 Model B Rev 1.4",
cpuinfo_serial="100000002a6d96dc", cpuinfo_serial="100000002a6d96dc",
cpuinfo_model="Raspberry Pi 4 Model B Rev 1.4", 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): def test_arm_cpuinfo_model_when_no_dt(self, driver):
raw = _dmi_output( raw = _dmi_output(
"", "", "", "", "", "none", "",
"",
"",
"",
"",
"none",
dt_model="", dt_model="",
cpuinfo_serial="0000000012345678", cpuinfo_serial="0000000012345678",
cpuinfo_model="Raspberry Pi 3 Model B Plus Rev 1.3", 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. # d="none\nnone" which shifted subsequent lines off by one.
# With the ${d:-none} fix this no longer happens. # With the ${d:-none} fix this no longer happens.
raw = _dmi_output( raw = _dmi_output(
"", "", "", "", "", "none", "",
"",
"",
"",
"",
"none",
dt_model="", dt_model="",
cpuinfo_serial="0000000012345678", cpuinfo_serial="0000000012345678",
cpuinfo_model="Raspberry Pi 3 Model B Rev 1.2", 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): def test_arm_fallback_ignored_when_dmi_present(self, driver):
raw = _dmi_output( 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", dt_model="some-dt-model",
cpuinfo_serial="deadbeef", cpuinfo_serial="deadbeef",
cpuinfo_model="some cpuinfo model", cpuinfo_model="some cpuinfo model",
@@ -744,20 +800,23 @@ class TestCollectPlatformInfo:
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@pytest.mark.parametrize("model,expected", [ @pytest.mark.parametrize(
("Raspberry Pi 4 Model B Rev 1.4", "Raspberry Pi Foundation"), "model,expected",
("Raspberry Pi 3 Model B Rev 1.2", "Raspberry Pi Foundation"), [
("ODROID-N2L", "Hardkernel"), ("Raspberry Pi 4 Model B Rev 1.4", "Raspberry Pi Foundation"),
("NVIDIA Jetson Nano Developer Kit", "NVIDIA"), ("Raspberry Pi 3 Model B Rev 1.2", "Raspberry Pi Foundation"),
("Rock Pi 4C", "Radxa"), ("ODROID-N2L", "Hardkernel"),
("Orange Pi 5 Plus", "Xunlong Software"), ("NVIDIA Jetson Nano Developer Kit", "NVIDIA"),
("Banana Pi BPI-R3", "SinoVoip"), ("Rock Pi 4C", "Radxa"),
("NanoPi R4S", "FriendlyElec"), ("Orange Pi 5 Plus", "Xunlong Software"),
("PINE64 RockPro64", "Pine64"), ("Banana Pi BPI-R3", "SinoVoip"),
("BeagleBone Black", "BeagleBoard.org"), ("NanoPi R4S", "FriendlyElec"),
("Unknown Board 1.0", "Unknown Board"), ("PINE64 RockPro64", "Pine64"),
("SomeSingleWordBoard", "SomeSingleWordBoard"), ("BeagleBone Black", "BeagleBoard.org"),
]) ("Unknown Board 1.0", "Unknown Board"),
("SomeSingleWordBoard", "SomeSingleWordBoard"),
],
)
def test_arm_vendor_from_model(model, expected): def test_arm_vendor_from_model(model, expected):
assert _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): 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() facts = driver.get_facts()
assert facts["vendor"] == "Dell Inc." assert facts["vendor"] == "Dell Inc."
assert facts["model"] == "PowerEdge R720" 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): def test_get_facts_vm_kvm(driver):
platform = { 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() facts = driver.get_facts()
assert facts["vendor"] == "KVM" assert facts["vendor"] == "KVM"
assert facts["model"] == "Virtual Machine" 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): def test_get_facts_fallback_vendor_when_dmi_empty(driver):
platform = {"vendor": "", "model": "", "serial": "", "is_vm": False} 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() facts = driver.get_facts()
assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute 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.""" only the fix that lets the login user use Docker stays."""
def test_the_cli_methods_are_gone(self): 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 assert not hasattr(LinuxDriver, name), name
def test_the_engine_access_and_the_permission_fix_stay(self): 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): def test_the_echoed_command_does_not_count_as_the_marker(self, driver):
"""Its literal ``$?`` is no number.""" """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 driver._device = channel
assert _split_status(driver._send(_TRUE_WITH_STATUS))[1] == 0 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): def test_every_status_marker_is_waited_for(self, driver, command):
marker = re.search(r"echo (__[A-Z_]+=)", command).group(1) 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 driver._device = channel
output = driver._send(command) output = driver._send(command)
@@ -1167,7 +1262,11 @@ def _kernel_wire(report: str) -> str:
import base64 import base64
import gzip 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): def test_get_kernel_facts_carries_the_shared_command_across(driver):
@@ -1369,12 +1468,16 @@ class TestAvailableUpdates:
driver._pkg_manager = "dnf" driver._pkg_manager = "dnf"
driver._device.send_command.side_effect = [ driver._device.send_command.side_effect = [
"0", # id -u "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 # DNF_SECURITY_COMMAND pipes through cat: the outer status is cat's, dnf's own
# is in the __DNF_RC marker. # 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()} updates = {u["name"]: u for u in driver.get_available_updates()}
@@ -1402,10 +1505,14 @@ class TestAvailableUpdates:
driver._pkg_manager = "yum" driver._pkg_manager = "yum"
driver._device.send_command.side_effect = [ driver._device.send_command.side_effect = [
"0", # id -u "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()} 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): def test_dnf_that_failed_raises(self, driver):
driver._pkg_manager = "dnf" 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): with pytest.raises(RuntimeError):
driver.get_available_updates() driver.get_available_updates()
@@ -1461,7 +1571,10 @@ class TestRefreshAvailableUpdates:
def test_apt_refreshes_its_index_as_root(self, driver): def test_apt_refreshes_its_index_as_root(self, driver):
driver._pkg_manager = "apt" 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() result = driver.refresh_available_updates()
@@ -1470,7 +1583,10 @@ class TestRefreshAvailableUpdates:
def test_without_a_sudo_password_it_never_waits_for_one(self, driver): def test_without_a_sudo_password_it_never_waits_for_one(self, driver):
driver._pkg_manager = "apt" 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() result = driver.refresh_available_updates()
+4 -2
View File
@@ -27,7 +27,9 @@ OTHER_KEY = paramiko.ECDSAKey.generate()
def _device() -> MagicMock: def _device() -> MagicMock:
device = MagicMock() device = MagicMock()
device.send_command.return_value = "/usr/bin/apt" 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 return device
@@ -76,7 +78,7 @@ def test_a_changed_key_is_refused():
try: try:
raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY) raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
except paramiko.SSHException as exc: # what netmiko does with it 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)) driver = _driver(seen, SshHostKey.of(OTHER_KEY))
with patch("napalm_linux.linux.ConnectHandler", side_effect=refuse): with patch("napalm_linux.linux.ConnectHandler", side_effect=refuse):
+1 -1
View File
@@ -30,7 +30,7 @@ class Channel:
"""An exec channel answering with *output* and *status*; it refuses a line """An exec channel answering with *output* and *status*; it refuses a line
that holds the password, which is what ``ps`` would show.""" that holds the password, which is what ``ps`` would show."""
def __init__(self, host: "Transport"): def __init__(self, host: Transport):
self.host = host self.host = host
self.command = "" self.command = ""
self.stdin = b"" self.stdin = b""