Christian Manivong
ac288823a7
uninstall_package judged success by searching apt/dnf/apk/pacman output
for failure words. That is guesswork in both directions: apt's commonest
failure ("E: Sub-process /usr/bin/dpkg returned an error code (1)") read
as success until the previous change, and a prerm that prints "Failed to
stop ..." while the removal completes still reads as failure. The exit
status is the answer the package manager actually gives, but every
command went through `_sudo(... || true)`, which throws it away.
Add `_sudo_status()`, which runs the command via `_sudo` followed by
`; echo __NETORK_RC=$?` and returns `(output, exit_status)` with the
marker stripped. The `|| true` of other `_sudo` callers is untouched:
they still want output rather than a status. The marker is matched only
on a line of its own with digits, so an echoed command line (literal
`$?`) is never mistaken for it. If the marker never arrives the status
is None -- unknown, not success.
uninstall_package and its dpkg fallback now use it, and
`_uninstall_failed(output, rc)` lets rc decide whenever it is known,
falling back to the keyword check only when it is not.
Behaviour change worth knowing: removing a package that is not installed
exits 0 on apt (and dnf), so it now reports success where the keyword
"is not installed" used to report failure. The package is absent
afterwards, which is what the caller asked for, and netOrk dropping it
from the installed record is then correct.
Refs christianmanivong/netork#267
1148 lines
46 KiB
Python
1148 lines
46 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
|