Christian ManivongandClaude Sonnet 4.6
1bfc8dd245
On ARM boards (Raspberry Pi, ODROID, etc.) /sys/class/dmi/id/ does not
exist. _collect_platform_info() now falls back to:
- /sys/firmware/devicetree/base/model (preferred)
- /proc/cpuinfo Model: / Serial: (fallback)
Three bugs fixed in the process:
1. systemd-detect-virt exits 1 on bare metal, so the old
"|| echo none" pattern produced d="none\nnone" (two lines),
shifting all subsequent fields by one. Fixed with ${d:-none}.
2. _send() calls .strip() on output, silently eating the five leading
blank lines that represent empty DMI fields on ARM. Fixed by
prefixing the printf output with a DMIBEGIN sentinel so the parser
can locate field 0 regardless of leading whitespace.
3. Vendor was always empty for ARM, falling back to the generic "Linux"
constant. Added _ARM_VENDOR_PREFIXES lookup table and
_arm_vendor_from_model() to derive the canonical vendor name from
the model string (e.g. "Raspberry Pi Foundation" for any RPi board).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
624 lines
24 KiB
Python
624 lines
24 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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _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
|