feat: add get_route_to() via ip route show; family-Feld für IPv4/IPv6
Initial commit mit bestehendem Code inkl. neuem get_route_to(): - Parsed ip -4 route show und ip -6 route show - Protokoll-Map: kernel/dhcp/ra/boot→connected, static→static, ospf→ospf, bgp→bgp - family-Feld aus Netzadresse oder Next-Hop (: = ipv6) - default/default6 → 0.0.0.0/0 / ::/0; Host-Routen ohne Prefix bekommen /32 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
+44
@@ -0,0 +1,44 @@
|
||||
# Python
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*.pyo
|
||||
*.pyd
|
||||
*.so
|
||||
*.egg
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
.eggs/
|
||||
wheels/
|
||||
|
||||
# Virtual environments
|
||||
.venv/
|
||||
venv/
|
||||
env/
|
||||
.env
|
||||
|
||||
# Packaging
|
||||
*.tar.gz
|
||||
*.whl
|
||||
MANIFEST
|
||||
|
||||
# Testing
|
||||
.pytest_cache/
|
||||
.coverage
|
||||
.coverage.*
|
||||
htmlcov/
|
||||
coverage.xml
|
||||
|
||||
# Type checking
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
|
||||
# IDEs
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
*~
|
||||
|
||||
# OS
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
@@ -0,0 +1,28 @@
|
||||
# Changelog
|
||||
|
||||
All notable changes to this project will be documented in this file.
|
||||
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.1.0] – 2026-05-29
|
||||
|
||||
### Added
|
||||
|
||||
- Initial release
|
||||
- `LinuxDriver` inheriting from `napalm_device_types.OSDriver`
|
||||
- SSH transport via Netmiko `linux` device type
|
||||
- Auto-detection of package manager: `apt`, `dnf`, `yum`, `apk`, `pacman`
|
||||
- Standard NAPALM methods: `get_facts()`, `get_interfaces()`, `get_interfaces_ip()`,
|
||||
`get_arp_table()`, `get_config()`, `ping()`
|
||||
- OS-specific methods: `get_packages()`, `get_pending_updates()`, `apply_updates()`,
|
||||
`get_services()`, `get_users()`, `get_processes()`, `get_cron_jobs()`
|
||||
- `apply_updates()` validates all package names against `^[a-zA-Z0-9_\-\+\.]+$`
|
||||
to prevent shell injection before passing them to the package manager
|
||||
- SysV fallback for `get_services()` on non-systemd systems
|
||||
- Unit tests with mocked SSH connections
|
||||
|
||||
[Unreleased]: https://github.com/chrismanivong/napalm-linux/compare/v0.1.0...HEAD
|
||||
[0.1.0]: https://github.com/chrismanivong/napalm-linux/releases/tag/v0.1.0
|
||||
@@ -0,0 +1,169 @@
|
||||
# napalm-linux
|
||||
|
||||
NAPALM driver for **generic Linux systems** — Debian, Ubuntu, RHEL, Rocky, Alpine, Arch and any
|
||||
other distribution reachable via SSH.
|
||||
|
||||
Connects over SSH using [Netmiko](https://github.com/ktbyers/netmiko) (`linux` device type) and
|
||||
**automatically detects** the installed package manager (`apt`, `dnf`, `yum`, `apk`, `pacman`).
|
||||
|
||||
## Requirements
|
||||
|
||||
| Dependency | Minimum version |
|
||||
|---|---|
|
||||
| Python | 3.9 |
|
||||
| NAPALM | 4.0 |
|
||||
| Netmiko | 4.0 |
|
||||
| napalm-device-types | 0.2.0 |
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
pip install napalm napalm-linux
|
||||
```
|
||||
|
||||
Or from source:
|
||||
|
||||
```bash
|
||||
git clone https://github.com/chrismanivong/napalm-linux
|
||||
pip install -e napalm-linux/
|
||||
```
|
||||
|
||||
When working in the NetOrk monorepo, install both packages as editable:
|
||||
|
||||
```bash
|
||||
pip install -e vendor/napalm-device-types/ -e vendor/napalm-linux/
|
||||
```
|
||||
|
||||
## Quick start
|
||||
|
||||
```python
|
||||
from napalm import get_network_driver
|
||||
|
||||
Driver = get_network_driver("linux")
|
||||
|
||||
with Driver(
|
||||
"10.0.0.5",
|
||||
"admin",
|
||||
"s3cr3t",
|
||||
optional_args={
|
||||
# "port": 22,
|
||||
# "pkg_manager": "apt", # force package manager; auto-detected by default
|
||||
# "secret": "sudo-pass", # password for sudo / enable (defaults to login password)
|
||||
# "debugging": True, # enable verbose logging
|
||||
},
|
||||
) as dev:
|
||||
facts = dev.get_facts()
|
||||
print(facts)
|
||||
|
||||
# OS-specific methods (from napalm-device-types OSDriver)
|
||||
packages = dev.get_packages()
|
||||
updates = dev.get_pending_updates()
|
||||
services = dev.get_services()
|
||||
users = dev.get_users()
|
||||
procs = dev.get_processes()
|
||||
jobs = dev.get_cron_jobs()
|
||||
|
||||
# Upgrade specific packages
|
||||
result = dev.apply_updates(["openssh-server", "curl"])
|
||||
print(result) # {"success": True, "output": "..."}
|
||||
|
||||
# Upgrade everything with pending updates
|
||||
result = dev.apply_updates([])
|
||||
```
|
||||
|
||||
## Supported NAPALM methods
|
||||
|
||||
### Standard NAPALM
|
||||
|
||||
| Method | Supported | Notes |
|
||||
|---|---|---|
|
||||
| `open()` / `close()` | ✅ | SSH via netmiko `linux` |
|
||||
| `is_alive()` | ✅ | |
|
||||
| `get_facts()` | ✅ | DMI / `/proc/uptime` / `ip link` |
|
||||
| `get_interfaces()` | ✅ | `ip link show` |
|
||||
| `get_interfaces_ip()` | ✅ | `ip addr show` |
|
||||
| `get_arp_table()` | ✅ | `ip neigh show` |
|
||||
| `get_config()` | ✅ | Returns `ip addr` + `ip route` output |
|
||||
| `ping()` | ✅ | Executes `ping` on the remote host |
|
||||
| `load_merge_candidate()` | ❌ | Not applicable for generic Linux |
|
||||
| `load_replace_candidate()` | ❌ | Not applicable for generic Linux |
|
||||
| `compare_config()` | ❌ | Not applicable for generic Linux |
|
||||
| `commit_config()` | ❌ | Not applicable for generic Linux |
|
||||
| `discard_config()` | ❌ | Not applicable for generic Linux |
|
||||
| `rollback()` | ❌ | Not applicable for generic Linux |
|
||||
|
||||
### OSDriver extensions (napalm-device-types)
|
||||
|
||||
| Method | Supported | Package managers |
|
||||
|---|---|---|
|
||||
| `get_packages()` | ✅ | apt, dnf, yum, apk, pacman |
|
||||
| `get_pending_updates()` | ✅ | apt, dnf, yum, apk, pacman |
|
||||
| `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman |
|
||||
| `get_services()` | ✅ | systemd (fallback: SysV `service`) |
|
||||
| `get_users()` | ✅ | `/etc/passwd` + `/etc/group` |
|
||||
| `get_processes()` | ✅ | `ps axo` |
|
||||
| `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` |
|
||||
|
||||
## Package manager auto-detection
|
||||
|
||||
The driver probes for each binary in order via `command -v`:
|
||||
|
||||
```
|
||||
apt → dnf → yum → apk → pacman
|
||||
```
|
||||
|
||||
Force a specific package manager:
|
||||
|
||||
```python
|
||||
optional_args={"pkg_manager": "dnf"}
|
||||
```
|
||||
|
||||
## SSH user permissions
|
||||
|
||||
The SSH user needs read access to:
|
||||
|
||||
| Data | Required permission |
|
||||
|---|---|
|
||||
| `/etc/passwd`, `/etc/group` | world-readable (default) |
|
||||
| `/proc/uptime`, `/sys/class/dmi/…` | world-readable (default) |
|
||||
| User crontabs (`/var/spool/cron/…`) | `root` or `sudo` required |
|
||||
| `systemctl is-enabled <unit>` | unprivileged on most distros |
|
||||
| `apt list --upgradable` | may require `apt-get update` (root) |
|
||||
| `dnf check-update` / `yum check-update` | unprivileged, but slower without cache |
|
||||
|
||||
For full functionality it is recommended to run as `root` or grant passwordless `sudo` for
|
||||
the above commands.
|
||||
|
||||
## Tested distributions
|
||||
|
||||
| Distribution | Version | Package manager | Tested |
|
||||
|---|---|---|---|
|
||||
| Debian | 12 (Bookworm) | apt | ✅ |
|
||||
| Ubuntu | 22.04 LTS | apt | ✅ |
|
||||
| Rocky Linux | 9 | dnf | planned |
|
||||
| Alpine Linux | 3.19 | apk | planned |
|
||||
| Arch Linux | rolling | pacman | planned |
|
||||
|
||||
Contributions for additional distributions and versions are welcome.
|
||||
|
||||
## Development
|
||||
|
||||
```bash
|
||||
# Create venv
|
||||
python -m venv .venv
|
||||
source .venv/bin/activate
|
||||
|
||||
# Install in editable mode with dev dependencies
|
||||
pip install -e ../napalm-device-types/ -e ".[dev]"
|
||||
|
||||
# Run tests
|
||||
pytest tests/ -v
|
||||
|
||||
# Lint / format
|
||||
ruff check napalm_linux/
|
||||
ruff format napalm_linux/
|
||||
```
|
||||
|
||||
## License
|
||||
|
||||
Apache 2.0
|
||||
@@ -0,0 +1,5 @@
|
||||
"""napalm-linux – NAPALM driver for generic Linux systems."""
|
||||
|
||||
from napalm_linux.linux import LinuxDriver
|
||||
|
||||
__all__ = ["LinuxDriver"]
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,65 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=68", "wheel"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "napalm-linux"
|
||||
version = "0.1.0"
|
||||
description = "NAPALM driver for generic Linux systems via SSH"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.9"
|
||||
license = { text = "Apache-2.0" }
|
||||
authors = [
|
||||
{ name = "Christian Manivong", email = "christian@manivong.de" },
|
||||
]
|
||||
keywords = [
|
||||
"napalm",
|
||||
"network",
|
||||
"automation",
|
||||
"linux",
|
||||
"ssh",
|
||||
"driver",
|
||||
]
|
||||
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",
|
||||
"Topic :: System :: Networking",
|
||||
"Topic :: System :: Systems Administration",
|
||||
"Typing :: Typed",
|
||||
]
|
||||
dependencies = [
|
||||
"napalm>=4.0",
|
||||
"napalm-device-types>=0.3.0",
|
||||
"netmiko>=4.0.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"pytest",
|
||||
"pytest-cov",
|
||||
"black",
|
||||
"ruff",
|
||||
"mypy",
|
||||
]
|
||||
|
||||
[project.entry-points."napalm.drivers"]
|
||||
linux = "napalm_linux:LinuxDriver"
|
||||
|
||||
[project.urls]
|
||||
Repository = "https://github.com/chrismanivong/napalm-linux"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["napalm_linux*"]
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 100
|
||||
target-version = "py39"
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for napalm-linux driver."""
|
||||
@@ -0,0 +1,341 @@
|
||||
"""Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed)."""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from napalm_linux.linux import LinuxDriver
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixture – driver without a real connection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def driver():
|
||||
"""Return a LinuxDriver instance with netmiko mocked out."""
|
||||
d = LinuxDriver.__new__(LinuxDriver)
|
||||
d.hostname = "testhost"
|
||||
d.username = "user"
|
||||
d.password = "pass" # noqa: S106
|
||||
d.timeout = 60
|
||||
d.port = 22
|
||||
d._secret = "pass" # noqa: S105
|
||||
d._forced_pkg_manager = None
|
||||
d._pkg_manager = "apt"
|
||||
d.netmiko_optional_args = {}
|
||||
d._device = MagicMock()
|
||||
return d
|
||||
|
||||
|
||||
def _mock_send(driver_fixture, output: str):
|
||||
"""Patch _send to return *output* for any command."""
|
||||
driver_fixture._device.send_command.return_value = output
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _parse_cron_line
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseCronLine:
|
||||
def test_regular_user_cron(self):
|
||||
line = "0 4 * * * /usr/local/bin/backup.sh # nightly backup"
|
||||
job = LinuxDriver._parse_cron_line(line, source_user="root", has_user_field=False)
|
||||
assert job is not None
|
||||
assert job["user"] == "root"
|
||||
assert job["schedule"] == "0 4 * * *"
|
||||
assert job["command"] == "/usr/local/bin/backup.sh"
|
||||
assert job.get("description") == "nightly backup"
|
||||
|
||||
def test_cron_d_with_user_field(self):
|
||||
line = "*/5 * * * * www-data /usr/bin/php /var/www/cron.php"
|
||||
job = LinuxDriver._parse_cron_line(line, source_user="root", has_user_field=True)
|
||||
assert job is not None
|
||||
assert job["user"] == "www-data"
|
||||
assert job["schedule"] == "*/5 * * * *"
|
||||
assert "/usr/bin/php" in job["command"]
|
||||
|
||||
def test_comment_line_returns_none(self):
|
||||
assert LinuxDriver._parse_cron_line("# this is a comment", "root", False) is None
|
||||
|
||||
def test_blank_line_returns_none(self):
|
||||
assert LinuxDriver._parse_cron_line(" ", "root", False) is None
|
||||
|
||||
def test_mailto_returns_none(self):
|
||||
assert LinuxDriver._parse_cron_line("MAILTO=root", "root", False) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_interfaces (parsing)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
IP_LINK_OUTPUT = """\
|
||||
1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN mode DEFAULT group default qlen 1000\\ link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
|
||||
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP mode DEFAULT group default qlen 1000\\ link/ether aa:bb:cc:dd:ee:ff brd ff:ff:ff:ff:ff:ff
|
||||
3: eth1: <BROADCAST,MULTICAST> mtu 1500 qdisc noop state DOWN mode DEFAULT group default qlen 1000\\ link/ether 11:22:33:44:55:66 brd ff:ff:ff:ff:ff:ff
|
||||
"""
|
||||
|
||||
|
||||
def test_get_interfaces_parses_state(driver):
|
||||
with patch.object(driver, "_send", return_value=IP_LINK_OUTPUT):
|
||||
result = driver.get_interfaces()
|
||||
assert "eth0" in result
|
||||
assert result["eth0"]["is_up"] is True
|
||||
assert result["eth0"]["mtu"] == 1500
|
||||
assert result["eth0"]["mac_address"] == "aa:bb:cc:dd:ee:ff"
|
||||
assert "eth1" in result
|
||||
assert result["eth1"]["is_up"] is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _parse_uptime
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parse_uptime(driver):
|
||||
with patch.object(driver, "_send", return_value="86400.12 1234.56"):
|
||||
assert driver._parse_uptime() == 86400
|
||||
|
||||
|
||||
def test_parse_uptime_invalid(driver):
|
||||
with patch.object(driver, "_send", return_value=""):
|
||||
assert driver._parse_uptime() == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_packages (apt)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
APT_PKG_OUTPUT = (
|
||||
"openssh-server\t1:9.2p1-2+deb12u2\t512\tsecure shell server\n"
|
||||
"curl\t7.88.1-10+deb12u5\t1024\tcommand line tool for transferring data\n"
|
||||
)
|
||||
|
||||
|
||||
def test_get_packages_apt(driver):
|
||||
driver._pkg_manager = "apt"
|
||||
with patch.object(driver, "_send", return_value=APT_PKG_OUTPUT):
|
||||
pkgs = driver.get_packages()
|
||||
assert len(pkgs) == 2
|
||||
assert pkgs[0]["name"] == "openssh-server"
|
||||
assert pkgs[0]["version"] == "1:9.2p1-2+deb12u2"
|
||||
assert pkgs[0]["installed"] is True
|
||||
assert pkgs[0]["source"] == "apt"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_pending_updates (apt)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
APT_UPGRADABLE = (
|
||||
"Listing... Done\n"
|
||||
"openssh-server/stable 1:9.2p1-2+deb12u2 amd64 [upgradable from: 1:9.2p1-2+deb12u1]\n"
|
||||
"curl/stable 7.88.1-10+deb12u6 amd64 [upgradable from: 7.88.1-10+deb12u5]\n"
|
||||
)
|
||||
|
||||
|
||||
def test_get_pending_updates_apt(driver):
|
||||
driver._pkg_manager = "apt"
|
||||
with patch.object(driver, "_send", side_effect=["", APT_UPGRADABLE]):
|
||||
updates = driver.get_pending_updates()
|
||||
assert len(updates) == 2
|
||||
assert updates[0]["name"] == "openssh-server"
|
||||
assert updates[0]["current_version"] == "1:9.2p1-2+deb12u1"
|
||||
assert updates[0]["new_version"] == "1:9.2p1-2+deb12u2"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_users
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
PASSWD_OUT = (
|
||||
"root:x:0:0:root:/root:/bin/bash\n"
|
||||
"admin:x:1000:1000:Admin User:/home/admin:/bin/bash\n"
|
||||
"daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin\n"
|
||||
)
|
||||
|
||||
GROUP_OUT = (
|
||||
"sudo:x:27:admin\n"
|
||||
"docker:x:999:admin\n"
|
||||
"adm:x:4:admin\n"
|
||||
)
|
||||
|
||||
|
||||
def test_get_users(driver):
|
||||
with patch.object(driver, "_send", side_effect=[PASSWD_OUT, GROUP_OUT]):
|
||||
users = driver.get_users()
|
||||
|
||||
admin = next(u for u in users if u["username"] == "admin")
|
||||
assert admin["uid"] == 1000
|
||||
assert admin["gid"] == 1000
|
||||
assert admin["home"] == "/home/admin"
|
||||
assert admin["shell"] == "/bin/bash"
|
||||
assert set(admin["groups"]) == {"sudo", "docker", "adm"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ping parsing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
PING_OUTPUT = """\
|
||||
PING 8.8.8.8 (8.8.8.8) 100(128) bytes of data.
|
||||
108 bytes from 8.8.8.8: icmp_seq=1 ttl=118 time=12.3 ms
|
||||
108 bytes from 8.8.8.8: icmp_seq=2 ttl=118 time=11.9 ms
|
||||
108 bytes from 8.8.8.8: icmp_seq=3 ttl=118 time=12.1 ms
|
||||
|
||||
--- 8.8.8.8 ping statistics ---
|
||||
3 packets transmitted, 3 received, 0% packet loss, time 2003ms
|
||||
rtt min/avg/max/mdev = 11.900/12.100/12.300/0.163 ms
|
||||
"""
|
||||
|
||||
|
||||
def test_ping_parses_output(driver):
|
||||
with patch.object(driver, "_send", return_value=PING_OUTPUT):
|
||||
result = driver.ping("8.8.8.8", count=3)
|
||||
assert "success" in result
|
||||
assert result["success"]["probes_sent"] == 3
|
||||
assert result["success"]["packet_loss"] == 0
|
||||
assert result["success"]["rtt_avg"] == pytest.approx(12.1)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# apply_updates – input validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestApplyUpdatesValidation:
|
||||
def test_valid_package_names_accepted(self, driver):
|
||||
driver._pkg_manager = "apt"
|
||||
apt_output = "Reading package lists... Done\nThe following packages will be upgraded:\n openssh-server\n1 upgraded."
|
||||
with patch.object(driver, "_send", return_value=apt_output):
|
||||
result = driver.apply_updates(["openssh-server", "curl", "lib32-foo+bar.so"])
|
||||
assert result["success"] is True
|
||||
|
||||
def test_invalid_package_name_raises(self, driver):
|
||||
with pytest.raises(ValueError, match="Invalid package name"):
|
||||
driver.apply_updates(["open;ssh"])
|
||||
|
||||
def test_shell_injection_blocked(self, driver):
|
||||
with pytest.raises(ValueError, match="Invalid package name"):
|
||||
driver.apply_updates(["pkg && rm -rf /"])
|
||||
|
||||
def test_space_in_name_blocked(self, driver):
|
||||
with pytest.raises(ValueError, match="Invalid package name"):
|
||||
driver.apply_updates(["my package"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# apply_updates – apt
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
APT_UPGRADE_SUCCESS = (
|
||||
"Reading package lists... Done\n"
|
||||
"Building dependency tree... Done\n"
|
||||
"The following packages will be upgraded:\n"
|
||||
" openssh-server\n"
|
||||
"1 upgraded, 0 newly installed, 0 to remove and 0 not upgraded.\n"
|
||||
)
|
||||
|
||||
APT_UPGRADE_ERROR = (
|
||||
"Reading package lists... Done\n"
|
||||
"E: Unable to lock the administration directory (/var/lib/dpkg/), "
|
||||
"is another process using it?\n"
|
||||
)
|
||||
|
||||
|
||||
def test_apply_updates_apt_success(driver):
|
||||
driver._pkg_manager = "apt"
|
||||
with patch.object(driver, "_send", return_value=APT_UPGRADE_SUCCESS):
|
||||
result = driver.apply_updates(["openssh-server"])
|
||||
assert result["success"] is True
|
||||
assert "openssh-server" in result["output"]
|
||||
assert "error" not in result
|
||||
|
||||
|
||||
def test_apply_updates_apt_error(driver):
|
||||
driver._pkg_manager = "apt"
|
||||
with patch.object(driver, "_send", return_value=APT_UPGRADE_ERROR):
|
||||
result = driver.apply_updates(["openssh-server"])
|
||||
assert result["success"] is False
|
||||
assert "error" in result
|
||||
assert result["error"].startswith("E:")
|
||||
|
||||
|
||||
def test_apply_updates_apt_all_packages(driver):
|
||||
"""Empty list should upgrade everything (no package name args)."""
|
||||
driver._pkg_manager = "apt"
|
||||
sent_commands = []
|
||||
|
||||
def capture_send(cmd):
|
||||
sent_commands.append(cmd)
|
||||
return APT_UPGRADE_SUCCESS
|
||||
|
||||
with patch.object(driver, "_send", side_effect=capture_send):
|
||||
result = driver.apply_updates([])
|
||||
|
||||
assert result["success"] is True
|
||||
# Should use 'apt-get upgrade' without specific package args
|
||||
assert any("upgrade" in cmd and "install" not in cmd for cmd in sent_commands)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# apply_updates – dnf
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
DNF_UPGRADE_SUCCESS = (
|
||||
"Last metadata expiration check: 0:01:23 ago.\n"
|
||||
"Dependencies resolved.\n"
|
||||
"Upgraded:\n openssh-server-9.4p1-1.el9.x86_64\n"
|
||||
"Complete!\n"
|
||||
)
|
||||
|
||||
DNF_UPGRADE_ERROR = (
|
||||
"Last metadata expiration check: 0:01:23 ago.\n"
|
||||
"Error: No match for argument: nonexistent-pkg\n"
|
||||
)
|
||||
|
||||
|
||||
def test_apply_updates_dnf_success(driver):
|
||||
driver._pkg_manager = "dnf"
|
||||
with patch.object(driver, "_send", return_value=DNF_UPGRADE_SUCCESS):
|
||||
result = driver.apply_updates(["openssh-server"])
|
||||
assert result["success"] is True
|
||||
|
||||
|
||||
def test_apply_updates_dnf_error(driver):
|
||||
driver._pkg_manager = "dnf"
|
||||
with patch.object(driver, "_send", return_value=DNF_UPGRADE_ERROR):
|
||||
result = driver.apply_updates(["nonexistent-pkg"])
|
||||
assert result["success"] is False
|
||||
assert "error" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# apply_updates – exception path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_apply_updates_ssh_exception_returns_failure(driver):
|
||||
driver._pkg_manager = "apt"
|
||||
with patch.object(driver, "_send", side_effect=RuntimeError("SSH timeout")):
|
||||
result = driver.apply_updates(["curl"])
|
||||
assert result["success"] is False
|
||||
assert "SSH timeout" in result.get("error", "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# apply_updates – unsupported package manager
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_apply_updates_unsupported_pm_raises(driver):
|
||||
driver._pkg_manager = "zypper"
|
||||
with pytest.raises(NotImplementedError):
|
||||
driver.apply_updates(["curl"])
|
||||
Reference in New Issue
Block a user