fix: decide uninstall success by exit status, not by keywords
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
This commit is contained in:
+54
-21
@@ -50,6 +50,12 @@ logger = logging.getLogger("napalm_linux")
|
||||
# Package managers in detection order
|
||||
_PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"]
|
||||
|
||||
#: Printed after a command by ``_sudo_status`` so its exit status survives the
|
||||
#: trip through an interactive shell. Matched only on a line of its own with a
|
||||
#: number after it — an echoed command line carries the literal ``$?`` instead.
|
||||
_RC_MARKER = "__NETORK_RC="
|
||||
_RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE)
|
||||
|
||||
# DMI field values that carry no useful information (OEM defaults, blanks)
|
||||
_BAD_DMI: frozenset[str] = frozenset({
|
||||
"", "none", "n/a", "not specified", "not applicable",
|
||||
@@ -267,6 +273,25 @@ class LinuxDriver(OSDriver):
|
||||
return self._send(wrapped, read_timeout=read_timeout)
|
||||
return self._send(f'sudo {command}', read_timeout=read_timeout)
|
||||
|
||||
def _sudo_status(self, command: str, read_timeout: float = 100) -> tuple[str, int | None]:
|
||||
"""Run *command* via sudo and return ``(output, exit_status)``.
|
||||
|
||||
``_sudo`` callers append ``|| true`` so a failing command yields output
|
||||
instead of an error, which throws the exit status away. This variant
|
||||
echoes ``$?`` straight after the sudo pipeline instead — sudo passes
|
||||
the command's status through, and a failed password is non-zero too.
|
||||
|
||||
The status is ``None`` when the marker never arrived (output cut short),
|
||||
so a caller can tell "unknown" from "succeeded".
|
||||
"""
|
||||
raw = self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout)
|
||||
matches = list(_RC_MARKER_RE.finditer(raw))
|
||||
if not matches:
|
||||
return raw, None
|
||||
last = matches[-1]
|
||||
output = (raw[: last.start()] + raw[last.end():]).strip()
|
||||
return output, int(last.group(1))
|
||||
|
||||
def _detect_pkg_manager(self) -> str | None:
|
||||
"""Return the first package manager binary found on PATH."""
|
||||
for pm in _PKG_MANAGERS:
|
||||
@@ -1046,6 +1071,7 @@ class LinuxDriver(OSDriver):
|
||||
return {"success": success, "output": raw.strip()}
|
||||
|
||||
#: Words in a package manager's output that mean it did not do the job.
|
||||
#: Only consulted when the exit status is unknown; see ``_uninstall_failed``.
|
||||
_UNINSTALL_FAILED = ("error:", "failed", "not found", "is not installed", "no packages")
|
||||
|
||||
def uninstall_package(self, name: str, purge: bool = False) -> dict[str, Any]:
|
||||
@@ -1077,42 +1103,49 @@ class LinuxDriver(OSDriver):
|
||||
pm = self._pkg_manager
|
||||
if pm == "apt":
|
||||
action = "purge" if purge else "remove"
|
||||
raw = self._sudo(
|
||||
f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1 || true"
|
||||
)
|
||||
cmd = f"DEBIAN_FRONTEND=noninteractive apt-get {action} -y {safe} 2>&1"
|
||||
elif pm in ("dnf", "yum"):
|
||||
raw = self._sudo(f"{pm} remove -y {safe} 2>&1 || true")
|
||||
cmd = f"{pm} remove -y {safe} 2>&1"
|
||||
elif pm == "apk":
|
||||
# apk and pacman have no separate purge; asking for one is not an
|
||||
# error, it simply has nothing extra to do.
|
||||
raw = self._sudo(f"apk del {safe} 2>&1 || true")
|
||||
cmd = f"apk del {safe} 2>&1"
|
||||
elif pm == "pacman":
|
||||
raw = self._sudo(f"pacman -R --noconfirm {safe} 2>&1 || true")
|
||||
cmd = f"pacman -R --noconfirm {safe} 2>&1"
|
||||
else:
|
||||
return {"success": False, "output": f"Unsupported package manager: {pm}"}
|
||||
|
||||
if self._uninstall_failed(raw) and pm == "apt":
|
||||
forced = self._sudo(f"dpkg --purge --force-all {safe} 2>&1 || true")
|
||||
if not self._uninstall_failed(forced):
|
||||
return {
|
||||
"success": True,
|
||||
"output": f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}",
|
||||
}
|
||||
raw, rc = self._sudo_status(cmd)
|
||||
failed = self._uninstall_failed(raw, rc)
|
||||
|
||||
if failed and pm == "apt":
|
||||
forced, forced_rc = self._sudo_status(f"dpkg --purge --force-all {safe} 2>&1")
|
||||
raw = f"{raw.strip()}\n--- dpkg --purge --force-all ---\n{forced.strip()}"
|
||||
failed = self._uninstall_failed(forced, forced_rc)
|
||||
|
||||
return {"success": not self._uninstall_failed(raw), "output": raw.strip()}
|
||||
return {"success": not failed, "output": raw.strip()}
|
||||
|
||||
def _uninstall_failed(self, output: str) -> bool:
|
||||
"""Whether the package manager said it did not do the job.
|
||||
def _uninstall_failed(self, output: str, rc: int | None = None) -> bool:
|
||||
"""Whether the package manager did not do the job.
|
||||
|
||||
apt prefixes its own errors with ``E: `` at the start of a line, and
|
||||
the commonest of them — ``E: Sub-process /usr/bin/dpkg returned an
|
||||
error code (1)`` — contains neither "error:" nor "failed". The keyword
|
||||
list alone therefore read a failed removal as a success, which is the
|
||||
worst direction for this particular answer to be wrong in.
|
||||
The exit status decides whenever there is one (netork#267): it is the
|
||||
answer the package manager actually gives, where the output is prose
|
||||
that every tool phrases differently. A prerm printing "Failed to stop
|
||||
…" while the removal completes is a success; a non-zero exit with
|
||||
nothing alarming in the output is not.
|
||||
|
||||
Only when the status is unknown (``rc is None``) is the output read,
|
||||
as the best answer left. apt prefixes its own errors with ``E: `` at
|
||||
the start of a line, and the commonest of them — ``E: Sub-process
|
||||
/usr/bin/dpkg returned an error code (1)`` — contains neither "error:"
|
||||
nor "failed". The keyword list alone therefore read a failed removal
|
||||
as a success, which is the worst direction for this particular answer
|
||||
to be wrong in.
|
||||
|
||||
Matched at line start rather than anywhere: "note: " ends in "e: ".
|
||||
"""
|
||||
if rc is not None:
|
||||
return rc != 0
|
||||
low = output.lower()
|
||||
if any(line.lstrip().startswith("e: ") for line in low.splitlines()):
|
||||
return True
|
||||
|
||||
@@ -984,3 +984,164 @@ class TestUninstallPackage:
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user