"""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"])