"""Unit tests for LinuxDriver – parsing helpers (no real SSH connection needed).""" import pytest from unittest.mock import MagicMock, patch from napalm_linux.linux import LinuxDriver # --------------------------------------------------------------------------- # Fixture – driver without a real connection # --------------------------------------------------------------------------- @pytest.fixture() def driver(): """Return a LinuxDriver instance with netmiko mocked out.""" d = LinuxDriver.__new__(LinuxDriver) d.hostname = "testhost" d.username = "user" d.password = "pass" # noqa: S106 d.timeout = 60 d.port = 22 d._secret = "pass" # noqa: S105 d._forced_pkg_manager = None d._pkg_manager = "apt" d.netmiko_optional_args = {} d._device = MagicMock() return d def _mock_send(driver_fixture, output: str): """Patch _send to return *output* for any command.""" driver_fixture._device.send_command.return_value = output # --------------------------------------------------------------------------- # _parse_cron_line # --------------------------------------------------------------------------- class TestParseCronLine: def test_regular_user_cron(self): line = "0 4 * * * /usr/local/bin/backup.sh # nightly backup" job = LinuxDriver._parse_cron_line(line, source_user="root", has_user_field=False) assert job is not None assert job["user"] == "root" assert job["schedule"] == "0 4 * * *" assert job["command"] == "/usr/local/bin/backup.sh" assert job.get("description") == "nightly backup" def test_cron_d_with_user_field(self): line = "*/5 * * * * www-data /usr/bin/php /var/www/cron.php" job = LinuxDriver._parse_cron_line(line, source_user="root", has_user_field=True) assert job is not None assert job["user"] == "www-data" assert job["schedule"] == "*/5 * * * *" assert "/usr/bin/php" in job["command"] def test_comment_line_returns_none(self): assert LinuxDriver._parse_cron_line("# this is a comment", "root", False) is None def test_blank_line_returns_none(self): assert LinuxDriver._parse_cron_line(" ", "root", False) is None def test_mailto_returns_none(self): assert LinuxDriver._parse_cron_line("MAILTO=root", "root", False) is None # --------------------------------------------------------------------------- # get_interfaces (parsing) # --------------------------------------------------------------------------- IP_LINK_OUTPUT = """\ 1: lo: 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: 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: 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"]) # --------------------------------------------------------------------------- # _collect_platform_info # --------------------------------------------------------------------------- def _dmi_output( sys_vendor: str, product_name: str, product_version: str, product_serial: str, product_uuid: str, detect_virt: str, ) -> str: return "\n".join([sys_vendor, product_name, product_version, product_serial, product_uuid, detect_virt]) 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 # --------------------------------------------------------------------------- # 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