christianmanivong
c8fc46c373
Where docker lives is device-specific; what to do with it is not. QNAP's Container Station installs docker under /share/<pool>/.qpkg/ and never puts it on PATH, so a QTS driver inheriting this class found no docker at all. Rather than reimplementing the Docker surface in the vendor driver, the path becomes a single overridable method and every call site goes through it. Per docs/ARCHITECTURE.md 4.4, generic logic belongs to the shared driver and only the device-specific mechanics belong to the vendor one. A test asserts that *every* docker call site uses the hook — a half converted set would let detection find the binary while the actual queries still missed it, which only shows up against real hardware.
856 lines
33 KiB
Python
856 lines
33 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"
|
||
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)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
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"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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"]):
|
||
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
|
||
|
||
|
||
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"]):
|
||
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"]):
|
||
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}"
|