feat: add get_route_to() via ip route show; family-Feld für IPv4/IPv6

Initial commit mit bestehendem Code inkl. neuem get_route_to(): - Parsed ip -4 route show und ip -6 route show - Protokoll-Map: kernel/dhcp/ra/boot→connected, static→static, ospf→ospf, bgp→bgp - family-Feld aus Netzadresse oder Next-Hop (: = ipv6) - default/default6 → 0.0.0.0/0 / ::/0; Host-Routen ohne Prefix bekommen /32 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-06 15:50:02 +02:00
commit 2712389818
8 changed files with 2500 additions and 0 deletions
@@ -0,0 +1 @@
+"""Tests for napalm-linux driver."""
@@ -0,0 +1,341 @@
+"""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: <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"])