Christian ManivongandClaude Opus 5
e8eadb46c6
OSV states Debian ranges in *source* package versions. A source package that ships several binaries gives each its own upstream version, and the two are unrelated numbers: `libldb2` is `2:2.11.0+samba4.22.11+dfsg-0+deb13u1` while its source, samba, is `2:4.22.11+dfsg-…`. A consumer that resolves the coordinate on `source_package` — which is what this driver's `source:Package` is for — and then compares `version` is comparing ldb's version against samba's range. dpkg reads `2.11.0` as older than the `2:4.17.4+dfsg-1` that fixed CVE-2022-44640, so a Debian 13 host running samba 4.22.11, five releases past the fix, was reported vulnerable on four packages at once. This driver cannot fix that comparison. It is the only place that can supply the number to make it with. Empty when dpkg considers it equal to `Version`, and empty on a dpkg that does not know the field, so it falls back to `Version` — which is the previous behaviour and correct everywhere except the shape above. `maxsplit` goes from 4 to 5 with the extra field. Summary stays last, so it keeps whatever it contains. **The fixture was carrying four fields against a format string asking for five.** `source_package` had been silently receiving the description, and no assertion looked at it. It now carries what dpkg-query actually returns, including a package whose source version is a different number from its own. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
915 lines
36 KiB
Python
915 lines
36 KiB
Python
"""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, _arm_vendor_from_model
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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"
|
||
# Set by __init__, which this fixture bypasses via __new__. Without it every
|
||
# call through _sudo() raises AttributeError, which the callers' broad
|
||
# `except Exception` turns into a plain {"success": False} -- so the tests
|
||
# failed for a reason that had nothing to do with what they were testing.
|
||
d._sudo_password = None
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_config
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
IP_ADDR_ROUTE_WITH_VETH = """\
|
||
1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000
|
||
link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
|
||
inet 127.0.0.1/8 scope host lo
|
||
valid_lft forever preferred_lft forever
|
||
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP group default qlen 1000
|
||
link/ether aa:bb:cc:dd:ee:ff brd ff:ff:ff:ff:ff:ff
|
||
inet 192.168.1.10/24 brd 192.168.1.255 scope global dynamic eth0
|
||
valid_lft 3542sec preferred_lft 3542sec
|
||
3512: veth5d7e34d@if2: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue master br-ebb930396f3b state UP group default
|
||
link/ether 02:42:ac:11:00:02 brd ff:ff:ff:ff:ff:ff link-netnsid 0
|
||
default via 192.168.1.1 dev eth0
|
||
192.168.1.0/24 dev eth0 proto kernel scope link src 192.168.1.10
|
||
"""
|
||
|
||
|
||
def test_get_config_strips_dhcp_lease_timers(driver):
|
||
with patch.object(driver, "_send", return_value=IP_ADDR_ROUTE_WITH_VETH):
|
||
result = driver.get_config()
|
||
assert "valid_lft" not in result["running"]
|
||
assert "preferred_lft" not in result["running"]
|
||
|
||
|
||
def test_get_config_strips_veth_interfaces(driver):
|
||
"""Docker creates/destroys veth pairs on every container restart — including
|
||
them would make get_config() report a false-positive change on every poll."""
|
||
with patch.object(driver, "_send", return_value=IP_ADDR_ROUTE_WITH_VETH):
|
||
result = driver.get_config()
|
||
assert "veth5d7e34d" not in result["running"]
|
||
assert "3512:" not in result["running"]
|
||
# Real interfaces and routes must survive the filter
|
||
assert "eth0" in result["running"]
|
||
assert "default via 192.168.1.1" in result["running"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _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)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
#: What `dpkg-query` actually returns for the format this driver asks for.
|
||
#:
|
||
#: The previous fixture carried four fields against a format string asking for
|
||
#: five, so `source_package` was silently receiving the description and no
|
||
#: assertion noticed. A fixture simpler than the data cannot fail the way the
|
||
#: data does.
|
||
APT_PKG_OUTPUT = (
|
||
"openssh-server\t1:9.2p1-2+deb12u2\t512\topenssh\t1:9.2p1-2+deb12u2"
|
||
"\tsecure shell server\n"
|
||
"curl\t7.88.1-10+deb12u5\t1024\tcurl\t7.88.1-10+deb12u5"
|
||
"\tcommand line tool for transferring data\n"
|
||
# The shape that matters: a binary package whose own upstream version has
|
||
# nothing to do with its source package's. ldb 2.11.0 is built from samba
|
||
# 4.22.11, and OSV states Debian ranges in source versions.
|
||
"libldb2\t2:2.11.0+samba4.22.11+dfsg-0+deb13u1\t2048\tsamba"
|
||
"\t2:4.22.11+dfsg-0+deb13u1\tLDB shared library\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) == 3
|
||
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"
|
||
assert pkgs[0]["description"] == "secure shell server"
|
||
|
||
|
||
def test_get_packages_apt_keeps_the_source_package_and_its_version(driver):
|
||
"""OSV states Debian ranges in *source* package versions.
|
||
|
||
A consumer that matches on the source package and then compares the binary
|
||
package's version is comparing two unrelated numbers. On a Debian 13 host
|
||
that reported four Samba libraries as vulnerable to CVE-2022-44640 while
|
||
running samba 4.22.11 — five releases past the fix — because dpkg reads
|
||
ldb's own `2.11.0` as older than samba's `2:4.17.4+dfsg-1`.
|
||
|
||
The driver cannot fix the comparison, but it is the only place that can
|
||
supply the number to compare.
|
||
"""
|
||
driver._pkg_manager = "apt"
|
||
with patch.object(driver, "_send", return_value=APT_PKG_OUTPUT):
|
||
pkgs = {p["name"]: p for p in driver.get_packages()}
|
||
|
||
assert pkgs["libldb2"]["source_package"] == "samba"
|
||
assert pkgs["libldb2"]["source_version"] == "2:4.22.11+dfsg-0+deb13u1"
|
||
assert pkgs["libldb2"]["version"] == "2:2.11.0+samba4.22.11+dfsg-0+deb13u1"
|
||
assert pkgs["libldb2"]["description"] == "LDB shared library"
|
||
|
||
|
||
def test_get_packages_apt_falls_back_when_dpkg_gives_no_source(driver):
|
||
"""`source:Package` is empty for a package whose source name equals its own.
|
||
Older dpkg builds leave `source:Version` empty in that case too."""
|
||
driver._pkg_manager = "apt"
|
||
with patch.object(driver, "_send", return_value="curl\t7.88.1-10\t1024\t\t\ttool\n"):
|
||
(pkg,) = driver.get_packages()
|
||
|
||
assert pkg["source_package"] == "curl"
|
||
assert pkg["source_version"] == "7.88.1-10"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_available_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_available_updates_apt(driver):
|
||
driver._pkg_manager = "apt"
|
||
with patch.object(driver, "_send", return_value=APT_UPGRADABLE):
|
||
updates = driver.get_available_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, **kwargs):
|
||
# _sudo() passes read_timeout as a keyword; without **kwargs this raises
|
||
# TypeError, which the caller's `except Exception` reports as a failed
|
||
# upgrade rather than a broken test double.
|
||
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"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# TYPE_LABEL
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_type_label_is_linux():
|
||
assert LinuxDriver.TYPE_LABEL == "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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _collect_platform_info
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _dmi_output(
|
||
sys_vendor: str,
|
||
product_name: str,
|
||
product_version: str,
|
||
product_serial: str,
|
||
product_uuid: str,
|
||
detect_virt: str,
|
||
dt_model: str = "",
|
||
cpuinfo_serial: str = "",
|
||
cpuinfo_model: str = "",
|
||
) -> str:
|
||
# Mirrors the real shell output: DMIBEGIN sentinel followed by 9 fields.
|
||
# The sentinel prevents _send()'s .strip() from eating leading blank lines
|
||
# (which occur on ARM where all DMI files are absent).
|
||
return "\n".join([
|
||
"DMIBEGIN",
|
||
sys_vendor, product_name, product_version, product_serial,
|
||
product_uuid, detect_virt, dt_model, cpuinfo_serial, cpuinfo_model,
|
||
])
|
||
|
||
|
||
class TestCollectPlatformInfo:
|
||
def test_baremetal_dell(self, driver):
|
||
raw = _dmi_output(
|
||
"Dell Inc.", "PowerEdge R720", "Not Specified", "ABC123",
|
||
"8a2e3f00-dead-beef-0000-123456789abc", "none",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "Dell Inc."
|
||
assert info["model"] == "PowerEdge R720"
|
||
assert info["serial"] == "ABC123"
|
||
assert info["is_vm"] is False
|
||
|
||
def test_baremetal_lenovo_product_version_preferred(self, driver):
|
||
raw = _dmi_output(
|
||
"LENOVO", "10M8000VUS", "ThinkCentre M910x", "MP1234",
|
||
"8a2e3f00-dead-beef-0000-123456789abc", "none",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "LENOVO"
|
||
assert info["model"] == "ThinkCentre M910x"
|
||
assert info["serial"] == "MP1234"
|
||
assert info["is_vm"] is False
|
||
|
||
def test_vm_kvm(self, driver):
|
||
raw = _dmi_output(
|
||
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "pc-i440fx-9.1", "",
|
||
"4c4c4544-0000-2010-8020-b4c04f534a31", "kvm",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "KVM"
|
||
assert info["model"] == "Virtual Machine"
|
||
assert info["serial"] == "4c4c4544-0000-2010-8020-b4c04f534a31"
|
||
assert info["is_vm"] is True
|
||
|
||
def test_vm_vmware(self, driver):
|
||
raw = _dmi_output(
|
||
"VMware, Inc.", "VMware Virtual Platform", "None", "VMware-42 12 34 56",
|
||
"4244560c-dead-beef-0000-abcdef123456", "vmware",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "VMware ESXi"
|
||
assert info["model"] == "Virtual Machine"
|
||
assert info["serial"] == "VMware-42 12 34 56"
|
||
assert info["is_vm"] is True
|
||
|
||
def test_vm_hyperv(self, driver):
|
||
raw = _dmi_output(
|
||
"Microsoft Corporation", "Virtual Machine", "Hyper-V UEFI Release v4.1", "",
|
||
"7C5B4B1F-1234-5678-ABCD-000000000001", "microsoft",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "Microsoft Hyper-V"
|
||
assert info["model"] == "Virtual Machine"
|
||
assert info["serial"] == "7C5B4B1F-1234-5678-ABCD-000000000001"
|
||
assert info["is_vm"] is True
|
||
|
||
def test_junk_dmi_values_filtered(self, driver):
|
||
raw = _dmi_output(
|
||
"To Be Filled By O.E.M.", "To Be Filled By O.E.M.", "Not Specified",
|
||
"To Be Filled By O.E.M.", "", "none",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == ""
|
||
assert info["model"] == ""
|
||
assert info["is_vm"] is False
|
||
|
||
def test_vm_kvm_fallback_via_dmi_when_detect_virt_unavailable(self, driver):
|
||
# systemd-detect-virt returns "none" (not installed), sys_vendor reveals QEMU
|
||
raw = _dmi_output(
|
||
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "", "",
|
||
"4c4c4544-0000-2010-8020-b4c04f534a31", "none",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["is_vm"] is True
|
||
assert info["vendor"] == "KVM"
|
||
assert info["model"] == "Virtual Machine"
|
||
|
||
def test_container_docker(self, driver):
|
||
raw = _dmi_output(
|
||
"QEMU", "Standard PC (i440FX + PIIX, 1996)", "", "",
|
||
"4c4c4544-0000-2010-8020-b4c04f534a31", "docker",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "Docker"
|
||
assert info["model"] == "Container"
|
||
assert info["is_vm"] is True
|
||
|
||
def test_container_lxc(self, driver):
|
||
raw = _dmi_output("", "", "", "", "", "lxc")
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "LXC"
|
||
assert info["model"] == "Container"
|
||
assert info["is_vm"] is True
|
||
|
||
def test_ssh_failure_returns_safe_defaults(self, driver):
|
||
with patch.object(driver, "_send", side_effect=Exception("SSH error")):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == ""
|
||
assert info["model"] == ""
|
||
assert info["is_vm"] is False
|
||
|
||
def test_arm_device_tree_fallback(self, driver):
|
||
raw = _dmi_output(
|
||
"", "", "", "", "", "none",
|
||
dt_model="Raspberry Pi 4 Model B Rev 1.4",
|
||
cpuinfo_serial="100000002a6d96dc",
|
||
cpuinfo_model="Raspberry Pi 4 Model B Rev 1.4",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "Raspberry Pi Foundation"
|
||
assert info["model"] == "Raspberry Pi 4 Model B Rev 1.4"
|
||
assert info["serial"] == "100000002a6d96dc"
|
||
assert info["is_vm"] is False
|
||
|
||
def test_arm_cpuinfo_model_when_no_dt(self, driver):
|
||
raw = _dmi_output(
|
||
"", "", "", "", "", "none",
|
||
dt_model="",
|
||
cpuinfo_serial="0000000012345678",
|
||
cpuinfo_model="Raspberry Pi 3 Model B Plus Rev 1.3",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "Raspberry Pi Foundation"
|
||
assert info["model"] == "Raspberry Pi 3 Model B Plus Rev 1.3"
|
||
assert info["serial"] == "0000000012345678"
|
||
assert info["is_vm"] is False
|
||
|
||
def test_arm_no_fallback_data_returns_empty(self, driver):
|
||
raw = _dmi_output("", "", "", "", "", "none")
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == ""
|
||
assert info["model"] == ""
|
||
assert info["is_vm"] is False
|
||
|
||
def test_bare_metal_detect_virt_exit1_does_not_shift_arm_fields(self, driver):
|
||
# systemd-detect-virt exits 1 on bare metal, old "|| echo none" caused
|
||
# d="none\nnone" which shifted subsequent lines off by one.
|
||
# With the ${d:-none} fix this no longer happens.
|
||
raw = _dmi_output(
|
||
"", "", "", "", "", "none",
|
||
dt_model="",
|
||
cpuinfo_serial="0000000012345678",
|
||
cpuinfo_model="Raspberry Pi 3 Model B Rev 1.2",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "Raspberry Pi Foundation"
|
||
assert info["model"] == "Raspberry Pi 3 Model B Rev 1.2"
|
||
assert info["serial"] == "0000000012345678"
|
||
|
||
def test_arm_fallback_ignored_when_dmi_present(self, driver):
|
||
raw = _dmi_output(
|
||
"Dell Inc.", "PowerEdge R720", "Not Specified", "XYZ999",
|
||
"8a2e3f00-dead-beef-0000-123456789abc", "none",
|
||
dt_model="some-dt-model",
|
||
cpuinfo_serial="deadbeef",
|
||
cpuinfo_model="some cpuinfo model",
|
||
)
|
||
with patch.object(driver, "_send", return_value=raw):
|
||
info = driver._collect_platform_info()
|
||
assert info["vendor"] == "Dell Inc."
|
||
assert info["model"] == "PowerEdge R720"
|
||
assert info["serial"] == "XYZ999"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _arm_vendor_from_model
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize("model,expected", [
|
||
("Raspberry Pi 4 Model B Rev 1.4", "Raspberry Pi Foundation"),
|
||
("Raspberry Pi 3 Model B Rev 1.2", "Raspberry Pi Foundation"),
|
||
("ODROID-N2L", "Hardkernel"),
|
||
("NVIDIA Jetson Nano Developer Kit", "NVIDIA"),
|
||
("Rock Pi 4C", "Radxa"),
|
||
("Orange Pi 5 Plus", "Xunlong Software"),
|
||
("Banana Pi BPI-R3", "SinoVoip"),
|
||
("NanoPi R4S", "FriendlyElec"),
|
||
("PINE64 RockPro64", "Pine64"),
|
||
("BeagleBone Black", "BeagleBoard.org"),
|
||
("Unknown Board 1.0", "Unknown Board"),
|
||
("SomeSingleWordBoard", "SomeSingleWordBoard"),
|
||
])
|
||
def test_arm_vendor_from_model(model, expected):
|
||
assert _arm_vendor_from_model(model) == expected
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_facts uses _collect_platform_info
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_get_facts_baremetal_vendor_model_serial(driver):
|
||
platform = {"vendor": "Dell Inc.", "model": "PowerEdge R720", "serial": "ABC123", "is_vm": False}
|
||
with patch.object(driver, "_collect_platform_info", return_value=platform), \
|
||
patch.object(driver, "_parse_uptime", return_value=86400), \
|
||
patch.object(driver, "_send", side_effect=["myhost", "myhost.example.com", "Debian GNU/Linux 12", "eth0\neth1",
|
||
"6.1.0-18-amd64"]):
|
||
facts = driver.get_facts()
|
||
assert facts["vendor"] == "Dell Inc."
|
||
assert facts["model"] == "PowerEdge R720"
|
||
assert facts["serial_number"] == "ABC123"
|
||
assert facts["hostname"] == "myhost"
|
||
assert facts["uptime"] == 86400
|
||
# Booted kernel, not the newest installed one — kernel CVE relevance needs it.
|
||
assert facts["running_kernel"] == "6.1.0-18-amd64"
|
||
|
||
|
||
def test_get_facts_vm_kvm(driver):
|
||
platform = {
|
||
"vendor": "KVM", "model": "Virtual Machine",
|
||
"serial": "4c4c4544-0000-2010-8020-b4c04f534a31", "is_vm": True,
|
||
}
|
||
with patch.object(driver, "_collect_platform_info", return_value=platform), \
|
||
patch.object(driver, "_parse_uptime", return_value=3600), \
|
||
patch.object(driver, "_send", side_effect=["vmhost", "vmhost.local", "Ubuntu 22.04 LTS", "eth0",
|
||
"5.15.0-91-generic"]):
|
||
facts = driver.get_facts()
|
||
assert facts["vendor"] == "KVM"
|
||
assert facts["model"] == "Virtual Machine"
|
||
assert facts["serial_number"] == "4c4c4544-0000-2010-8020-b4c04f534a31"
|
||
|
||
|
||
def test_get_facts_fallback_vendor_when_dmi_empty(driver):
|
||
platform = {"vendor": "", "model": "", "serial": "", "is_vm": False}
|
||
with patch.object(driver, "_collect_platform_info", return_value=platform), \
|
||
patch.object(driver, "_parse_uptime", return_value=0), \
|
||
patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0", "6.6.7-0-lts"]):
|
||
facts = driver.get_facts()
|
||
assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _action_fix_snmp / _action_apt_update_upgrade
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
"""Canned responses covering every _send() call _action_fix_snmp makes."""
|
||
if command.startswith("sudo -n true"):
|
||
return "" # passwordless sudo works
|
||
if command.startswith("command -v apt"):
|
||
return "/usr/bin/apt"
|
||
if command.startswith("command -v"):
|
||
return "" # ufw/iptables not found, other pkg managers not found
|
||
if command.startswith("ss -tnp"):
|
||
return "" # no netork_ip detected — skips firewall step
|
||
if command.startswith("cat /etc/snmp/snmpd.conf"):
|
||
return "agentAddress udp:161\nrocommunity public\n"
|
||
if command.startswith("snmpget"):
|
||
return "STRING: Linux test"
|
||
return ""
|
||
|
||
|
||
class TestActionFixSnmp:
|
||
def test_runs_apt_get_update_before_install(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
sudo_calls: list[str] = []
|
||
|
||
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
sudo_calls.append(command)
|
||
return ""
|
||
|
||
with (
|
||
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
|
||
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
|
||
):
|
||
driver._action_fix_snmp()
|
||
|
||
update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c)
|
||
install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c)
|
||
assert update_idx < install_idx
|
||
|
||
def test_install_exception_returns_failure_instead_of_raising(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
|
||
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
if "apt-get install" in command:
|
||
raise TimeoutError("connection timed out")
|
||
return ""
|
||
|
||
with (
|
||
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
|
||
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
|
||
):
|
||
result = driver._action_fix_snmp()
|
||
|
||
assert result["success"] is False
|
||
assert "install failed" in result["output"]
|
||
|
||
def test_apt_get_update_failure_is_non_fatal(self, driver):
|
||
"""apt-get update failing (e.g. transient network issue) must not
|
||
abort the whole action — install is still attempted."""
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
|
||
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
if "apt-get update" in command:
|
||
raise TimeoutError("network unreachable")
|
||
return ""
|
||
|
||
with (
|
||
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
|
||
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
|
||
):
|
||
result = driver._action_fix_snmp()
|
||
|
||
assert "apt-get update failed" in result["output"]
|
||
|
||
|
||
class TestActionAptUpdateUpgrade:
|
||
def test_skips_when_not_apt(self, driver):
|
||
driver._pkg_manager = "dnf"
|
||
result = driver._action_apt_update_upgrade()
|
||
assert result["success"] is True
|
||
assert "Skipped" in result["output"]
|
||
|
||
def test_fails_when_sudo_needs_password_and_none_configured(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"):
|
||
result = driver._action_apt_update_upgrade()
|
||
assert result["success"] is False
|
||
assert "sudo requires a password" in result["output"]
|
||
|
||
def test_runs_update_then_upgrade_successfully(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = "secret" # noqa: S105
|
||
with (
|
||
patch.object(driver, "_send", return_value=""),
|
||
patch.object(
|
||
driver,
|
||
"_sudo",
|
||
side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"],
|
||
) as mock_sudo,
|
||
):
|
||
result = driver._action_apt_update_upgrade()
|
||
|
||
assert result["success"] is True
|
||
assert "[update]" in result["output"]
|
||
assert "[upgrade]" in result["output"]
|
||
update_call, upgrade_call = mock_sudo.call_args_list
|
||
assert "apt-get update" in update_call.args[0]
|
||
# full-upgrade (not plain upgrade) — plain upgrade silently holds back
|
||
# packages whose newer version needs to install/remove dependencies.
|
||
assert "apt-get full-upgrade" in upgrade_call.args[0]
|
||
|
||
def test_exception_during_upgrade_returns_failure(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = "secret" # noqa: S105
|
||
with (
|
||
patch.object(driver, "_send", return_value=""),
|
||
patch.object(
|
||
driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")]
|
||
),
|
||
):
|
||
result = driver._action_apt_update_upgrade()
|
||
|
||
assert result["success"] is False
|
||
assert "[error]" in result["output"]
|
||
|
||
|
||
class TestRunDeviceActionDispatch:
|
||
def test_apt_update_upgrade_action_dispatches(self, driver):
|
||
with patch.object(
|
||
driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""}
|
||
) as mock_action:
|
||
driver.run_device_action("apt_update_upgrade")
|
||
mock_action.assert_called_once()
|
||
|
||
|
||
class TestDockerBinHook:
|
||
"""Where the docker binary lives is device-specific; what to do with it is not.
|
||
|
||
QTS puts Container Station's docker under /share/<pool>/.qpkg/ and not on
|
||
PATH. Rather than duplicating the whole Docker surface in the QNAP driver,
|
||
the path is a one-method hook here and the logic stays generic. See
|
||
docs/ARCHITECTURE.md §4.4, "Generisch vs. treiberspezifisch".
|
||
"""
|
||
|
||
def test_defaults_to_docker_on_path(self, driver):
|
||
assert driver._docker_bin() == "docker"
|
||
|
||
def test_detection_uses_the_hook(self, driver):
|
||
"""A subclass pointing elsewhere must not be probed for a PATH docker."""
|
||
sent = []
|
||
|
||
def _record(cmd, **kwargs):
|
||
sent.append(cmd)
|
||
return ""
|
||
|
||
with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
|
||
with patch.object(driver, "_send", side_effect=_record):
|
||
result = driver.get_docker_info()
|
||
|
||
assert result == {"available": False}
|
||
assert any("/opt/cs/docker" in cmd for cmd in sent)
|
||
assert not any("command -v docker " in cmd for cmd in sent)
|
||
|
||
def test_all_docker_subcommands_use_the_hook(self, driver):
|
||
"""Half-converted call sites are the failure mode here: detection would
|
||
find the binary and the actual queries would still miss it."""
|
||
sent = []
|
||
|
||
def _record(cmd, **kwargs):
|
||
sent.append(cmd)
|
||
if "command -v" in cmd:
|
||
return "/opt/cs/docker"
|
||
if "---CONTAINERS---" in cmd:
|
||
return "---CONTAINERS---\n---IMAGES---\n---VOLUMES---\n---NETWORKS---\n"
|
||
return ""
|
||
|
||
with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
|
||
with patch.object(driver, "_send", side_effect=_record):
|
||
driver.get_docker_info()
|
||
|
||
docker_cmds = [c for c in sent if "docker" in c]
|
||
assert docker_cmds
|
||
for cmd in docker_cmds:
|
||
assert "/opt/cs/docker" in cmd, f"unconverted call site: {cmd}"
|