Christian Manivong
84717ff53b
napalm-device-types' SystemdServicesMixin (2.2.0) now provides get_services() and manage_service(); this driver supplies only the transport (#7): - _run_service_command(): unprivileged reads and root logins run as they are -- the user is asked once per session with "id -u", so a root login on a box without sudo is not prefixed with one. With a sudo password the command goes through _sudo(); without one through "sudo -n", which fails at once instead of hanging the session on a password prompt until the read timeout. - get_services(): one round trip instead of an is-enabled and a show per unit (6.0 s -> 0.8 s on a 184-unit Ubuntu host). A host without systemd still falls back to "service --status-all". - manage_service(): when sudo wants a password netOrk does not have, the failure says how to fix it -- the same hint the apt and SNMP actions give, now one constant (_SUDO_PASSWORD_HINT) instead of two copies. OpenMediaVault and QNAP inherit this driver. OMV gets service control with it; QNAP opts out (napalm-qnap-qts), since QTS has no systemd. README: the sudo option is sudo_password, not secret; manage_service and the systemctl permissions are listed. Closes #7
1285 lines
51 KiB
Python
1285 lines
51 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"
|
||
# Set by __init__, which this fixture bypasses via __new__. Without it every
|
||
# call through _sudo() raises AttributeError, which the callers' broad
|
||
# `except Exception` turns into a plain {"success": False} -- so the tests
|
||
# failed for a reason that had nothing to do with what they were testing.
|
||
d._sudo_password = None
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_config
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
IP_ADDR_ROUTE_WITH_VETH = """\
|
||
1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000
|
||
link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
|
||
inet 127.0.0.1/8 scope host lo
|
||
valid_lft forever preferred_lft forever
|
||
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc mq state UP group default qlen 1000
|
||
link/ether aa:bb:cc:dd:ee:ff brd ff:ff:ff:ff:ff:ff
|
||
inet 192.168.1.10/24 brd 192.168.1.255 scope global dynamic eth0
|
||
valid_lft 3542sec preferred_lft 3542sec
|
||
3512: veth5d7e34d@if2: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue master br-ebb930396f3b state UP group default
|
||
link/ether 02:42:ac:11:00:02 brd ff:ff:ff:ff:ff:ff link-netnsid 0
|
||
default via 192.168.1.1 dev eth0
|
||
192.168.1.0/24 dev eth0 proto kernel scope link src 192.168.1.10
|
||
"""
|
||
|
||
|
||
def test_get_config_strips_dhcp_lease_timers(driver):
|
||
with patch.object(driver, "_send", return_value=IP_ADDR_ROUTE_WITH_VETH):
|
||
result = driver.get_config()
|
||
assert "valid_lft" not in result["running"]
|
||
assert "preferred_lft" not in result["running"]
|
||
|
||
|
||
def test_get_config_strips_veth_interfaces(driver):
|
||
"""Docker creates/destroys veth pairs on every container restart — including
|
||
them would make get_config() report a false-positive change on every poll."""
|
||
with patch.object(driver, "_send", return_value=IP_ADDR_ROUTE_WITH_VETH):
|
||
result = driver.get_config()
|
||
assert "veth5d7e34d" not in result["running"]
|
||
assert "3512:" not in result["running"]
|
||
# Real interfaces and routes must survive the filter
|
||
assert "eth0" in result["running"]
|
||
assert "default via 192.168.1.1" in result["running"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _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)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
#: What `dpkg-query` actually returns for the format this driver asks for.
|
||
#:
|
||
#: The previous fixture carried four fields against a format string asking for
|
||
#: five, so `source_package` was silently receiving the description and no
|
||
#: assertion noticed. A fixture simpler than the data cannot fail the way the
|
||
#: data does.
|
||
APT_PKG_OUTPUT = (
|
||
"openssh-server\t1:9.2p1-2+deb12u2\t512\topenssh\t1:9.2p1-2+deb12u2"
|
||
"\tsecure shell server\n"
|
||
"curl\t7.88.1-10+deb12u5\t1024\tcurl\t7.88.1-10+deb12u5"
|
||
"\tcommand line tool for transferring data\n"
|
||
# The shape that matters: a binary package whose own upstream version has
|
||
# nothing to do with its source package's. ldb 2.11.0 is built from samba
|
||
# 4.22.11, and OSV states Debian ranges in source versions.
|
||
"libldb2\t2:2.11.0+samba4.22.11+dfsg-0+deb13u1\t2048\tsamba"
|
||
"\t2:4.22.11+dfsg-0+deb13u1\tLDB shared library\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) == 3
|
||
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"
|
||
assert pkgs[0]["description"] == "secure shell server"
|
||
|
||
|
||
def test_get_packages_apt_keeps_the_source_package_and_its_version(driver):
|
||
"""OSV states Debian ranges in *source* package versions.
|
||
|
||
A consumer that matches on the source package and then compares the binary
|
||
package's version is comparing two unrelated numbers. On a Debian 13 host
|
||
that reported four Samba libraries as vulnerable to CVE-2022-44640 while
|
||
running samba 4.22.11 — five releases past the fix — because dpkg reads
|
||
ldb's own `2.11.0` as older than samba's `2:4.17.4+dfsg-1`.
|
||
|
||
The driver cannot fix the comparison, but it is the only place that can
|
||
supply the number to compare.
|
||
"""
|
||
driver._pkg_manager = "apt"
|
||
with patch.object(driver, "_send", return_value=APT_PKG_OUTPUT):
|
||
pkgs = {p["name"]: p for p in driver.get_packages()}
|
||
|
||
assert pkgs["libldb2"]["source_package"] == "samba"
|
||
assert pkgs["libldb2"]["source_version"] == "2:4.22.11+dfsg-0+deb13u1"
|
||
assert pkgs["libldb2"]["version"] == "2:2.11.0+samba4.22.11+dfsg-0+deb13u1"
|
||
assert pkgs["libldb2"]["description"] == "LDB shared library"
|
||
|
||
|
||
def test_get_packages_apt_falls_back_when_dpkg_gives_no_source(driver):
|
||
"""`source:Package` is empty for a package whose source name equals its own.
|
||
Older dpkg builds leave `source:Version` empty in that case too."""
|
||
driver._pkg_manager = "apt"
|
||
with patch.object(driver, "_send", return_value="curl\t7.88.1-10\t1024\t\t\ttool\n"):
|
||
(pkg,) = driver.get_packages()
|
||
|
||
assert pkg["source_package"] == "curl"
|
||
assert pkg["source_version"] == "7.88.1-10"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_available_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_available_updates_apt(driver):
|
||
driver._pkg_manager = "apt"
|
||
with patch.object(driver, "_send", return_value=APT_UPGRADABLE):
|
||
updates = driver.get_available_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, **kwargs):
|
||
# _sudo() passes read_timeout as a keyword; without **kwargs this raises
|
||
# TypeError, which the caller's `except Exception` reports as a failed
|
||
# upgrade rather than a broken test double.
|
||
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",
|
||
"6.1.0-18-amd64"]):
|
||
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
|
||
# Booted kernel, not the newest installed one — kernel CVE relevance needs it.
|
||
assert facts["running_kernel"] == "6.1.0-18-amd64"
|
||
|
||
|
||
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",
|
||
"5.15.0-91-generic"]):
|
||
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", "6.6.7-0-lts"]):
|
||
facts = driver.get_facts()
|
||
assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _action_fix_snmp / _action_apt_update_upgrade
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
"""Canned responses covering every _send() call _action_fix_snmp makes."""
|
||
if command.startswith("sudo -n true"):
|
||
return "" # passwordless sudo works
|
||
if command.startswith("command -v apt"):
|
||
return "/usr/bin/apt"
|
||
if command.startswith("command -v"):
|
||
return "" # ufw/iptables not found, other pkg managers not found
|
||
if command.startswith("ss -tnp"):
|
||
return "" # no netork_ip detected — skips firewall step
|
||
if command.startswith("cat /etc/snmp/snmpd.conf"):
|
||
return "agentAddress udp:161\nrocommunity public\n"
|
||
if command.startswith("snmpget"):
|
||
return "STRING: Linux test"
|
||
return ""
|
||
|
||
|
||
class TestActionFixSnmp:
|
||
def test_runs_apt_get_update_before_install(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
sudo_calls: list[str] = []
|
||
|
||
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
sudo_calls.append(command)
|
||
return ""
|
||
|
||
with (
|
||
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
|
||
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
|
||
):
|
||
driver._action_fix_snmp()
|
||
|
||
update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c)
|
||
install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c)
|
||
assert update_idx < install_idx
|
||
|
||
def test_install_exception_returns_failure_instead_of_raising(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
|
||
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
if "apt-get install" in command:
|
||
raise TimeoutError("connection timed out")
|
||
return ""
|
||
|
||
with (
|
||
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
|
||
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
|
||
):
|
||
result = driver._action_fix_snmp()
|
||
|
||
assert result["success"] is False
|
||
assert "install failed" in result["output"]
|
||
|
||
def test_apt_get_update_failure_is_non_fatal(self, driver):
|
||
"""apt-get update failing (e.g. transient network issue) must not
|
||
abort the whole action — install is still attempted."""
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
|
||
def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
|
||
if "apt-get update" in command:
|
||
raise TimeoutError("network unreachable")
|
||
return ""
|
||
|
||
with (
|
||
patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
|
||
patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
|
||
):
|
||
result = driver._action_fix_snmp()
|
||
|
||
assert "apt-get update failed" in result["output"]
|
||
|
||
|
||
class TestActionAptUpdateUpgrade:
|
||
def test_skips_when_not_apt(self, driver):
|
||
driver._pkg_manager = "dnf"
|
||
result = driver._action_apt_update_upgrade()
|
||
assert result["success"] is True
|
||
assert "Skipped" in result["output"]
|
||
|
||
def test_fails_when_sudo_needs_password_and_none_configured(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = None
|
||
with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"):
|
||
result = driver._action_apt_update_upgrade()
|
||
assert result["success"] is False
|
||
assert "sudo requires a password" in result["output"]
|
||
|
||
def test_runs_update_then_upgrade_successfully(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = "secret" # noqa: S105
|
||
with (
|
||
patch.object(driver, "_send", return_value=""),
|
||
patch.object(
|
||
driver,
|
||
"_sudo",
|
||
side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"],
|
||
) as mock_sudo,
|
||
):
|
||
result = driver._action_apt_update_upgrade()
|
||
|
||
assert result["success"] is True
|
||
assert "[update]" in result["output"]
|
||
assert "[upgrade]" in result["output"]
|
||
update_call, upgrade_call = mock_sudo.call_args_list
|
||
assert "apt-get update" in update_call.args[0]
|
||
# full-upgrade (not plain upgrade) — plain upgrade silently holds back
|
||
# packages whose newer version needs to install/remove dependencies.
|
||
assert "apt-get full-upgrade" in upgrade_call.args[0]
|
||
|
||
def test_exception_during_upgrade_returns_failure(self, driver):
|
||
driver._pkg_manager = "apt"
|
||
driver._sudo_password = "secret" # noqa: S105
|
||
with (
|
||
patch.object(driver, "_send", return_value=""),
|
||
patch.object(
|
||
driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")]
|
||
),
|
||
):
|
||
result = driver._action_apt_update_upgrade()
|
||
|
||
assert result["success"] is False
|
||
assert "[error]" in result["output"]
|
||
|
||
|
||
class TestRunDeviceActionDispatch:
|
||
def test_apt_update_upgrade_action_dispatches(self, driver):
|
||
with patch.object(
|
||
driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""}
|
||
) as mock_action:
|
||
driver.run_device_action("apt_update_upgrade")
|
||
mock_action.assert_called_once()
|
||
|
||
|
||
class TestDockerBinHook:
|
||
"""Where the docker binary lives is device-specific; what to do with it is not.
|
||
|
||
QTS puts Container Station's docker under /share/<pool>/.qpkg/ and not on
|
||
PATH. Rather than duplicating the whole Docker surface in the QNAP driver,
|
||
the path is a one-method hook here and the logic stays generic. See
|
||
docs/ARCHITECTURE.md §4.4, "Generisch vs. treiberspezifisch".
|
||
"""
|
||
|
||
def test_defaults_to_docker_on_path(self, driver):
|
||
assert driver._docker_bin() == "docker"
|
||
|
||
def test_detection_uses_the_hook(self, driver):
|
||
"""A subclass pointing elsewhere must not be probed for a PATH docker."""
|
||
sent = []
|
||
|
||
def _record(cmd, **kwargs):
|
||
sent.append(cmd)
|
||
return ""
|
||
|
||
with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
|
||
with patch.object(driver, "_send", side_effect=_record):
|
||
result = driver.get_docker_info()
|
||
|
||
assert result == {"available": False}
|
||
assert any("/opt/cs/docker" in cmd for cmd in sent)
|
||
assert not any("command -v docker " in cmd for cmd in sent)
|
||
|
||
def test_all_docker_subcommands_use_the_hook(self, driver):
|
||
"""Half-converted call sites are the failure mode here: detection would
|
||
find the binary and the actual queries would still miss it."""
|
||
sent = []
|
||
|
||
def _record(cmd, **kwargs):
|
||
sent.append(cmd)
|
||
if "command -v" in cmd:
|
||
return "/opt/cs/docker"
|
||
if "---CONTAINERS---" in cmd:
|
||
return "---CONTAINERS---\n---IMAGES---\n---VOLUMES---\n---NETWORKS---\n"
|
||
return ""
|
||
|
||
with patch.object(driver, "_docker_bin", return_value="/opt/cs/docker"):
|
||
with patch.object(driver, "_send", side_effect=_record):
|
||
driver.get_docker_info()
|
||
|
||
docker_cmds = [c for c in sent if "docker" in c]
|
||
assert docker_cmds
|
||
for cmd in docker_cmds:
|
||
assert "/opt/cs/docker" in cmd, f"unconverted call site: {cmd}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# uninstall_package – purge, and getting out of `install ok unpacked`
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestUninstallPackage:
|
||
"""Removing a package that does not want to go.
|
||
|
||
Both cases here were found during a fleet-wide Wazuh rollback. Of thirteen
|
||
hosts carrying the agent, seven sat at `install ok unpacked` with the unit
|
||
failed — an upgrade whose postinst could not reach a manager that no longer
|
||
existed. `apt-get remove` cannot help there: apt configures a package before
|
||
removing it, and configuring is exactly what was broken.
|
||
|
||
And `remove` leaves the configuration behind by design, which for the Wazuh
|
||
agent means its apt source keeps being fetched on every update, long after
|
||
the package is gone.
|
||
"""
|
||
|
||
def test_remove_is_still_the_default(self, driver):
|
||
"""Callers that did not ask for a purge must not get one: configuration
|
||
somebody may want back is not this function's to delete."""
|
||
_mock_send(driver, "Removing wazuh-agent ...")
|
||
|
||
driver.uninstall_package("wazuh-agent")
|
||
|
||
sent = driver._device.send_command.call_args[0][0]
|
||
assert "apt-get remove" in sent
|
||
assert "purge" not in sent
|
||
|
||
def test_purge_is_asked_for_explicitly(self, driver):
|
||
_mock_send(driver, "Purging configuration files for wazuh-agent ...")
|
||
|
||
driver.uninstall_package("wazuh-agent", purge=True)
|
||
|
||
assert "apt-get purge" in driver._device.send_command.call_args[0][0]
|
||
|
||
def test_a_half_configured_package_falls_back_to_dpkg(self, driver):
|
||
"""`install ok unpacked` is the state apt cannot get out of. On one host
|
||
only `dpkg --purge --force-all` removed it."""
|
||
driver._device.send_command.side_effect = [
|
||
"E: Sub-process /usr/bin/dpkg returned an error code (1)",
|
||
"Removing wazuh-agent (4.14.7-1) ...",
|
||
]
|
||
|
||
result = driver.uninstall_package("wazuh-agent", purge=True)
|
||
|
||
assert result["success"] is True
|
||
second = driver._device.send_command.call_args_list[1][0][0]
|
||
assert "dpkg --purge --force-all" in second
|
||
|
||
def test_the_fallback_is_not_tried_when_the_first_pass_worked(self, driver):
|
||
"""A forced dpkg purge is a bigger hammer than apt and must stay a last
|
||
resort, not a routine second step."""
|
||
_mock_send(driver, "Removing wazuh-agent ...")
|
||
|
||
driver.uninstall_package("wazuh-agent", purge=True)
|
||
|
||
assert driver._device.send_command.call_count == 1
|
||
|
||
def test_a_package_manager_without_purge_still_removes(self, driver):
|
||
"""apk and pacman have no separate purge; asking for one must not turn
|
||
into a failure or a command they do not understand."""
|
||
driver._pkg_manager = "apk"
|
||
_mock_send(driver, "(1/1) Purging wazuh-agent")
|
||
|
||
result = driver.uninstall_package("wazuh-agent", purge=True)
|
||
|
||
assert result["success"] is True
|
||
assert "apk del" in driver._device.send_command.call_args[0][0]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# uninstall_package – success from the exit status, not from prose (netork#267)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _with_rc(output: str, rc: int) -> str:
|
||
"""What the shell prints for a command run through ``_sudo_status``."""
|
||
return f"{output}\n__NETORK_RC={rc}"
|
||
|
||
|
||
class TestSudoStatus:
|
||
"""``_sudo_status`` keeps the exit status that ``|| true`` throws away."""
|
||
|
||
def test_returns_output_and_exit_status(self, driver):
|
||
_mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
|
||
|
||
assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
|
||
"Removing wazuh-agent ...",
|
||
0,
|
||
)
|
||
|
||
def test_a_non_zero_exit_status_is_reported(self, driver):
|
||
_mock_send(driver, _with_rc("E: Unable to locate package nope", 100))
|
||
|
||
assert driver._sudo_status("apt-get remove -y nope")[1] == 100
|
||
|
||
def test_the_status_is_read_right_after_sudo_returns(self, driver):
|
||
"""``$?`` must be read straight after the sudo pipeline — with an
|
||
``|| true`` in between, every command would report 0."""
|
||
driver._sudo_password = "pw" # noqa: S105
|
||
_mock_send(driver, _with_rc("", 0))
|
||
|
||
driver._sudo_status("apt-get remove -y x 2>&1")
|
||
|
||
sent = driver._device.send_command.call_args[0][0]
|
||
assert sent.startswith("echo pw | sudo -S")
|
||
assert sent.endswith("apt-get remove -y x 2>&1; echo __NETORK_RC=$?")
|
||
assert "|| true" not in sent
|
||
|
||
def test_a_missing_marker_means_unknown_not_success(self, driver):
|
||
"""Output cut short before the marker arrived says nothing about the
|
||
exit status; ``None`` says so instead of guessing 0."""
|
||
_mock_send(driver, "Removing wazuh-agent ...")
|
||
|
||
assert driver._sudo_status("apt-get remove -y wazuh-agent") == (
|
||
"Removing wazuh-agent ...",
|
||
None,
|
||
)
|
||
|
||
def test_the_command_echo_is_not_mistaken_for_the_marker(self, driver):
|
||
"""A terminal may echo the command line back; its literal ``$?`` is not
|
||
a number, and only the marker on a line of its own counts."""
|
||
_mock_send(
|
||
driver,
|
||
"sudo apt-get remove -y x; echo __NETORK_RC=$?\nRemoving x ...\n__NETORK_RC=1",
|
||
)
|
||
|
||
output, rc = driver._sudo_status("apt-get remove -y x")
|
||
|
||
assert rc == 1
|
||
assert "__NETORK_RC=1" not in output
|
||
|
||
|
||
class TestUninstallExitStatus:
|
||
"""Whether a removal worked is what the package manager's exit status says.
|
||
|
||
Reading it out of human-readable output was guesswork in both directions:
|
||
apt's commonest failure (``E: Sub-process /usr/bin/dpkg returned an error
|
||
code (1)``) read as success until #240, and a successful removal whose
|
||
prerm merely *mentions* a failure read as a failure.
|
||
"""
|
||
|
||
def test_a_non_zero_exit_is_a_failure_whatever_the_output_says(self, driver):
|
||
"""Nothing in this output matches a failure keyword; only the exit
|
||
status knows."""
|
||
driver._pkg_manager = "dnf"
|
||
_mock_send(driver, _with_rc("Removing: wazuh-agent", 1))
|
||
|
||
result = driver.uninstall_package("wazuh-agent")
|
||
|
||
assert result["success"] is False
|
||
|
||
def test_a_zero_exit_is_a_success_even_if_the_output_mentions_failure(self, driver):
|
||
"""A prerm that cannot stop an already-dead unit prints "Failed" and
|
||
still lets the removal complete."""
|
||
_mock_send(
|
||
driver,
|
||
_with_rc(
|
||
"Removing wazuh-agent (4.14.7-1) ...\n"
|
||
"Failed to stop wazuh-agent.service: Unit wazuh-agent.service not loaded.",
|
||
0,
|
||
),
|
||
)
|
||
|
||
result = driver.uninstall_package("wazuh-agent")
|
||
|
||
assert result["success"] is True
|
||
|
||
def test_the_marker_does_not_reach_the_caller(self, driver):
|
||
_mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
|
||
|
||
result = driver.uninstall_package("wazuh-agent")
|
||
|
||
assert result["output"] == "Removing wazuh-agent ..."
|
||
|
||
def test_the_uninstall_command_keeps_its_exit_status(self, driver):
|
||
_mock_send(driver, _with_rc("Removing wazuh-agent ...", 0))
|
||
|
||
driver.uninstall_package("wazuh-agent")
|
||
|
||
sent = driver._device.send_command.call_args[0][0]
|
||
assert "|| true" not in sent
|
||
assert sent.endswith("; echo __NETORK_RC=$?")
|
||
|
||
def test_apt_failing_by_exit_status_falls_back_to_dpkg(self, driver):
|
||
driver._device.send_command.side_effect = [
|
||
_with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
|
||
_with_rc("Removing wazuh-agent (4.14.7-1) ...", 0),
|
||
]
|
||
|
||
result = driver.uninstall_package("wazuh-agent", purge=True)
|
||
|
||
assert result["success"] is True
|
||
second = driver._device.send_command.call_args_list[1][0][0]
|
||
assert "dpkg --purge --force-all" in second
|
||
assert "|| true" not in second
|
||
assert "__NETORK_RC" not in result["output"]
|
||
|
||
def test_the_dpkg_fallback_failing_is_a_failure(self, driver):
|
||
driver._device.send_command.side_effect = [
|
||
_with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
|
||
_with_rc("dpkg: error processing package wazuh-agent (--purge):", 1),
|
||
]
|
||
|
||
result = driver.uninstall_package("wazuh-agent", purge=True)
|
||
|
||
assert result["success"] is False
|
||
assert "dpkg --purge --force-all" in result["output"]
|
||
|
||
def test_a_zero_exit_does_not_trigger_the_fallback(self, driver):
|
||
"""Even when the output contains words that used to mean failure: apt
|
||
exits 0 for a package that is already gone, which is the state the
|
||
caller asked for."""
|
||
_mock_send(driver, _with_rc("Package 'x' is not installed, so not removed", 0))
|
||
|
||
result = driver.uninstall_package("x", purge=True)
|
||
|
||
assert result["success"] is True
|
||
assert driver._device.send_command.call_count == 1
|
||
|
||
def test_without_an_exit_status_the_output_is_read_as_before(self, driver):
|
||
"""If the marker never arrived, the keyword check is still the best
|
||
answer available — and it errs towards failure on apt's ``E:``."""
|
||
driver._pkg_manager = "dnf"
|
||
_mock_send(driver, "E: Sub-process /usr/bin/dpkg returned an error code (1)")
|
||
|
||
result = driver.uninstall_package("wazuh-agent")
|
||
|
||
assert result["success"] is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_kernel_facts -- the command and its parse live in napalm-device-types
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _kernel_wire(report: str) -> str:
|
||
import base64
|
||
import gzip
|
||
|
||
return "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(report.encode())).decode() + "KFACTS_END"
|
||
|
||
|
||
def test_get_kernel_facts_carries_the_shared_command_across(driver):
|
||
from napalm_device_types import KernelFactsMixin
|
||
from napalm_device_types.kernel import KERNEL_FACTS_COMMAND
|
||
|
||
assert isinstance(driver, KernelFactsMixin)
|
||
|
||
report = "[release]\n6.1.0-25-amd64\n[loaded]\ntipc\n[available]\nkernel/net/tipc/tipc.ko.xz\n"
|
||
with patch.object(driver, "_send", return_value=_kernel_wire(report)) as send:
|
||
facts = driver.get_kernel_facts()
|
||
|
||
assert send.call_args.args[0] == KERNEL_FACTS_COMMAND
|
||
assert facts["release"] == "6.1.0-25-amd64"
|
||
assert facts["loaded"] == ["tipc"]
|
||
assert facts["available"] == ["tipc"]
|
||
assert facts["builtin"] is None
|
||
|
||
|
||
def test_get_kernel_facts_raises_on_output_without_a_report(driver):
|
||
with patch.object(driver, "_send", return_value="sh: base64: not found"):
|
||
with pytest.raises(ValueError):
|
||
driver.get_kernel_facts()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Services: listed in one round trip, controlled through systemctl (#7)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_REPORT = (
|
||
"SVC_BEGIN\n[files]\ncron.service enabled enabled\n[units]\n"
|
||
"MainPID=640\nId=cron.service\nNames=cron.service\nLoadState=loaded\n"
|
||
"ActiveState=active\nSubState=running\nUnitFileState=enabled\n"
|
||
"[generated]\nSVC_END\n"
|
||
)
|
||
|
||
|
||
class TestGetServices:
|
||
def test_one_command_lists_every_service(self, driver):
|
||
driver._device.send_command.return_value = _REPORT
|
||
|
||
services = driver.get_services()
|
||
|
||
assert services == [{"name": "cron", "running": True, "enabled": True, "pid": 640}]
|
||
assert driver._device.send_command.call_count == 1
|
||
assert "systemctl show" in driver._device.send_command.call_args[0][0]
|
||
|
||
def test_a_host_without_systemd_falls_back_to_service(self, driver):
|
||
driver._device.send_command.side_effect = [
|
||
"SVC_BEGIN\n[no-systemd]\n[files]\n[units]\n[generated]\nSVC_END\n",
|
||
" [ + ] cron\n [ - ] rsync\n",
|
||
]
|
||
|
||
services = driver.get_services()
|
||
|
||
assert {s["name"]: s["running"] for s in services} == {"cron": True, "rsync": False}
|
||
assert "service --status-all" in driver._device.send_command.call_args[0][0]
|
||
|
||
|
||
class TestManageService:
|
||
def _sent(self, driver) -> list[str]:
|
||
return [c[0][0] for c in driver._device.send_command.call_args_list]
|
||
|
||
def test_as_root_the_command_runs_as_it_is(self, driver):
|
||
driver._device.send_command.side_effect = ["0", "__SVC_RC=0"]
|
||
|
||
assert driver.manage_service("cron", "restart") == {"success": True, "output": ""}
|
||
uid, action = self._sent(driver)
|
||
assert uid == "id -u"
|
||
assert action.startswith("timeout 45 systemctl --no-ask-password restart -- cron.service")
|
||
|
||
def test_with_a_sudo_password_it_goes_through_sudo(self, driver):
|
||
driver._sudo_password = "pw" # noqa: S105
|
||
driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"]
|
||
|
||
assert driver.manage_service("cron", "stop")["success"] is True
|
||
action = self._sent(driver)[1]
|
||
assert action.startswith("echo pw | sudo -S")
|
||
assert "timeout 45 systemctl --no-ask-password stop -- cron.service" in action
|
||
|
||
def test_without_one_sudo_never_waits_for_a_password(self, driver):
|
||
driver._device.send_command.side_effect = ["1000", "__SVC_RC=0"]
|
||
|
||
driver.manage_service("cron", "enable")
|
||
|
||
assert self._sent(driver)[1].startswith("sudo -n timeout 45 systemctl")
|
||
|
||
def test_a_missing_sudo_password_is_explained(self, driver):
|
||
driver._device.send_command.side_effect = [
|
||
"1000",
|
||
"sudo: a password is required\n__SVC_RC=1",
|
||
]
|
||
|
||
result = driver.manage_service("cron", "restart")
|
||
|
||
assert result["success"] is False
|
||
assert "sudo password" in result["output"]
|
||
assert "NOPASSWD" in result["output"]
|
||
|
||
def test_a_failure_keeps_systemctls_message(self, driver):
|
||
driver._device.send_command.side_effect = [
|
||
"0",
|
||
"Failed to start nope.service: Unit nope.service not found.\n__SVC_RC=5",
|
||
]
|
||
|
||
result = driver.manage_service("nope", "start")
|
||
|
||
assert result == {
|
||
"success": False,
|
||
"output": "Failed to start nope.service: Unit nope.service not found.",
|
||
}
|
||
|
||
def test_who_the_user_is_is_asked_once(self, driver):
|
||
driver._device.send_command.side_effect = ["0", "__SVC_RC=0", "__SVC_RC=0"]
|
||
|
||
driver.manage_service("cron", "stop")
|
||
driver.manage_service("cron", "start")
|
||
|
||
assert self._sent(driver).count("id -u") == 1
|
||
|
||
def test_an_invalid_name_is_refused_before_anything_is_sent(self, driver):
|
||
with pytest.raises(ValueError):
|
||
driver.manage_service("cron; reboot", "stop")
|
||
|
||
assert driver._device.send_command.call_count == 0
|