Christian ManivongandClaude Sonnet 4.6
2712389818
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>
342 lines
12 KiB
Python
342 lines
12 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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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"])
|