# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
from __future__ import annotations
import errno
import logging
import os
import re
import shlex
import shutil
import subprocess
import threading
import time
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from click_extra import (
COLOR_ENVVARS,
get_current_theme as theme,
terminate_live_processes,
)
from click_extra.execution import _LIVE_PROCESSES
from extra_platforms import ALL_PLATFORMS, UNIX, is_any_windows
from extra_platforms.pytest import write_fake_executable
from meta_package_manager.capabilities import Operations
from meta_package_manager.execution import (
_DIAGNOSIS_EXEMPT_OPERATIONS,
_MUTATING_OPERATIONS,
DEFAULT_TIMEOUT,
DIAGNOSIS_HEAD_LINES,
DIAGNOSIS_TAIL_LINES,
MUTATING_TIMEOUT,
OPERATION_TIMEOUTS,
PLAN_RECORDER,
READ_ONLY_TIMEOUT,
SPINNER_DELAY,
UNRUNNABLE_ERRNOS,
VERSION_PROBE,
WIN_DEFAULT_PATHEXT,
CLIError,
CLIExecutor,
_lean_command,
_spinner_label,
format_plan_command,
)
from meta_package_manager.managers.npm import NPM
from meta_package_manager.pool import pool
from meta_package_manager.sudo import (
_SUDO_CACHE_WARM,
ESCALATORS,
UNANSWERED_PROMPT_DETAIL,
Escalator,
)
from .fake_manager import FakeManager
from .test_sudo import only_escalator
# A UNIX and a non-UNIX platform, pinned so the tests read the same on every host:
# escalation itself ignores the platform, while the dormant-marker bookkeeping of
# `build_cli` still gates on UNIX.
_UNIX_PLATFORM = next(iter(UNIX))
_NON_UNIX_PLATFORM = next(p for p in ALL_PLATFORMS if p not in UNIX)
[docs]
def test_operation_timeouts_cover_all_operations():
"""Every routable operation has a per-operation timeout default.
So the map and the `Operations` enum never drift apart.
"""
for operation in Operations:
assert operation.name in OPERATION_TIMEOUTS
[docs]
def test_operation_name_sets_match_enum():
"""The literal operation-name sets stay subsets of the `Operations` vocabulary.
`_MUTATING_OPERATIONS` (plan-mode capture) and the
`internal_sudo_operations` of each manager (watchdog arming) are spelled as
string literals: renaming an `Operations` member would otherwise silently
desync them, stopping plan capture or watchdog arming for the renamed
operation.
Only a mutating operation can run the tool's own `sudo`, so the second set
stays inside the first on every manager.
"""
operation_names = {operation.name for operation in Operations}
assert _MUTATING_OPERATIONS <= operation_names
for manager in pool.values():
assert manager.internal_sudo_operations <= _MUTATING_OPERATIONS, manager.id
# The timeout map covers nothing outside the enum plus the version probe.
assert set(OPERATION_TIMEOUTS) <= operation_names | {VERSION_PROBE}
[docs]
def test_read_only_tier_is_shorter_than_mutating_tier():
"""Read-only probes fail faster than state-changing operations."""
assert READ_ONLY_TIMEOUT < MUTATING_TIMEOUT
# Unknown operations stay on the conservative (long) side.
assert DEFAULT_TIMEOUT == MUTATING_TIMEOUT
[docs]
@pytest.mark.parametrize("active_operation", (None, "version", "search", "install"))
def test_resolve_timeout_explicit_wins(active_operation):
"""An explicit timeout wins for every operation, including unknown ones.
`--timeout` or a per-manager override.
"""
manager = FakeManager()
manager.timeout = 42
manager._active_operation = active_operation
assert manager._resolve_timeout() == 42
[docs]
@pytest.mark.parametrize(
("active_operation", "expected"),
(
("version", READ_ONLY_TIMEOUT),
("installed", READ_ONLY_TIMEOUT),
("outdated", READ_ONLY_TIMEOUT),
("search", READ_ONLY_TIMEOUT),
("install", MUTATING_TIMEOUT),
("upgrade", MUTATING_TIMEOUT),
("upgrade_all", MUTATING_TIMEOUT),
("remove", MUTATING_TIMEOUT),
("sync", MUTATING_TIMEOUT),
("cleanup", MUTATING_TIMEOUT),
# A CLI call with no known operation gets the conservative default.
(None, DEFAULT_TIMEOUT),
),
)
def test_resolve_timeout_per_operation(active_operation, expected):
"""With no explicit timeout, each operation resolves to its tier default."""
manager = FakeManager()
manager.timeout = None
manager._active_operation = active_operation
assert manager._resolve_timeout() == expected
[docs]
@pytest.mark.parametrize(
("progress", "live"),
(
# Without the progress opt-in, the spinner is forced off.
pytest.param(False, "never", id="progress-off"),
# With progress on, the spinner auto-detects a TTY at runtime.
pytest.param(True, "auto", id="progress-on"),
),
)
def test_make_spinner_live_mode_follows_progress(progress, live):
manager = FakeManager()
manager.progress = progress
assert manager._make_spinner().live == live
[docs]
@pytest.fixture
def plain_labels(monkeypatch):
"""Force spinner labels to plain text, whatever the host terminal answers.
Every recognized color variable is cleared and `NO_COLOR` set, so the label
carries no ANSI and the assertions below compare bare strings. Without it a
developer's `FORCE_COLOR` would style the label and fail them.
"""
for envvar in COLOR_ENVVARS:
monkeypatch.delenv(envvar, raising=False)
monkeypatch.setenv("NO_COLOR", "1")
# One argv per shape the strip has to handle, all captured from a real `--dry-run`
# except the escalated one. `cli_path` is what splits an escalation prefix off the
# manager's own binary.
LEAN_COMMAND_CASES = (
# Trailing scope selector only, the `cask` shape.
(
"/opt/homebrew/bin/brew upgrade --quiet --yes --cask",
"/opt/homebrew/bin/brew",
(),
("--cask",),
"brew upgrade --quiet --yes",
),
# Leading plumbing only, the `npm` shape: the case that motivated the strip.
(
(
"/opt/homebrew/bin/npm --global --no-progress --no-update-notifier"
" --no-fund --no-audit search --json --no-description jq"
),
"/opt/homebrew/bin/npm",
(
"--global",
"--no-progress",
"--no-update-notifier",
"--no-fund",
"--no-audit",
),
(),
"npm search --json --no-description jq",
),
# Neither, the version-probe shape: nothing to strip, nothing lost.
("/usr/bin/apt --version", "/usr/bin/apt", ("--quiet",), (), "apt --version"),
# An escalated call keeps its prefix, both binaries read by base name.
(
"/usr/bin/sudo --non-interactive /usr/bin/apt --quiet install --yes jq",
"/usr/bin/apt",
("--quiet",),
(),
"sudo --non-interactive apt install --yes jq",
),
# An argument needing shell quoting survives as one argument.
(
"/opt/homebrew/bin/brew search --quiet '/^jq$/' --formula",
"/opt/homebrew/bin/brew",
(),
("--formula",),
"brew search --quiet '/^jq$/'",
),
)
[docs]
@pytest.mark.parametrize(
("argv", "cli_path", "pre_args", "post_args", "expected"),
LEAN_COMMAND_CASES,
)
def test_lean_command_strips_manager_plumbing(
argv, cli_path, pre_args, post_args, expected
):
"""Static plumbing is dropped; every operation-specific argument is kept."""
assert _lean_command(shlex.split(argv), cli_path, pre_args, post_args) == expected
[docs]
def test_lean_command_keeps_unmatched_plumbing():
"""The strip matches the actual prefix and suffix, never slices by length.
A call built with `auto_pre_args=False` carries none of the declared
plumbing, so a length-based slice would eat a real argument instead.
"""
argv = ("/usr/bin/gem", "--version")
assert _lean_command(argv, "/usr/bin/gem", ("--norc", "--backtrace")) == (
"gem --version"
)
[docs]
def test_lean_command_empty_argv():
assert _lean_command(()) == ""
[docs]
def test_spinner_label_is_plain_without_styling(plain_labels):
assert (
_spinner_label("cask.upgrade_all", "brew upgrade")
== "cask.upgrade_all: brew upgrade"
)
[docs]
def test_styling_gate_reaches_worker_threads(monkeypatch):
"""`--no-color` reaches a label composed on a worker thread.
Trail lines are composed on the dispatch's workers, where the thread-local
command context is absent. The gate reads the color the invocation published
for every thread instead, and that answer outranks the environment.
"""
for envvar in COLOR_ENVVARS:
monkeypatch.delenv(envvar, raising=False)
monkeypatch.setenv("FORCE_COLOR", "1")
monkeypatch.setattr("click_extra.color._invocation_color", False)
labels = []
worker = threading.Thread(
target=lambda: labels.append(_spinner_label("cask.upgrade_all", "brew upgrade"))
)
worker.start()
worker.join()
assert labels == ["cask.upgrade_all: brew upgrade"]
[docs]
def test_spinner_label_subject_alone_without_styling(plain_labels):
assert _spinner_label("cask.upgrade_all", "") == "cask.upgrade_all"
[docs]
@pytest.mark.parametrize(
("term", "italic"),
(
# Measured with `infocmp`: these declare `sitm`, the rest do not.
("xterm-256color", True),
("xterm-ghostty", True),
("tmux-256color", True),
("screen-256color", False),
("linux", False),
),
)
def test_spinner_label_italic_follows_terminal_capability(monkeypatch, term, italic):
"""The subject is always painted; the command takes italic where it renders.
The subject is compared against the theme's own rendering rather than a
literal escape, the `invoked_command` slot resolving to different codes on a
light and a dark background.
"""
for envvar in COLOR_ENVVARS:
monkeypatch.delenv(envvar, raising=False)
monkeypatch.setenv("FORCE_COLOR", "1")
monkeypatch.setenv("TERM", term)
label = _spinner_label("cask.upgrade_all", "brew upgrade")
assert theme().invoked_command("cask.upgrade_all") in label
assert ("\x1b[3m" in label) is italic
[docs]
def test_make_spinner_label_names_operation_and_command(plain_labels):
manager = FakeManager()
manager._active_operation = "search"
label = manager._make_spinner((str(manager.cli_path), "--version")).label
assert label == f"{manager.id}.search: {manager.cli_path.name} --version"
[docs]
def test_make_spinner_label_without_operation(plain_labels):
manager = FakeManager()
manager._active_operation = None
assert manager._make_spinner().label == str(manager.id)
[docs]
@pytest.mark.parametrize(
("color_enabled", "expected"),
(
(True, "\x1b[2m (24.2s)\x1b[0m"),
(False, " (24.2s)"),
),
)
def test_spinner_clock_dims_the_parentheses_too(color_enabled, expected):
"""The whole fragment is faint, brackets included, or none of it is.
Painting the duration alone through the `timer` callable would leave the
parentheses upstream adds around it at full weight, which is why the spinner
takes a `timer_style`.
"""
spinner = FakeManager()._make_spinner()
spinner._start_time, spinner._stop_time = 0.0, 24.2
lines = spinner.frame_lines(color=color_enabled)
assert lines
assert all(line.endswith(expected) for line in lines)
[docs]
def test_spinner_clock_is_plain_before_start():
"""A spinner that never started reports color off, so it never dims."""
spinner = FakeManager()._make_spinner()
assert spinner._color is False
# Tiny cross-platform CLIs that exit non-zero, differing only in which stream
# carries the diagnostic. `FakeManager` runs the Python interpreter as its
# binary, so `run_cli("-c", ...)` executes these directly.
FAIL_ON_STDOUT = "import sys; sys.stdout.write('boom'); sys.exit(8)"
"""Fails the steamcmd way: a non-zero exit with its message on `<stdout>` and an
empty `<stderr>`."""
FAIL_ON_STDERR = "import sys; sys.stderr.write('boom'); sys.exit(8)"
"""Fails the conventional way: a non-zero exit with its message on `<stderr>`."""
[docs]
@pytest.mark.parametrize(
("stop_on_error", "must_succeed", "script", "expectation"),
(
# The steamcmd-on-Windows regression: in the action context (a patched
# stop_on_error, no must_succeed) a non-zero exit with an empty <stderr>
# must now be a failure, not a silent success.
pytest.param(True, False, FAIL_ON_STDOUT, "raise", id="action-empty-stderr"),
# A parsed read (must_succeed) tolerates the same exit as a benign status
# code, like npm and pnpm outdated exiting 1 when updates exist.
pytest.param(False, True, FAIL_ON_STDOUT, "tolerate", id="read-empty-stderr"),
# Neither opt-in: the lenient default tolerates it too.
pytest.param(
False, False, FAIL_ON_STDOUT, "tolerate", id="default-empty-stderr"
),
# A conventional failure (<stderr> populated) still raises when the caller
# demanded success, whether via the action context or must_succeed...
pytest.param(True, False, FAIL_ON_STDERR, "raise", id="action-stderr"),
pytest.param(False, True, FAIL_ON_STDERR, "raise", id="read-stderr"),
# ...and is accumulated for the end-of-run summary otherwise.
pytest.param(False, False, FAIL_ON_STDERR, "accumulate", id="default-stderr"),
),
)
def test_run_failure_gate(stop_on_error, must_succeed, script, expectation):
"""The failure gate decides to raise, accumulate or tolerate.
From the exit code, the `<stderr>` content, and the `stop_on_error`/`must_succeed`
context.
"""
manager = FakeManager()
manager.stop_on_error = stop_on_error
if expectation == "raise":
with pytest.raises(CLIError) as excinfo:
manager.run_cli("-c", script, must_succeed=must_succeed)
assert excinfo.value.code == 8
# Raised or not, the failure is accumulated: the very same assertion as
# the "accumulate" branch below.
assert [error.code for error in manager.cli_errors] == [8]
elif expectation == "accumulate":
manager.run_cli("-c", script, must_succeed=must_succeed)
assert [error.code for error in manager.cli_errors] == [8]
else: # tolerate.
output = manager.run_cli("-c", script, must_succeed=must_succeed)
assert output == "boom"
assert manager.cli_errors == []
NOISY_STATUS = (
"import sys; sys.stdout.write('boom'); sys.stderr.write('chatter\\n'); sys.exit(8)"
)
"""Exits non-zero as a status, with only routine chatter on `<stderr>`."""
NOISY_FAILURE = "import sys; sys.stderr.write('chatter\\nbroken\\n'); sys.exit(8)"
"""Fails for real, its diagnosis written after the same routine chatter."""
[docs]
def test_run_failure_gate_tolerates_stderr_noise(monkeypatch):
"""A `<stderr>` holding only declared noise is as good as a silent one."""
manager = FakeManager()
monkeypatch.setattr(manager, "stderr_noise", re.compile(r"^chatter$"))
assert manager.run_cli("-c", NOISY_STATUS, must_succeed=True) == "boom"
assert manager.cli_errors == []
# The snapshot keeps the whole stream.
assert manager._last_run == (8, "boom", "chatter")
[docs]
def test_run_failure_gate_drops_stderr_noise_from_the_diagnosis(monkeypatch):
"""A real failure still raises, its diagnosis cleared of the noise."""
manager = FakeManager()
monkeypatch.setattr(manager, "stderr_noise", re.compile(r"^chatter$"))
with pytest.raises(CLIError) as excinfo:
manager.run_cli("-c", NOISY_FAILURE, must_succeed=True)
assert excinfo.value.error == "broken"
assert manager.cli_errors == [excinfo.value]
# Diagnosis relay: a failed run promotes its own error report to WARNING at the
# failure gate, so the default verbosity carries the "why" and not just the ✘
# signal (issue 1968). Successful chatter, tolerated exits, DEBUG-level runs and
# exempt operations stay silent.
DIAGNOSIS_CAP = DIAGNOSIS_HEAD_LINES + DIAGNOSIS_TAIL_LINES
"""Longest report `CLIError.diagnosis` relays whole."""
def _numbered_lines(stop, start=0):
"""Lines `line-{start}` up to `line-{stop - 1}`, one per row."""
return "\n".join(f"line-{i}" for i in range(start, stop))
[docs]
@pytest.mark.parametrize(
("error", "output", "expected"),
(
pytest.param("boom-err", "boom-out", "boom-err", id="stderr-first"),
pytest.param("", "boom-out", "boom-out", id="stdout-fallback"),
pytest.param(" \n ", "", "Exited 8 with no output.", id="silent-death"),
pytest.param(
_numbered_lines(DIAGNOSIS_CAP),
"",
_numbered_lines(DIAGNOSIS_CAP),
id="cap-boundary-untouched",
),
pytest.param(
_numbered_lines(DIAGNOSIS_CAP + 1),
"",
_numbered_lines(DIAGNOSIS_HEAD_LINES)
+ "\n(...) 1 line truncated.\n"
+ _numbered_lines(DIAGNOSIS_CAP + 1, start=DIAGNOSIS_HEAD_LINES + 1),
id="cap-overflow-singular",
),
pytest.param(
_numbered_lines(DIAGNOSIS_CAP + 2),
"",
_numbered_lines(DIAGNOSIS_HEAD_LINES)
+ "\n(...) 2 lines truncated.\n"
+ _numbered_lines(DIAGNOSIS_CAP + 2, start=DIAGNOSIS_HEAD_LINES + 2),
id="cap-overflow-plural",
),
),
)
def test_cli_error_diagnosis(error, output, expected):
"""`CLIError.diagnosis` prefers `<stderr>`, then falls back on `<stdout>`.
It states a silent death, and keeps the head and the tail around a truncation
counter.
"""
assert CLIError(8, output, error).diagnosis == expected
[docs]
@pytest.mark.parametrize(
("code", "error", "expected"),
(
pytest.param(
100,
"E: Unable to locate package jq",
"<CLIError(100, 'E: Unable to locate package jq')>",
id="short",
),
pytest.param(
1,
"\n".join((
"ERROR: the orchard gate is locked",
*(f"\tframe-{i}" for i in range(4)),
)),
"<CLIError(1, 'ERROR: the orchard gate is locked frame-0 frame-1(...)')>",
id="folded-and-shortened",
),
),
)
def test_cli_error_repr(code, error, expected):
"""The repr folds the error onto one line, and shortens it."""
assert repr(CLIError(code, "", error)) == expected
[docs]
def test_cli_error_diagnosis_keeps_an_opening_error():
"""A report that opens with its error and trails a backtrace keeps the error.
The tail alone would hold frames and no reason.
"""
report = "\n".join((
"ERROR: the orchard gate is locked",
*(f"\tframe-{i}" for i in range(DIAGNOSIS_CAP * 2)),
))
diagnosis = CLIError(1, "", report).diagnosis.splitlines()
assert diagnosis[0] == "ERROR: the orchard gate is locked"
assert diagnosis[-1] == f"\tframe-{DIAGNOSIS_CAP * 2 - 1}"
assert len(diagnosis) == DIAGNOSIS_CAP + 1
[docs]
@pytest.mark.parametrize(
("stop_on_error", "script", "expected"),
(
# The conventional failure relays its <stderr> tail.
pytest.param(False, FAIL_ON_STDERR, "boom", id="accumulated-stderr"),
# The action context relays the steamcmd-style <stdout> report.
pytest.param(True, FAIL_ON_STDOUT, "boom", id="raised-stdout-fallback"),
),
)
def test_failed_run_relays_diagnosis(stop_on_error, script, expected, caplog):
"""The failure gate logs the diagnosis at WARNING, tagged with the manager.
For accumulated and raised failures alike.
"""
manager = FakeManager()
manager.stop_on_error = stop_on_error
with caplog.at_level(logging.INFO):
if stop_on_error:
with pytest.raises(CLIError):
manager.run_cli("-c", script)
else:
manager.run_cli("-c", script)
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert [r.getMessage() for r in warnings] == [expected]
assert warnings[0].label == manager.subject
[docs]
@pytest.mark.parametrize(
("script", "level"),
(
# A tolerated non-zero exit (empty <stderr>) is not a failure.
pytest.param(FAIL_ON_STDOUT, logging.INFO, id="tolerated-exit"),
# A successful command's <stderr> chatter stays at DEBUG.
pytest.param(
"import sys; sys.stderr.write('chatter')",
logging.INFO,
id="success-chatter",
),
# At DEBUG verbosity the raw output was already streamed inline.
pytest.param(FAIL_ON_STDERR, logging.DEBUG, id="debug-streams-inline"),
),
)
def test_no_diagnosis_relay(script, level, caplog):
"""No WARNING relay without a genuine failure.
Nor when DEBUG already streamed the raw output inline.
"""
manager = FakeManager()
with caplog.at_level(level):
manager.run_cli("-c", script)
assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
[docs]
@pytest.mark.parametrize("operation", sorted(_DIAGNOSIS_EXEMPT_OPERATIONS))
def test_exempt_operations_skip_diagnosis_relay(operation, caplog):
"""Version probes and doctor fail without the WARNING relay.
Version probes fire per candidate on selection, and doctor relays its report
verbatim.
"""
manager = FakeManager()
with caplog.at_level(logging.INFO), manager.acting_as(operation):
manager.run_cli("-c", FAIL_ON_STDERR)
assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
# The relay exemption says nothing about accumulation: only the probe is
# exempt from that too, and it has its own test below.
expected = [] if operation == VERSION_PROBE else [8]
assert [error.code for error in manager.cli_errors] == expected
[docs]
def test_version_probe_failure_stays_out_of_cli_errors():
"""A binary rejecting the version verb records no error.
The probe is how mpm settles that a `PATH` hit is not the manager it names,
so its rejection is a verdict `unavailable_reason` already reports. Left in
`cli_errors`, it named a manager mpm had correctly dropped in the end-of-run
summary of every command: issue 2119, where PearCleaner's `pear` symlink had
`mpm outdated` warn about a PHP manager the user never installed.
"""
manager = FakeManager()
with manager.acting_as(VERSION_PROBE):
manager.run_cli("-c", FAIL_ON_STDERR)
assert manager.cli_errors == []
# The run is still published, so the caller reads an empty <stdout> and
# reports the manager unavailable instead of crashing.
last_run = manager._last_run
assert last_run is not None
assert last_run[0] == 8
[docs]
def test_version_probe_incomplete_run_stays_an_error():
"""A probe that never completed is plumbing, not a verdict.
It stays in the tally, mirroring `doctor`'s own split.
"""
manager = FakeManager()
manager.timeout = 1
with manager.acting_as(VERSION_PROBE):
manager.run_cli("-c", "import time; time.sleep(30)")
assert len(manager.cli_errors) == 1
assert manager.cli_errors[0].code is None
assert "Timed out" in manager.cli_errors[0].error
assert manager._last_run is None
# CLIExecutor.run_cache: a lane's peers replay a byte-identical command instead of
# re-running it, whether the lane is a lock family (dispatch) or a probe group
# (warm_availability). Each script appends a byte to a marker file, so its length counts
# how many subprocesses actually ran.
def _append_script(marker, payload="x", tail=""):
"""A one-liner that appends `payload` to `marker`, then runs `tail`."""
return f"open({str(marker)!r}, 'a', encoding='UTF-8').write({payload!r}); {tail}"
[docs]
def test_run_cache_replays_identical_command(tmp_path):
"""With a shared cache, a byte-identical command runs the subprocess once."""
marker = tmp_path / "runs.log"
script = _append_script(marker)
cache: dict = {}
first, second = FakeManager(), FakeManager()
first.run_cache = second.run_cache = cache
out_first = first.run_cli("-c", script)
out_second = second.run_cli("-c", script)
# The subprocess ran exactly once; the peer was served from the shared cache.
assert marker.read_text(encoding="UTF-8") == "x"
assert out_first == out_second == ""
[docs]
def test_run_cache_disabled_by_default(tmp_path):
"""Without a cache, identical commands each spawn their own subprocess."""
marker = tmp_path / "runs.log"
script = _append_script(marker)
manager = FakeManager()
assert manager.run_cache is None
manager.run_cli("-c", script)
manager.run_cli("-c", script)
assert marker.read_text(encoding="UTF-8") == "xx"
[docs]
def test_run_cache_replays_failure_to_every_member(tmp_path):
"""A cached failure is re-attributed to each peer, though the command runs once."""
marker = tmp_path / "runs.log"
script = _append_script(
marker, tail="import sys; sys.stderr.write('boom'); sys.exit(8)"
)
cache: dict = {}
first, second = FakeManager(), FakeManager()
first.stop_on_error = second.stop_on_error = False
first.run_cache = second.run_cache = cache
first.run_cli("-c", script)
second.run_cli("-c", script)
# One real execution, but both managers recorded the failure for the trail.
assert marker.read_text(encoding="UTF-8") == "x"
assert [error.code for error in first.cli_errors] == [8]
assert [error.code for error in second.cli_errors] == [8]
[docs]
def test_run_cache_keeps_distinct_commands_apart(tmp_path):
"""Only byte-identical invocations collapse; different args each run."""
marker = tmp_path / "runs.log"
manager = FakeManager()
manager.run_cache = {}
manager.run_cli("-c", _append_script(marker, "a"))
manager.run_cli("-c", _append_script(marker, "b"))
assert marker.read_text(encoding="UTF-8") == "ab"
[docs]
def test_run_cache_collapses_dry_run(caplog):
"""Under --dry-run a family's identical command is announced once."""
cache: dict = {}
first, second = FakeManager(), FakeManager()
first.dry_run = second.dry_run = True
first.run_cache = second.run_cache = cache
with caplog.at_level(logging.WARNING):
first.run_cli("-c", "pass")
second.run_cli("-c", "pass")
# The command is announced once; the peer is a silent (INFO) cache hit.
dry_run_lines = [
record
for record in caplog.records
if record.getMessage().startswith("Dry-run:")
]
assert len(dry_run_lines) == 1
# The first dry-run still seeded the cache, so the peer had something to reuse.
assert len(cache) == 1
[docs]
@pytest.mark.parametrize(
("plan", "runs"),
(
pytest.param(False, "rmr", id="ran-for-real"),
# A captured command changed nothing: the read still holds.
pytest.param(True, "r", id="captured-by-plan"),
),
)
def test_run_cache_drops_reads_once_the_lane_changes_state(tmp_path, plan, runs):
"""A read cached before a state-changing command spawns again after it."""
PLAN_RECORDER.reset()
marker = tmp_path / "runs.log"
read = _append_script(marker, "r")
manager = FakeManager()
manager.run_cache = {}
manager.plan = plan
with manager.acting_as("installed"):
manager.run_cli("-c", read)
# Nothing changed yet: the identical read is a replay.
manager.run_cli("-c", read)
with manager.acting_as("upgrade_all"):
manager.run_cli("-c", _append_script(marker, "m"))
with manager.acting_as("installed"):
manager.run_cli("-c", read)
assert marker.read_text(encoding="UTF-8") == runs
[docs]
def test_run_cache_drops_reads_when_the_state_change_times_out(tmp_path):
"""A command killed on its timeout may have changed the system all the same."""
marker = tmp_path / "runs.log"
read = _append_script(marker, "r")
manager = FakeManager()
manager.run_cache = {}
with manager.acting_as("installed"):
manager.run_cli("-c", read)
# The short timeout covers the state change alone: a read it killed on a loaded
# runner would fail the test for a reason the cache has no part in.
manager.timeout = 1
with manager.acting_as("upgrade_all"):
manager.run_cli(
"-c", _append_script(marker, "m", tail="import time; time.sleep(30)")
)
manager.timeout = None
with manager.acting_as("installed"):
manager.run_cli("-c", read)
assert len(manager.cli_errors) == 1
assert "Timed out" in manager.cli_errors[0].error
# The read spawned twice. A child killed before its first write leaves no `m`.
assert marker.read_text(encoding="UTF-8") in {"rmr", "rr"}
[docs]
def test_run_cache_keeps_state_changes_for_the_peers(tmp_path):
"""A state-changing result outlives the next one, so a peer replays both."""
marker = tmp_path / "runs.log"
first, second = FakeManager(), FakeManager()
first.run_cache = second.run_cache = {}
for manager in (first, second):
with manager.acting_as("cleanup"):
manager.run_cli("-c", _append_script(marker, "a"))
manager.run_cli("-c", _append_script(marker, "b"))
assert marker.read_text(encoding="UTF-8") == "ab"
[docs]
def test_run_cache_never_serves_a_read_from_a_state_change(tmp_path):
"""A read sharing the command line of a state change still runs on its own."""
marker = tmp_path / "runs.log"
script = _append_script(marker)
manager = FakeManager()
manager.run_cache = {}
with manager.acting_as("sync"):
manager.run_cli("-c", script)
with manager.acting_as("outdated"):
manager.run_cli("-c", script)
assert marker.read_text(encoding="UTF-8") == "xx"
[docs]
def test_plan_captures_mutating_command_without_running_it(tmp_path):
"""Plan mode records a state-changing command instead of executing it."""
PLAN_RECORDER.reset()
marker = tmp_path / "runs.log"
script = _append_script(marker)
manager = FakeManager()
manager.plan = True
manager._active_operation = "install"
output = manager.run_cli("-c", script)
# The subprocess never ran, yet the command was captured for the plan.
assert not marker.exists()
assert output == ""
captured = PLAN_RECORDER.render()
assert len(captured) == 1
assert str(manager.cli_path) in captured[0]
assert "-c" in captured[0]
[docs]
def test_plan_runs_read_only_operations(tmp_path):
"""Plan mode executes read-only queries so the plan resolves against real state."""
PLAN_RECORDER.reset()
marker = tmp_path / "runs.log"
script = _append_script(marker)
manager = FakeManager()
manager.plan = True
manager._active_operation = "search"
manager.run_cli("-c", script)
# The read ran for real, and nothing was captured as a mutation.
assert marker.read_text(encoding="UTF-8") == "x"
assert PLAN_RECORDER.render() == ()
class _SynthesizedUpgradeFakeManager(FakeManager):
"""Fake with a per-package upgrade CLI but no one-shot upgrade-all.
`upgrade --all` then exercises the synthesized one-by-one fallback of
`PackageManager.upgrade`, and its `outdated` listing runs a real (instant)
subprocess, so the packages only materialize if the read truly executed.
"""
_OUTDATED_REGEXP = re.compile(
r"(?P<package_id>\S+) (?P<installed_version>\S+) (?P<latest_version>\S+)"
)
@property
def outdated(self):
output = self.run_cli("-c", "print('fake-pkg-alpha 1.0.0 1.1.0')")
yield from self.parse_regex_lines(self._OUTDATED_REGEXP, output)
def upgrade_one_cli(self, package_id, version=None):
return self.build_cli("upgrade", package_id)
def upgrade_all_cli(self):
"""No one-shot upgrade-all: force the synthesized per-package fallback."""
raise NotImplementedError
[docs]
def test_plan_synthesized_upgrade_all_resolves_reads(invoke, patch_pool_with):
"""`mpm --plan upgrade` on a synthesized upgrade-all prints each command.
The per-package upgrade commands, resolved against a really-executed `outdated`
read.
Guards the read-stamp of the synthesized fallback in
`PackageManager.upgrade`: the whole run is stamped `upgrade_all` (a
mutating operation), so without re-stamping the inner listing as
`outdated`, plan mode captured the read itself — leaving the plan without
the actual mutating commands it exists to disclose.
"""
patch_pool_with(_SynthesizedUpgradeFakeManager())
result = invoke("--plan", "upgrade", "--all")
assert result.exit_code == 0
plan_lines = result.stdout.splitlines()
# The outdated read resolved one package, planned as its upgrade command.
assert any("upgrade fake-pkg-alpha" in line for line in plan_lines)
# The read itself was executed, never captured into the plan.
assert not any("print(" in line for line in plan_lines)
[docs]
def test_plan_ignores_force_exec(tmp_path):
"""A force_exec read runs for real even under plan mode (version probes, etc.)."""
PLAN_RECORDER.reset()
marker = tmp_path / "runs.log"
script = _append_script(marker)
manager = FakeManager()
manager.plan = True
# A mutating active operation, but force_exec must still run the command.
manager._active_operation = "install"
manager.run_cli("-c", script, force_exec=True)
assert marker.read_text(encoding="UTF-8") == "x"
assert PLAN_RECORDER.render() == ()
# Interrupt handling: the live-subprocess registry and the SIGINT handler live in
# click_extra.execution (and are tested there). This checks mpm's side of the
# contract: CLIExecutor.run() must route its subprocess through run_cli() so the
# child lands in that registry for the duration of the call.
[docs]
def test_run_registers_live_process_then_discards_it():
"""run() tracks its subprocess while it runs, and drops it once done.
A background call parks in a real subprocess. Once it is registered,
terminate_live_processes() unblocks it, and run_cli()'s `finally` clears the
registry: this is the exact path the SIGINT handler drives on Ctrl+C.
"""
manager = FakeManager()
def call():
manager.run_cli("-c", "import time; time.sleep(30)")
worker = threading.Thread(target=call)
worker.start()
try:
deadline = time.monotonic() + 5
while not _LIVE_PROCESSES and time.monotonic() < deadline:
time.sleep(0.01)
assert _LIVE_PROCESSES, "run() should register its live subprocess"
# Terminating the child unblocks the parked run() call.
terminate_live_processes()
worker.join(timeout=5)
assert not worker.is_alive()
finally:
terminate_live_processes()
worker.join(timeout=5)
# run_cli()'s finally discarded the child once it was reaped.
assert not _LIVE_PROCESSES
# Privilege escalation: a per-op marker gated by a per-manager policy, escalated with
# `sudo --non-interactive`. The priming, keepalive, policy inventories and stall
# watchdog are covered
# in test_sudo.py; these exercise the wrapping and failure-hint paths of execution.py.
[docs]
def test_build_cli_escalates_with_sudo_n_when_policy_on():
"""A privileged op escalates to `sudo --non-interactive` when opted in.
When the manager's policy opts in.
The escalator is pinned rather than detected: a Windows runner carries no
`sudo` on `PATH`, and detection finding none would skip the wrapping this
test asserts.
"""
manager = FakeManager()
manager.sudo = True
with (
only_escalator("sudo"),
patch(
"meta_package_manager.execution.os.geteuid", return_value=501, create=True
),
):
cli = manager.build_cli("install", "pkg", sudo=True)
assert cli[:2] == ("sudo", "--non-interactive")
[docs]
def test_build_cli_no_escalation_as_root():
"""Root runs a privileged op bare, whatever escalator the host carries.
Wrapping it would only add ways to fail: `pkexec` needs a running polkit
authority, which an OpenRC host may lack.
"""
manager = FakeManager()
manager.sudo = True
with (
only_escalator("pkexec"),
patch("meta_package_manager.execution.os.geteuid", return_value=0, create=True),
):
cli = manager.build_cli("install", "pkg", sudo=True)
assert cli[0] == str(manager.cli_path)
assert "pkexec" not in cli
[docs]
@pytest.mark.parametrize(
"options",
(
pytest.param(("--keep-cwd",), id="polkit-121"),
pytest.param((), id="polkit-0.104"),
),
)
def test_build_cli_passes_the_options_the_release_accepts(options):
"""The gated options sit between the escalator and the command."""
manager = FakeManager()
manager.sudo = True
with (
only_escalator("pkexec"),
patch.object(Escalator, "supported_options", return_value=options),
patch(
"meta_package_manager.execution.os.geteuid", return_value=501, create=True
),
):
cli = manager.build_cli("install", "pkg", sudo=True)
assert cli[: 3 + len(options)] == (
"pkexec",
"--disable-internal-agent",
*options,
str(manager.cli_path),
)
[docs]
def test_build_cli_no_escalation_when_policy_off():
"""`--no-sudo` (policy False) drops escalation even on a privileged op."""
manager = FakeManager()
manager.sudo = False
cli = manager.build_cli("install", "pkg", sudo=True)
assert "sudo" not in cli
[docs]
def test_build_cli_no_escalation_without_an_escalator():
"""A host carrying none runs unprivileged on any platform.
Rather than crashing on a missing binary.
"""
manager = FakeManager()
manager.sudo = True
with (
patch(
"meta_package_manager.execution.current_platform",
return_value=_NON_UNIX_PLATFORM,
),
only_escalator(None),
):
cli = manager.build_cli("install", "pkg", sudo=True)
assert "sudo" not in cli
assert "gsudo" not in cli
[docs]
def test_build_cli_marker_required_for_escalation():
"""Policy on but no per-op marker (sudo=False default) means no escalation."""
manager = FakeManager()
manager.sudo = True
with patch(
"meta_package_manager.execution.current_platform", return_value=_UNIX_PLATFORM
):
cli = manager.build_cli("list")
assert "sudo" not in cli
[docs]
def test_npm_sudo_marker_dormant_until_opted_in():
"""npm marks its global installs privileged, but stays unescalated.
Until the user opts in via --sudo / config (default_sudo is False).
"""
manager = NPM()
assert manager.default_sudo is False
# Pinned, not detected: a Windows runner carries no `sudo` on `PATH`.
with only_escalator("sudo"):
assert "sudo" not in manager.build_cli("update", sudo=True)
manager.sudo = True
assert manager.build_cli("update", sudo=True)[:2] == (
"sudo",
"--non-interactive",
)
[docs]
def test_dormant_marker_permission_failure_hints_the_opt_in(caplog):
"""A dormant privileged op dying on a permission error draws the hint.
The hint names the scoped opt-in, labeled and placed after the tool's own account of
the refusal.
"""
manager = FakeManager()
assert manager.sudo is None and manager.default_sudo is False
script = (
"import sys; "
"sys.stderr.write('EACCES: permission denied, access /opt/x'); "
"sys.exit(1)"
)
with (
patch(
"meta_package_manager.execution.current_platform",
return_value=_UNIX_PLATFORM,
),
patch(
"meta_package_manager.execution.os.geteuid", return_value=501, create=True
),
caplog.at_level(logging.WARNING),
):
manager.run_cli("-c", script, sudo=True)
hints = [r for r in caplog.records if "marked privileged" in r.getMessage()]
assert len(hints) == 1
assert "`mpm --fakemanager --sudo`" in hints[0].getMessage()
assert "`[mpm.overrides.fakemanager] sudo = true`" in hints[0].getMessage()
assert hints[0].label == manager.subject
[docs]
@pytest.mark.parametrize(
("marker", "stderr", "euid"),
(
pytest.param(False, "EACCES: permission denied", 501, id="no-marker"),
pytest.param(True, "404 not found", 501, id="not-a-permission-error"),
pytest.param(True, "EACCES: permission denied", 0, id="already-root"),
),
)
def test_dormant_marker_hint_negative_gates(caplog, marker, stderr, euid):
"""Each gate individually silences the hint.
No privileged marker, a failure that is not a permission refusal, or a process
already root.
"""
manager = FakeManager()
script = f"import sys; sys.stderr.write({stderr!r}); sys.exit(1)"
with (
patch(
"meta_package_manager.execution.current_platform",
return_value=_UNIX_PLATFORM,
),
patch(
"meta_package_manager.execution.os.geteuid",
return_value=euid,
create=True,
),
caplog.at_level(logging.WARNING),
):
manager.run_cli("-c", script, sudo=marker)
assert not any("marked privileged" in r.getMessage() for r in caplog.records)
[docs]
@pytest.mark.parametrize(
("stderr", "named"),
(
pytest.param("sudo: timed out reading password", True, id="sudo"),
pytest.param("sudo: timed out", True, id="sudo-rs"),
pytest.param("installer: timed out reading password", False, id="unsigned"),
pytest.param("sudo: a password is required", False, id="other-refusal"),
),
)
def test_failed_run_names_a_prompt_its_sudo_gave_up_on(caplog, stderr, named):
"""A tool whose `sudo` stopped waiting for a password fails like a timeout.
Like a timeout of mpm on that prompt: the same advice, after the tool's own account,
and the same cause for the trail line.
The two controls keep the plain relay: a line no `sudo` signed, and a
`sudo` that refuses for another reason.
"""
manager = FakeManager()
script = f"import sys; sys.stderr.write({stderr!r}); sys.exit(1)"
with caplog.at_level(logging.WARNING):
manager.run_cli("-c", script)
advice = (
"Failed on an unanswered password prompt. "
"Run the command again and answer the prompt."
)
messages = [record.getMessage() for record in caplog.records]
assert messages[0] == stderr
assert (advice in messages[1:]) is named
detail = UNANSWERED_PROMPT_DETAIL if named else None
assert manager.cli_errors[0].detail == detail
[docs]
@pytest.mark.skipif(is_any_windows(), reason="escalation is UNIX-only")
@pytest.mark.parametrize(
("escalator_id", "refusal", "returncode"),
(
pytest.param("sudo", "sudo: a password is required\n", 1, id="sudo"),
pytest.param("doas", "doas: Authentication required\n", 1, id="doas"),
# polkit 0.104 on SliTaz 5.0, under `--disable-internal-agent`.
pytest.param(
"pkexec",
"Error executing command as another user: No authentication agent found.\n",
127,
id="pkexec",
),
),
)
def test_run_hints_when_escalation_cannot_authenticate(
tmp_path, monkeypatch, caplog, escalator_id, refusal, returncode
):
"""A real escalator that cannot authenticate triggers the actionable hint.
The hint names that escalator and what its policy needs.
A fake escalator on `PATH` mimics the refusal, so the whole build_cli →
subprocess → failure-gate → hint path runs for real, without root.
"""
# write_fake_executable() writes a Python-shebang stand-in (not "#!/bin/sh"),
# so it still execs in hermetic build sandboxes that ship no /bin/sh (Guix,
# etc.). It mimics the refusal: a diagnostic on <stderr>, non-zero exit.
write_fake_executable(
tmp_path / escalator_id, stderr=refusal, returncode=returncode
)
monkeypatch.setenv("PATH", f"{tmp_path}{os.pathsep}{os.environ['PATH']}")
escalator = next(e for e in ESCALATORS if e.id == escalator_id)
manager = FakeManager()
manager.sudo = True
# `only_escalator` pins the choice, as the neighbouring build_cli tests do.
# Without it the fake above decides it: `write_fake_executable` answers
# every argv the same way, so `sudo --version` returns the credential
# diagnostic instead of `Sudo version`, `resolve_escalator` reads that as an
# impostor and falls through to whatever else the host carries. That is
# `run0` on a systemd runner and nothing on macOS, and `resolve_escalator`
# caches, so the failure only lands when this test is the first in its
# worker to resolve one.
with (
only_escalator(escalator_id),
patch(
"meta_package_manager.execution.os.geteuid", return_value=501, create=True
),
):
cli = manager.build_cli("-c", "pass", sudo=True)
assert cli[: len(escalator.escalate_args)] == escalator.escalate_args
with caplog.at_level(logging.WARNING):
manager.run(*cli)
assert any(
f"{escalator_id} could not authorize the command" in record.getMessage()
and escalator.remedy in record.getMessage()
for record in caplog.records
)
[docs]
@pytest.mark.parametrize("escalate", (True, False))
def test_escalated_calls_keep_the_controlling_terminal(escalate):
"""An escalated call is never detached into a POSIX session of its own.
`sudo` keys its credential cache per terminal (`tty_tickets`), so a child
spawned with `start_new_session` cannot read the cache
{func}`~meta_package_manager.sudo.prime_sudo` probed and warmed: it fails
with `sudo: interactive authentication is required` however fresh the
credentials are. Everything mpm does not escalate keeps the session
isolation that reaps wedged grandchildren.
"""
manager = FakeManager()
manager.sudo = True
with (
# Pinned, not detected: a Windows runner carries no `sudo` on `PATH`.
only_escalator("sudo"),
patch("meta_package_manager.execution.run_cli") as spawn,
):
spawn.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
manager.run(manager.build_cli("-c", "true", sudo=escalate))
assert spawn.call_args.kwargs["start_new_session"] is not escalate
[docs]
@pytest.mark.parametrize(
("internal_sudo", "resets", "operation", "warm", "detached"),
(
pytest.param(True, False, "install", True, False, id="spends-the-cache"),
pytest.param(True, False, "sync", True, False, id="sync-spends-it-too"),
pytest.param(True, False, "cleanup", True, True, id="never-escalating-call"),
pytest.param(True, False, "installed", True, True, id="read-only-call"),
pytest.param(True, False, "install", False, True, id="cold-cache"),
pytest.param(False, False, "install", True, True, id="no-internal-sudo"),
pytest.param(True, True, "sync", True, True, id="tool-drops-the-cache"),
),
)
def test_internal_escalator_spends_a_warm_cache_in_the_terminal_session(
monkeypatch, internal_sudo, resets, operation, warm, detached
):
"""An internal escalator call stays in the terminal session while cached.
While the credential cache is warm, in each operation where its tool may run `sudo`.
`sudo` keys a ticket on the terminal and on its session. Detached, the
tool's own `sudo` finds no ticket and no terminal to prompt on, so its
privileged step fails on the very cache meant to serve it.
Each control changes one condition and keeps the session isolation: an
operation in which the tool never escalates, a read-only call, a cold
cache, a manager that never escalates on its own, and a tool that drops
the cache, which has nothing to spend.
"""
assert not _SUDO_CACHE_WARM.is_set()
# The tool of this manager escalates in every mutating operation but one.
monkeypatch.setattr(
FakeManager, "internal_sudo_operations", _MUTATING_OPERATIONS - {"cleanup"}
)
manager = FakeManager()
manager.internal_sudo = internal_sudo
manager.resets_sudo_timestamp = resets
manager._active_operation = operation
if warm:
_SUDO_CACHE_WARM.set()
try:
with patch("meta_package_manager.execution.run_cli") as spawn:
spawn.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
manager.run_cli("-c", "true")
finally:
_SUDO_CACHE_WARM.clear()
assert spawn.call_args.kwargs["start_new_session"] is detached
[docs]
def test_run_forwards_a_forced_environment_across_run0():
"""The environment a call forces reaches the payload `run0` escalates.
run0 forks it off the service manager, which inherits nothing, so a
variable handed to the escalator's own process stops there. `run()` splices
the escalator's own forwarding argument instead, and does it after the
cooldown merge so that gate travels too. Measured on Arch with systemd 261:
`run0 --pipe --no-ask-password printenv LC_ALL` exits `1`, while the same
call carrying `--setenv=LC_ALL=C` prints `C`.
"""
manager = FakeManager()
manager.sudo = True
with (
only_escalator("run0"),
patch("meta_package_manager.execution.run_cli") as spawn,
):
spawn.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
manager.run(
manager.build_cli("-c", "true", sudo=True),
extra_env={"LC_ALL": "C"},
)
argv = spawn.call_args.args[0]
assert "--setenv=LC_ALL=C" in argv
# Ahead of the separator, so the payload's own flags stay shielded.
assert argv.index("--setenv=LC_ALL=C") < argv.index("--")
[docs]
def test_run_forwards_nothing_across_an_escalator_that_needs_no_help():
"""`sudo` resets the environment through a `sudoers` policy the host owns.
So mpm splices nothing of its own.
"""
manager = FakeManager()
manager.sudo = True
with (
only_escalator("sudo"),
patch("meta_package_manager.execution.run_cli") as spawn,
):
spawn.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
manager.run(
manager.build_cli("-c", "true", sudo=True),
extra_env={"LC_ALL": "C"},
)
assert not any(a.startswith("--setenv") for a in spawn.call_args.args[0])
[docs]
@pytest.mark.parametrize(
"args",
(
pytest.param(("install", "fake-pkg"), id="install"),
pytest.param(("upgrade", "--all"), id="upgrade-all"),
pytest.param(("upgrade", "fake-pkg"), id="upgrade-packages"),
pytest.param(("remove", "fake-pkg"), id="remove"),
pytest.param(("sync",), id="sync"),
pytest.param(("cleanup",), id="cleanup"),
),
)
def test_mutating_subcommands_prime_sudo(invoke, fake_pool, args):
"""Every mutating subcommand authenticates sudo up front.
So a future command cannot silently skip the priming.
"""
with patch("meta_package_manager.cli_maintenance.prime_sudo") as prime:
invoke("--dry-run", *args)
assert prime.called, f"`mpm {args[0]}` did not call prime_sudo"
[docs]
def test_restore_primes_sudo(invoke, fake_pool, tmp_path):
"""`restore` fans installs out too, so it also primes sudo up front."""
toml_file = tmp_path / "backup.toml"
toml_file.write_text("", encoding="UTF-8")
with patch("meta_package_manager.cli_snapshots.prime_sudo") as prime:
invoke("--dry-run", "restore", str(toml_file))
assert prime.called
[docs]
def test_read_subcommand_does_not_prime_sudo(invoke, fake_pool):
"""A read-only command never escalates, so it must not prompt for sudo.
Patched at the source module: the explore subcommands import no sudo
machinery at all, so any priming would have to reach it there.
"""
with patch("meta_package_manager.sudo.prime_sudo") as prime:
invoke("--dry-run", "installed")
assert not prime.called
[docs]
@pytest.mark.parametrize(
("args", "expected_operation"),
(
pytest.param(("remove", "fake-pkg-alpha"), "remove", id="remove"),
pytest.param(("upgrade", "fake-pkg-alpha"), "upgrade", id="upgrade-packages"),
),
)
def test_sourced_operation_stamps_active_operation(
invoke, fake_pool, monkeypatch, args, expected_operation
):
"""A sourced `remove`/`upgrade <packages>` runs under the real operation stamp.
Its mutating action carries the stamp.
Source discovery re-selects the managers with `installed`, which stamps
`_active_operation = "installed"` on the shared singletons. Each per-package
task must therefore re-stamp the mutating operation for the duration of its
own attempt (through `CLIExecutor.acting_as`), or the command would resolve
the read-only timeout and skip the escalator stall watchdog (both keyed on
`_active_operation`). This drives the real selection flow and observes the
stamp from inside the action itself: it fails if the dispatch stops stamping.
"""
# The sourced dispatch fetches the acting manager through `pool.get`; return the
# fake so a real task is built. `fake-pkg-alpha` is in the fake's installed set,
# so an untied spec resolves to it without touching a real binary.
monkeypatch.setattr(pool, "get", lambda manager_id: fake_pool)
# Cooldown gating is host-dependent; permit unconditionally so the task is built.
monkeypatch.setattr(
"meta_package_manager.cli_maintenance.cooldown_permits", lambda manager: True
)
recorded = {}
def record_operation(*args, **kwargs):
recorded["operation"] = fake_pool._active_operation
return ""
monkeypatch.setattr(fake_pool, expected_operation, record_operation)
invoke("--dry-run", *args)
assert recorded.get("operation") == expected_operation
[docs]
def test_untied_install_search_runs_under_read_only_stamp(
invoke, fake_pool, monkeypatch
):
"""The untied-install priority search runs under the `search` stamp, not `install`.
The search is read-only and can never escalate, so it must not resolve the mutating
timeout nor arm the internal-escalator stall watchdog (both keyed on
`_active_operation`). Without the stamp, a slow `brew search` of a cask would be
misread as a hidden password prompt.
"""
recorded = {}
def fake_search(*, extended, exact, query):
recorded["operation"] = fake_pool._active_operation
return iter(())
monkeypatch.setattr(fake_pool, "refiltered_search", fake_search)
# An untied name the fake does not install is unmatched, so it drops onto the
# sequential priority search that probes each manager with refiltered_search.
invoke("--dry-run", "install", "unmatched-name")
assert recorded["operation"] == Operations.search.name
[docs]
@pytest.mark.parametrize("operation", ("realpath", "is_dir"))
@pytest.mark.parametrize(
"error_code", (errno.EACCES, errno.EIO, errno.ENOENT, errno.ENOTDIR)
)
def test_search_all_cli_skips_inaccessible_directory(
tmp_path, monkeypatch, caplog, operation, error_code
):
"""An inaccessible PATH entry must not hide a later, usable CLI."""
blocked = tmp_path / "blocked"
usable = tmp_path / "usable"
blocked.mkdir()
usable.mkdir()
cli = usable / "discovery-probe"
cli.write_text("discovery fixture", encoding="UTF-8")
owner = os.path if operation == "realpath" else Path
original = getattr(owner, operation)
def probe(path, *args, **kwargs):
if path == blocked:
raise OSError(error_code, os.strerror(error_code), str(path))
return original(path, *args, **kwargs)
monkeypatch.setattr(owner, operation, probe)
monkeypatch.setattr("meta_package_manager.execution.is_any_windows", lambda: False)
with caplog.at_level(logging.DEBUG):
found = list(
CLIExecutor().search_all_cli(
(cli.name,),
env={"PATH": os.pathsep.join((str(blocked), str(usable)))},
)
)
assert found == [cli]
assert any(
str(blocked) in r.getMessage() and os.strerror(error_code) in r.getMessage()
for r in caplog.records
)
[docs]
@pytest.mark.parametrize("on_windows", (False, True))
@pytest.mark.parametrize(
"error_code", (errno.EACCES, errno.EIO, errno.ENOENT, errno.ENOTDIR)
)
def test_search_all_cli_skips_inaccessible_file(
tmp_path, monkeypatch, caplog, on_windows, error_code
):
"""Unreadable or vanished candidates must not abort the remaining search."""
blocked = tmp_path / "blocked"
usable = tmp_path / "usable"
blocked.mkdir()
usable.mkdir()
name = "discovery-probe.exe" if on_windows else "discovery-probe"
for directory in (blocked, usable):
(directory / name).write_text("discovery fixture", encoding="UTF-8")
operation = "lstat" if on_windows else "stat"
original = getattr(Path, operation)
def probe(path, *args, **kwargs):
if path == blocked / name:
raise OSError(error_code, os.strerror(error_code), str(path))
if on_windows and path == usable / name:
return SimpleNamespace(st_file_attributes=0)
return original(path, *args, **kwargs)
monkeypatch.setattr(Path, operation, probe)
monkeypatch.setattr(
"meta_package_manager.execution.is_any_windows", lambda: on_windows
)
monkeypatch.setenv("PATHEXT", ".EXE")
with caplog.at_level(logging.DEBUG):
found = list(
CLIExecutor().search_all_cli(
(name,),
env={"PATH": os.pathsep.join((str(blocked), str(usable)))},
)
)
assert found == [usable / name]
if error_code not in (errno.ENOENT, errno.ENOTDIR):
assert any(
str(blocked / name) in r.getMessage()
and os.strerror(error_code) in r.getMessage()
for r in caplog.records
)
[docs]
@pytest.mark.skipif(is_any_windows(), reason="A symlink needs privileges on Windows.")
@pytest.mark.parametrize("loop_at", ("search_path", "candidate"))
def test_search_all_cli_skips_symlink_loop(tmp_path, loop_at):
"""A symlink pointing to itself must not abort the remaining search."""
looping = tmp_path / "looping"
usable = tmp_path / "usable"
usable.mkdir()
cli = usable / "discovery-probe"
cli.write_text("discovery fixture", encoding="UTF-8")
if loop_at == "search_path":
looping.symlink_to(looping.name)
else:
looping.mkdir()
(looping / cli.name).symlink_to(cli.name)
found = list(
CLIExecutor().search_all_cli(
(cli.name,),
env={"PATH": os.pathsep.join((str(looping), str(usable)))},
)
)
assert found == [cli]
[docs]
def test_windows_pathext_tracks_cpython():
"""Our copy of CPython's default `PATHEXT` still matches the original.
`search_all_cli` mirrors {func}`shutil.which` on Windows by reading
`shutil._WIN_DEFAULT_PATHEXT`, which is private: a release may rename it,
drop it, or add a suffix to it, and only Windows would notice. The
attribute is defined on every platform, so this canary runs on every cell
of the test matrix, including the prerelease interpreter that sees the
change first, and turns a Windows-only detection failure into a named
failure everywhere.
On divergence: update {data}`~meta_package_manager.execution.WIN_DEFAULT_PATHEXT` to
the new value. If the attribute is gone for good, the fallback is already what the
code uses.
"""
cpython_value = getattr(shutil, "_WIN_DEFAULT_PATHEXT", None)
assert cpython_value is not None, (
"shutil._WIN_DEFAULT_PATHEXT is gone: mpm now relies on its own "
"WIN_DEFAULT_PATHEXT fallback, so confirm the value is still current."
)
assert cpython_value == WIN_DEFAULT_PATHEXT, (
f"CPython now defaults PATHEXT to {cpython_value!r}, but "
f"WIN_DEFAULT_PATHEXT still copies {WIN_DEFAULT_PATHEXT!r}."
)
[docs]
@pytest.mark.skipif(is_any_windows(), reason="`ENOEXEC` is POSIX-only.")
def test_a_file_that_is_not_a_program_raises_enoexec(tmp_path):
"""The premise {data}`~meta_package_manager.execution.UNRUNNABLE_ERRNOS` rests on.
`subprocess` execs directly rather than through a shell, so it never falls
back to interpreting a file the kernel refused. A wrong-architecture binary
and a text file carrying the executable bit both land here.
"""
not_a_program = tmp_path / "pnpm"
not_a_program.write_bytes(b"\x7fELF is how one starts, and this is not one")
not_a_program.chmod(0o755)
with pytest.raises(OSError) as caught:
subprocess.run((str(not_a_program),), capture_output=True, check=False)
assert caught.value.errno == errno.ENOEXEC
[docs]
@pytest.mark.parametrize(
"error_code",
tuple(sorted(UNRUNNABLE_ERRNOS)),
ids=lambda code: errno.errorcode[code],
)
def test_unrunnable_cli_degrades_instead_of_crashing(error_code, caplog):
"""A CLI that cannot be spawned marks its manager unavailable and yields "".
mpm probes every manager it can see, so one unrunnable binary anywhere on
`PATH` aborts every subcommand instead of skipping that single manager.
`ENOEXEC` is the one that reached CI: GitHub's `ubuntu-26.04-arm` image
carries an x86 `pnpm` at `/usr/local/bin/pnpm`.
"""
manager = FakeManager()
refusal = OSError(error_code, os.strerror(error_code), "/usr/local/bin/pnpm")
with (
patch("meta_package_manager.execution.run_cli", side_effect=refusal),
caplog.at_level(logging.DEBUG),
):
output = manager.run("--version")
assert output == ""
assert manager.executable is False
assert any("cannot be executed" in r.getMessage() for r in caplog.records)