"""CLI tests.
These cover the units that decide *what* the tool will do, which is where a
mistake is expensive: misclassifying a downgrade as an upgrade means offering to
apply something pacman will refuse, and forgetting to record a baseline means
every audit stays a first audit forever.
"""
from __future__ import annotations
import io
import json
import os
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from updater import aur, cli # noqa: E402
CLONE = Path.home() / ".cache/paru/clone"
def _code(fn) -> str:
"""A function's source with all whitespace removed.
These assertions are about what the code does, and the exact line breaks
are the formatter's business -- pinning them made every one of these tests
fail the first time the tree was formatted.
"""
return "".join(_code(fn).split())
def pkg(name: str, version: str, maintainer: str = "someone") -> aur.AurPackage:
return aur.AurPackage(name=name, version=version, maintainer=maintainer)
# --------------------------------------------------------------------------
# argument handling
# --------------------------------------------------------------------------
def _code(fn) -> str:
"""A function's source with all whitespace removed.
These assertions are about what the code does; where the line breaks fall
is the formatter's business. Stripping whitespace from both the source and
the snippet means a reformat cannot make them fail.
"""
import inspect
return _strip(inspect.getsource(fn))
def _strip(s: str) -> str:
return "".join(s.split())
def test_a_failed_package_is_recorded_as_failed_in_the_observer():
"""The watcher saw the process exit and nothing said otherwise, so it
filed the package as done: the stream carried `finished state=done` and
`failed` for the same package, and `done` reached the total with packages
missing."""
source = _code(cli._apply_aur_each)
assert _strip("observer.fail") in source, source[-800:]
def test_shared_flags_are_accepted_before_and_after_the_subcommand():
parser = cli.build_parser()
assert parser.parse_args(["--offline", "plan"]).offline is True
assert parser.parse_args(["plan", "--offline"]).offline is True
assert parser.parse_args(["plan", "--no-aur", "--json"]).no_aur is True
assert parser.parse_args(["plan", "--json"]).json is True
def test_a_flag_given_once_is_not_replaced_by_the_subcommand_default():
"""The subparser copy carries no default, or it would clobber the real one."""
args = cli.build_parser().parse_args(["--offline", "plan"])
assert getattr(args, "offline", False) is True
assert cli.options_from(args).offline is True
def test_plan_implies_dry_run_without_forcing_it_on_other_commands():
assert cli.options_from(cli.build_parser().parse_args(["plan"])).dry_run is True
assert cli.options_from(cli.build_parser().parse_args(["hygiene"])).dry_run is False
assert cli.options_from(cli.build_parser().parse_args([])).dry_run is False
def test_there_is_no_workers_flag_to_misuse():
"""paru builds serially, so dividing cores across workers would only slow it."""
import contextlib
with contextlib.redirect_stderr(io.StringIO()):
try:
cli.build_parser().parse_args(["--workers", "8"])
except SystemExit:
pass
else:
raise AssertionError("--workers is still offered but cannot do anything")
assert cli.options_from(cli.build_parser().parse_args([])).workers == 1
def test_hygiene_subcommand_parses_its_own_flags():
args = cli.build_parser().parse_args(["log", "--tail"])
assert args.command == "log" and args.tail is True
# --------------------------------------------------------------------------
# AUR classification: the trap that produced 25 phantom updates
# --------------------------------------------------------------------------
def test_vcs_packages_are_never_treated_as_version_upgrades():
"""The AUR records a -git pkgver at commit time, often older than local."""
installed = {"thing-git": "r900.abc-1"}
meta = {"thing-git": pkg("thing-git", "r600.def-1")}
upgrades, downgrades = cli._classify_aur(installed, meta)
assert upgrades == {} and downgrades == {}, "a VCS package was classified by version comparison"
def test_a_genuine_upgrade_is_an_upgrade():
upgrades, downgrades = cli._classify_aur({"a": "1.0-1"}, {"a": pkg("a", "2.0-1")})
assert upgrades == {"a": ("1.0-1", "2.0-1")}
assert downgrades == {}
def test_an_older_aur_version_is_a_downgrade_not_an_upgrade():
"""paru -Sua will not apply this, so calling it an update would be a lie."""
upgrades, downgrades = cli._classify_aur({"a": "2.0-1"}, {"a": pkg("a", "1.0-1")})
assert upgrades == {}
assert downgrades == {"a": ("2.0-1", "1.0-1")}
def test_an_unchanged_version_is_neither():
upgrades, downgrades = cli._classify_aur({"a": "1.0-1"}, {"a": pkg("a", "1.0-1")})
assert upgrades == {} and downgrades == {}
def test_packages_not_installed_are_ignored():
upgrades, downgrades = cli._classify_aur({}, {"a": pkg("a", "9.9-1")})
assert upgrades == {} and downgrades == {}
def test_classification_handles_epoch_and_pkgrel_ordering():
# an epoch bump outranks a numerically smaller pkgver
upgrades, _ = cli._classify_aur({"a": "1.0-1"}, {"a": pkg("a", "1:0.9-1")})
assert "a" in upgrades
# a pkgrel bump is a real upgrade
upgrades, _ = cli._classify_aur({"a": "1.0-1"}, {"a": pkg("a", "1.0-2")})
assert "a" in upgrades
# but a lower pkgrel is not
upgrades, downgrades = cli._classify_aur({"a": "1.0-2"}, {"a": pkg("a", "1.0-1")})
assert not upgrades and "a" in downgrades
def test_the_devel_list_only_contains_devel_packages():
"""`paru -Qua` lists every AUR upgrade, not only the devel ones.
Without a filter, each ordinary AUR update appeared a second time as a
"devel rebuild" -- the same package twice in the same section.
"""
foreign = ["helix-git", "android-apktool-bin", "ttf-iosevka", "solaar-git"]
paru_said = {
"helix-git": ("r1", "r2"),
"android-apktool-bin": ("1-1", "2-1"),
"ttf-iosevka": ("34.8.1-1", "34.9.0-1"),
"solaar-git": ("a", "b"),
}
rows = cli._vcs({}, foreign, paru_said)
assert [r.name for r in rows] == ["helix-git", "solaar-git"], rows
def test_the_devel_list_never_duplicates_the_upgrade_list():
"""A package must not be offered as both an upgrade and a devel rebuild."""
installed = {"helix-git": "r1", "demo-bin": "1-1"}
meta = {"helix-git": pkg("helix-git", "r2"), "demo-bin": pkg("demo-bin", "2-1")}
upgrades, _ = cli._classify_aur(installed, meta)
devel = cli._vcs(
installed,
["helix-git", "demo-bin"],
{"helix-git": ("r1", "r2"), "demo-bin": ("1-1", "2-1")},
)
assert "helix-git" not in upgrades, "a VCS package must not be a plain upgrade"
assert [r.name for r in devel] == ["helix-git"], devel
assert not set(upgrades) & {r.name for r in devel}
def test_no_devel_check_means_no_devel_rows():
"""Not knowing must not be reported as "nothing to rebuild"."""
assert cli._vcs({}, ["a-git"], None) == []
assert cli._vcs({}, ["a-git"], {}) == []
def test_a_left_the_aur_warning_names_its_packages():
"""Free text cannot be scoped to a selection; a list can."""
note = cli._package_warning(["alpha", "beta"], known=set())
assert note.packages == ["alpha", "beta"], note.packages
assert note.title and note.headline and note.flagged == 2
def test_the_warning_row_is_one_row_however_many_kinds_there_are():
"""A row each turned "look at this" into a column of near-identical rows."""
run = cli.Run(options=cli.Options())
run.warning_notes = [
cli.WarningNote(label="warning", title="a", headline="first"),
cli.WarningNote(label="warning", title="b", headline="second"),
]
rows = [c for c in cli._checks(run) if c.label == "warning"]
assert len(rows) == 1, [c.detail for c in rows]
assert "2 kinds" in rows[0].detail, rows[0].detail
def test_a_failed_phase_reports_why():
"""`exited 1` is not an error message."""
import tempfile
from pathlib import Path as _Path
with tempfile.TemporaryDirectory() as td:
log = _Path(td) / "pacman.log"
log.write_text("error: failed to commit transaction (conflicting files)\n")
got = cli._failure_reason(log, "the repository phase failed")
assert "conflicting files" in got, got
assert got.startswith("the repository phase failed"), got
def test_the_sudo_failure_gets_its_own_explanation():
"""It is the failure that looks most like nothing happening at all."""
import tempfile
from pathlib import Path as _Path
with tempfile.TemporaryDirectory() as td:
log = _Path(td) / "paru.log"
log.write_text(
"sudo: a terminal is required to read the password; either use "
"the -S option\nsudo: a password is required\n"
)
got = cli._failure_reason(log, "the aur phase failed")
assert "sudo -v" in got, got
assert "terminal" in got, got
def test_a_missing_log_still_reports_something():
from pathlib import Path as _Path
got = cli._failure_reason(_Path("/nonexistent/nope.log"), "the aur phase failed")
assert got == "the aur phase failed"
def test_an_empty_log_does_not_produce_a_dangling_dash():
import tempfile
from pathlib import Path as _Path
with tempfile.TemporaryDirectory() as td:
log = _Path(td) / "pacman.log"
log.write_text("\n\n")
got = cli._failure_reason(log, "the repository phase failed")
assert got == "the repository phase failed", got
assert not got.endswith("—"), got
def test_the_apply_plans_from_the_screen_s_own_source():
"""Both must read the same databases, or the apply can silently do less."""
source = _code(cli.cmd_apply)
assert _strip("_plan_handle(run)") in source, source[:400]
assert _strip("with repo.connect() as handle") not in source
def test_the_apply_checks_sudo_before_running_anything():
"""A run that could never authenticate should not stream a build screen
and then report an exit code."""
source = _code(cli.cmd_apply)
assert _strip("live.can_sudo()") in source, source[:400]
# and the check must come before the first phase is dispatched
assert source.index(_strip("live.can_sudo()")) < source.index(_strip('if "repo" in phases')), (
"the sudo check runs after the repository phase"
)
def test_a_package_conflict_is_explained_not_just_reported():
"""A conflict is a question, not a fault.
pacman refuses to replace a package under --noconfirm because replacing one
is exactly what it should ask about. The useful part of the log -- which
packages conflict -- sits above the error line a naive reader would report.
"""
import tempfile
from pathlib import Path as _Path
with tempfile.TemporaryDirectory() as td:
log = _Path(td) / "paru.log"
log.write_text(
":: Resolving dependencies...\n"
":: Calculating conflicts...\n"
":: Conflicts found:\n"
" scenefx: scenefx0.5 (scenefx) \n\n"
":: Conflicting packages will have to be confirmed manually\n"
"error: can not install conflicting packages with --noconfirm\n"
)
got = cli._failure_reason(log, "the aur phase failed")
assert "scenefx" in got, got
assert "scenefx0.5" in got, got
# the fact that matters must survive truncation on a narrow terminal
assert got.startswith("conflicting packages ("), got
assert "nothing was installed" in got, got
def test_a_conflict_block_that_is_not_there_is_not_invented():
import tempfile
from pathlib import Path as _Path
with tempfile.TemporaryDirectory() as td:
log = _Path(td) / "paru.log"
log.write_text(":: Resolving dependencies...\nerror: something else\n")
got = cli._failure_reason(log, "the aur phase failed")
assert "conflict" not in got.lower(), got
assert "something else" in got, got
def test_a_conflict_with_no_pairs_falls_back():
import tempfile
from pathlib import Path as _Path
with tempfile.TemporaryDirectory() as td:
log = _Path(td) / "paru.log"
log.write_text(":: Conflicts found:\n\nerror: can not install\n")
got = cli._failure_reason(log, "the aur phase failed")
assert got, got
class _FakeRunner:
"""Succeeds for everything except the names it was told to fail."""
def __init__(self, fail):
self.fail = set(fail)
self.ran = []
self.logs = []
def run(self, argv, phase, log, total=0, since=None):
name = argv[-1]
self.ran.append(name)
self.logs.append(log)
from pathlib import Path as _P
_P(log).parent.mkdir(parents=True, exist_ok=True)
if name in self.fail:
_P(log).write_text(
":: Resolving dependencies...\n"
"error: could not find all required packages:\n"
f" {name} (target)\n"
)
return 1
_P(log).write_text(":: installed\n")
return 0
def test_the_other_packages_still_install_when_one_fails():
"""paru resolves its whole target list before building any of it.
Naming every package at once therefore made one unresolvable target fail
the command outright, installing nothing at all. The user asked for the
rest to go through and the problem to be reported at the end.
"""
import tempfile
from pathlib import Path as _P
runner = _FakeRunner({"mangowm-git"})
ordered = ["a-git", "mangowm-git", "b-git", "c-git"]
failures, remaining, _skipped = cli._apply_aur_each(runner, ordered, _P(tempfile.mkdtemp()))
assert runner.ran == ordered, runner.ran
assert [name for name, _ in failures] == ["mangowm-git"], failures
assert remaining == [], remaining
# each run gets its own log, so the failure can be read from the run that
# produced it rather than from a shared file
assert len(set(runner.logs)) == len(ordered), runner.logs
def test_a_clean_aur_run_reports_no_failures():
import tempfile
from pathlib import Path as _P
runner = _FakeRunner(set())
failures, remaining, _skipped = cli._apply_aur_each(runner, ["a", "b"], _P(tempfile.mkdtemp()))
assert failures == [] and remaining == []
assert runner.ran == ["a", "b"]
def test_each_failure_is_reported_as_it_happens():
"""A long run should show progress through the failures, not a wall at
the end."""
import tempfile
from pathlib import Path as _P
seen = []
runner = _FakeRunner({"a", "c"})
cli._apply_aur_each(runner, ["a", "b", "c"], _P(tempfile.mkdtemp()), report=seen.append)
assert len(seen) == 2, seen
# named once each, so a reader can tell which failure is which
assert all(m.startswith(("a:", "c:")) for m in seen), seen
assert all("a failed" not in m for m in seen), seen
def test_an_aborted_run_stops_instead_of_starting_the_next_package():
"""Ctrl+C used to unwind the running child and then carry on.
The interrupt is caught around the child so that paru can be shut down
cleanly, which meant the loop saw a phase that had merely finished and
started the next package -- so cancelling one build walked through all of
the remaining ones.
"""
import tempfile
from pathlib import Path as _P
class AbortingRunner:
def __init__(self):
self.ran = []
self.interrupted = False
def run(self, argv, phase, log, total=0, since=None):
name = argv[-1]
self.ran.append(name)
if name == "b-git":
# what the runner does when the user interrupts mid-child
self.interrupted = True
return 130
return 0
runner = AbortingRunner()
_failures, remaining, _skipped = cli._apply_aur_each(
runner, ["a-git", "b-git", "c-git", "d-git"], _P(tempfile.mkdtemp())
)
assert runner.ran == ["a-git", "b-git"], runner.ran
# and what is left is reported, so it can be offered as a resume
assert remaining == ["c-git", "d-git"], remaining
def test_a_run_that_was_not_interrupted_still_does_every_package():
import tempfile
from pathlib import Path as _P
runner = _FakeRunner(set())
_failures, remaining, _skipped = cli._apply_aur_each(
runner, ["a", "b", "c"], _P(tempfile.mkdtemp())
)
assert runner.ran == ["a", "b", "c"]
assert remaining == []
def test_stopping_offers_to_resume_what_is_left():
"""Cancelling should not mean starting over.
The packages never attempted are known, so they can be offered back rather
than making someone re-run the whole thing.
"""
source = _code(cli.cmd_apply)
assert _strip("unfinished = [name for name, _ in failures] + remaining") in source, source[
:1400
]
assert _strip("resume in a terminal") in source, source[:1400]
def test_the_summary_does_not_restate_the_count_three_ways():
"""It said how many failed, how many installed, and how to fix it -- the
middle one being something the reader already knows."""
source = _code(cli.cmd_apply)
assert _strip("The other ") not in source, source[:1800]
assert _strip("Press R to finish them") not in source, source[:1800]
def test_a_signal_exit_is_reported_as_a_stop_not_a_mystery():
"""makepkg's last line when interrupted is "==> Cleaning up…", which
explains nothing; quoting it made the message look like a puzzle."""
import tempfile
from pathlib import Path as _P
with tempfile.TemporaryDirectory() as td:
log = _P(td) / "p.log"
log.write_text("==> Starting build()...\n==> ERROR: Aborted by user!\n==> Cleaning up...\n")
assert cli._failure_reason(log, "failed (exit 130)", 130) == "stopped before it finished"
def test_a_progress_line_is_not_mistaken_for_a_failure():
import tempfile
from pathlib import Path as _P
with tempfile.TemporaryDirectory() as td:
log = _P(td) / "p.log"
log.write_text("==> Starting build()...\n==> Cleaning up...\n")
# nothing reads like an error, so do not invent one from the last line
assert cli._failure_reason(log, "failed (exit 1)", 1) == "failed (exit 1)"
def test_the_actual_error_line_is_preferred_over_the_last_line():
import tempfile
from pathlib import Path as _P
with tempfile.TemporaryDirectory() as td:
log = _P(td) / "p.log"
log.write_text(
"==> Starting build()...\n"
"error: could not compile `foo`\n"
"==> ERROR: A failure occurred in build().\n"
"==> Cleaning up...\n"
)
got = cli._failure_reason(log, "failed (exit 4)", 4)
assert "could not compile" in got, got
assert "Cleaning up" not in got, got
def test_only_devel_packages_that_moved_are_rebuilt():
"""A -git package is rebuilt whenever it is named.
--needed does not skip one the way it skips a released version, because
what is installed is the version generated at the last build. Naming every
foreign -git package therefore rebuilt all of them on every run, including
the ones finished minutes earlier.
"""
source = _code(cli.cmd_apply)
assert _strip("devel_upgrades()") in source, source[:1600]
assert _strip("n in devel]") in source, source[:1600]
# the old shape took the whole foreign set
assert _strip("vcs = [n for n in foreign if _is_vcs(n) and n not in gone]") not in source
def test_the_command_and_the_screen_agree_on_devel_packages():
"""The screen asked paru which ones moved; the command did not."""
from updater import cli as cli_mod
display = _code(cli_mod._vcs)
apply_source = _code(cli_mod.cmd_apply)
assert _strip("upgrades") in display and _strip("devel") in apply_source.lower()
# both filter the AUR set down to the VCS names paru reported
assert _strip("_is_vcs(name)") in display, display
assert _strip("_is_vcs(n)") in apply_source, apply_source[:1600]
def test_a_failed_devel_check_offers_nothing_rather_than_everything():
"""Asked and could not be told: rebuild nothing, not all of it."""
source = _code(cli.cmd_apply)
assert _strip("devel rebuilds not offered") in source, source[:1800]
def test_a_hold_naming_something_outside_the_plan_does_not_skip_the_repo_phase():
"""Holds cover repository and AUR packages alike.
The gate compared set sizes, so a hold for an AUR package made the counts
line up and the repository phase was skipped -- the soloed package was
never installed, silently.
"""
phases, targets, total = cli._apply_phases(
{"mesa", "linux", "git"},
{"mangowm-git"},
{"linux", "git", "mangowm-git"},
set(),
flatpaks=False,
)
assert phases == ["repo"], phases # mesa still needs installing
assert targets == [], targets # mangowm-git was held
assert total == 1, total # mesa alone
def test_a_held_aur_target_is_not_built():
phases, targets, total = cli._apply_phases(
{"mesa"}, {"foo-git"}, {"mesa", "foo-git"}, set(), flatpaks=False
)
assert phases == [], phases
assert targets == [], targets
assert total == 0, total
def test_holding_every_plan_update_skips_the_repo_phase():
phases, targets, total = cli._apply_phases(
{"a", "b"}, {"x-git"}, {"a", "b"}, set(), flatpaks=False
)
assert phases == ["aur"], phases
assert targets == ["x-git"], targets
assert total == 1, total
def test_nothing_held_runs_everything():
phases, targets, total = cli._apply_phases({"a", "b"}, {"x-git"}, set(), set(), flatpaks=True)
assert phases == ["repo", "aur", "flatpak"], phases
assert targets == ["x-git"], targets
assert total == 3, total
def test_a_skip_and_a_hold_both_remove_a_target():
_phases, targets, total = cli._apply_phases(
set(), {"a-git", "b-git", "c-git"}, {"b-git"}, {"c-git"}, flatpaks=False
)
assert targets == ["a-git"], targets
assert total == 1, total
def test_a_signal_exit_reads_as_a_stop_not_a_compiler_error():
"""subprocess reports a signal as a negative code, a shell as 128+n."""
import tempfile
from pathlib import Path as _P
with tempfile.TemporaryDirectory() as td:
log = _P(td) / "p.log"
log.write_text("error: could not compile `foo`\n")
assert cli._failure_reason(log, "failed (exit -2)", -2) == "stopped before it finished"
assert cli._failure_reason(log, "failed (exit 130)", 130) == "stopped before it finished"
def test_a_failed_aur_run_offers_a_way_to_finish_it():
"""An unattended run that hits a question is a dead end without this."""
source = _code(cli.cmd_apply)
assert _strip('emit("recovery"') in source, source[:400]
assert _strip("aur_command_interactive") in source, source[:400]
# offered for the packages that failed, not for the whole target list
assert _strip("[name for name, _ in failures]") in source, source[:900]
def test_the_aur_phase_is_not_reported_twice():
"""Per-package errors are emitted with each package's own log."""
source = _code(cli.cmd_apply)
assert _strip('if phase == "aur":\n continue') in source, source[-900:]
def test_a_partial_aur_success_still_records_baselines(monkeypatch=None):
"""One failed package must not erase the baselines of the rest.
Installing each package as it is built makes a partial success normal, and
gating the baseline on a zero exit threw away the record for every package
that did install -- so the audit kept reporting "no baseline" for work that
had actually happened.
"""
source = _code(cli.cmd_apply)
assert _strip("complete=not failures") in source, source[:400]
recorder = _code(cli._record_aur_baselines)
assert _strip("before.get(name) != after.get(name)") in recorder, recorder
assert _strip("only the ones") in recorder or _strip("actually moved") in recorder
def test_baselines_are_skipped_for_packages_that_did_not_move(monkeypatch=None):
"""A failed build leaves the clone at upstream's state, which was never
installed; recording it would claim otherwise."""
from updater import changes as changes_mod
recorded = []
class FakeState:
@staticmethod
def load():
return FakeState()
def record(self, name, **kwargs):
recorded.append(name)
def save(self):
pass
from pathlib import Path as _P
class FakeRun:
warnings: list[str] = []
class options:
clone_dir = _P("/nonexistent")
original_state = cli.aur.State
original_head = changes_mod.head
original_installed = cli._installed_versions
try:
cli.aur.State = FakeState
changes_mod.head = lambda path: type("H", (), {"sha": "abc"})()
# alpha moved, beta did not
cli._installed_versions = lambda: {"alpha": "2", "beta": "1"}
cli._record_aur_baselines(
FakeRun, ["alpha", "beta"], {}, {}, before={"alpha": "1", "beta": "1"}, complete=False
)
finally:
cli.aur.State = original_state
changes_mod.head = original_head
cli._installed_versions = original_installed
assert recorded == ["alpha"], recorded
def test_vcs_suffix_list_covers_the_common_schemes():
for name in ("x-git", "x-svn", "x-hg", "x-cvs", "x-bzr", "x-darcs", "x-nightly"):
assert cli._is_vcs(name), name
for name in ("x", "x-bin", "git-thing", "x-github"):
assert not cli._is_vcs(name), name
# --------------------------------------------------------------------------
# baseline recording: what makes audits improve over time
# --------------------------------------------------------------------------
def test_record_baselines_stores_the_built_commit_and_maintainer():
if not (CLONE / "python-tree-sitter-bash").is_dir():
return
with tempfile.TemporaryDirectory() as td:
previous = os.environ.get("XDG_STATE_HOME")
os.environ["XDG_STATE_HOME"] = td
try:
run = cli.Run(options=cli.Options(clone_dir=CLONE))
run.updates = {"python-tree-sitter-bash": ("0.25.1-2", "0.25.2-1")}
run.aur_meta = {
"python-tree-sitter-bash": pkg("python-tree-sitter-bash", "0.25.2-1", "kuglee")
}
recorded = cli.record_baselines(run)
assert recorded == 1
state = aur.State.load()
assert state.previous_maintainer("python-tree-sitter-bash") == "kuglee"
commit = state.previous_commit("python-tree-sitter-bash")
assert commit, "no commit was recorded"
assert len(commit) >= 7
finally:
if previous is None:
os.environ.pop("XDG_STATE_HOME", None)
else:
os.environ["XDG_STATE_HOME"] = previous
def test_record_baselines_is_a_no_op_with_nothing_to_record():
with tempfile.TemporaryDirectory() as td:
previous = os.environ.get("XDG_STATE_HOME")
os.environ["XDG_STATE_HOME"] = td
try:
run = cli.Run(options=cli.Options())
assert cli.record_baselines(run) == 0
assert not (Path(td) / "updater" / "aur-state.json").exists()
finally:
if previous is None:
os.environ.pop("XDG_STATE_HOME", None)
else:
os.environ["XDG_STATE_HOME"] = previous
# --------------------------------------------------------------------------
# the plan payload
# --------------------------------------------------------------------------
def test_plan_payload_is_json_serialisable_and_complete():
run = cli.Run(options=cli.Options())
run.plan.updates = []
payload = cli._plan_payload(run, cli.Review())
text = json.dumps(payload)
assert json.loads(text) == payload
for key in ("repo", "aur", "vcs", "checks", "flatpak", "vulnerabilities", "hygiene", "audits"):
assert key in payload, key
assert payload["hygiene"]["reboot"] is None
def test_the_plan_step_names_the_command_that_actually_runs():
"""A plan that says -Su while the code runs -Syu is a lie about the plan."""
from updater.repo import Update
run = cli.Run(options=cli.Options())
run.plan.updates = [Update("demo", "1.0-1", "2.0-1", "extra", 1024, 2048, 4096)]
step = next(s for s in cli._steps(run) if "pacman" in s.command)
captured: dict = {}
class Stop(Exception):
pass
class FakeRunner:
def run(self, argv, log):
captured["argv"] = argv
raise Stop
saved = cli._runner
cli._runner = lambda *a, **k: FakeRunner()
try:
cli.apply_repo(run)
except Stop:
pass
finally:
cli._runner = saved
argv = captured["argv"]
# argv is ["sudo", "pacman", "-Syu", "--noconfirm"]; the plan must name the
# same binary and flag, or the plan describes a different command
assert step.command.split() == argv[1:3], (step.command, argv)
def test_plan_steps_do_not_promise_to_apply_a_downgrade():
run = cli.Run(options=cli.Options())
run.downgrades = {"a": ("2.0-1", "1.0-1")}
steps = cli._steps(run)
commands = [s.command for s in steps]
assert "paru -Sua" not in commands, "a downgrade was offered as an update"
assert any("held back" in c for c in commands)
detail = next(s.detail for s in steps if s.command == "(held back)")
assert "not applied" in detail
def test_plan_steps_mention_a_reboot_only_when_recommended():
run = cli.Run(options=cli.Options())
assert "reboot" not in [s.command for s in cli._steps(run)]
run.hygiene.reboot = "running kernel 6.16.8 is no longer installed"
assert "reboot" in [s.command for s in cli._steps(run)]
# --------------------------------------------------------------------------
# checks
# --------------------------------------------------------------------------
def test_checks_report_an_offline_run_honestly():
run = cli.Run(options=cli.Options(offline=True))
labels = {c.label: c for c in cli._checks(run)}
assert "network" in labels
assert "offline" in labels["network"].detail
def test_checks_flag_unmerged_config_files():
from updater.hygiene import Drift
run = cli.Run(options=cli.Options())
run.hygiene.drift = Drift(pacnew=[Path("/etc/x.conf.pacnew")])
pacnew = next(c for c in cli._checks(run) if c.label == "pacnew")
assert pacnew.ok is False
assert "1 unresolved" in pacnew.detail
def test_checks_surface_warnings_rather_than_swallowing_them():
run = cli.Run(options=cli.Options())
run.warnings.append("6 packages no longer in the AUR")
warnings = [c for c in cli._checks(run) if c.label == "warning"]
assert warnings and "no longer in the AUR" in warnings[0].detail
def test_checks_never_raise_on_an_empty_run():
run = cli.Run(options=cli.Options())
assert isinstance(cli._checks(run), list)
assert isinstance(cli._steps(run), list)
# --------------------------------------------------------------------------
# the safety property: phase ordering and gating
# --------------------------------------------------------------------------
class _FakeResult:
def __init__(self, ok: bool, code: int = 0):
self.returncode = code
self.elapsed = 1.0
self.log_path = Path("/tmp/fake.log")
self.feed: list = []
self.interrupted = False
self.ok = ok
class _Harness:
"""Drive main() with the system-touching parts replaced."""
def __init__(self, repo_ok: bool = True, aur_ok: bool = True, flatpak_ok: bool = True):
self.calls: list[str] = []
self.saved: dict = {}
self.repo_ok, self.aur_ok, self.flatpak_ok = repo_ok, aur_ok, flatpak_ok
def __enter__(self):
from updater.render import RepoRow, Review
def keep(name, value):
self.saved[name] = getattr(cli, name)
setattr(cli, name, value)
# keep the harness quiet: main() builds its own Console
from rich.console import Console as _Console
keep(
"Console", lambda *a, **k: _Console(file=io.StringIO(), width=100, force_terminal=False)
)
keep(
"gather",
lambda run: Review(
repo=[RepoRow("U", "demo", "1.0-1", "2.0-1", "1 KiB")], repo_total=1
),
)
keep("confirm", lambda run, prompt: True)
keep("emit_plan", lambda run, review: None)
keep(
"write_summary",
lambda run, results: self.calls.append("summary") or Path("/tmp/summary.txt"),
)
keep("record_baselines", lambda run: self.calls.append("baselines") or 0)
def repo_phase(run):
self.calls.append("repo")
return _FakeResult(self.repo_ok, 0 if self.repo_ok else 1)
def aur_phase(run):
self.calls.append("aur")
return _FakeResult(self.aur_ok, 0 if self.aur_ok else 1)
def flatpak_phase(run):
self.calls.append("flatpak")
return _FakeResult(self.flatpak_ok, 0 if self.flatpak_ok else 1)
keep("apply_repo", repo_phase)
keep("apply_aur", aur_phase)
keep("apply_flatpak", flatpak_phase)
return self
def __exit__(self, *exc):
for name, value in self.saved.items():
setattr(cli, name, value)
return False
def test_a_failed_repo_phase_stops_before_any_aur_build():
"""Building AUR packages against a half-upgraded system breaks the tree."""
with _Harness(repo_ok=False) as h:
code = cli.main(["--yes", "--no-security", "--offline"])
assert "repo" in h.calls
assert "aur" not in h.calls, f"AUR built anyway: {h.calls}"
assert "flatpak" not in h.calls
assert code == cli.EXIT_FAILED
def test_a_successful_repo_phase_is_followed_by_aur_then_flatpak():
with _Harness() as h:
code = cli.main(["--yes", "--no-security", "--offline"])
# relative order is what matters; baseline recording legitimately sits
# between the AUR phase and the flatpak phase
assert h.calls.index("repo") < h.calls.index("aur") < h.calls.index("flatpak"), h.calls
assert h.calls.index("baselines") < h.calls.index("flatpak")
assert code == cli.EXIT_OK
def test_baselines_are_recorded_only_after_a_successful_aur_phase():
with _Harness(aur_ok=True) as h:
cli.main(["--yes", "--no-security", "--offline"])
assert "baselines" in h.calls
with _Harness(aur_ok=False) as h2:
code = cli.main(["--yes", "--no-security", "--offline"])
assert "baselines" not in h2.calls, "recorded a baseline for a failed build"
assert code == cli.EXIT_FAILED
def test_a_failed_flatpak_phase_does_not_prevent_the_summary():
with _Harness(flatpak_ok=False) as h:
code = cli.main(["--yes", "--no-security", "--offline"])
assert "summary" in h.calls
assert code == cli.EXIT_FAILED
def test_declining_the_plan_stops_everything():
with _Harness() as h:
cli.confirm = lambda run, prompt: False
code = cli.main(["--no-security", "--offline"])
assert h.calls == [], f"something ran after a decline: {h.calls}"
assert code == cli.EXIT_DECLINED
def test_dry_run_never_reaches_the_apply_phase():
with _Harness() as h:
code = cli.main(["--yes", "--dry-run", "--no-security", "--offline"])
assert h.calls == [], f"dry run applied something: {h.calls}"
assert code == cli.EXIT_OK
def main() -> int:
tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)]
failed = 0
for name, fn in tests:
try:
fn()
except Exception as exc: # noqa: BLE001
failed += 1
print(f"FAIL {name}\n {type(exc).__name__}: {exc}")
else:
print(f"ok {name}")
print(f"\n{len(tests) - failed}/{len(tests)} passed")
return 1 if failed else 0
# --------------------------------------------------------------------------
# the machine-readable contract
# --------------------------------------------------------------------------
def _run_json(argv: list[str], env: dict[str, str]) -> str:
"""Run the CLI in-process with a modified environment, capturing stdout."""
import contextlib
saved = {k: os.environ.get(k) for k in env}
os.environ.update(env)
out = io.StringIO()
try:
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(io.StringIO()):
cli.main(argv)
finally:
for key, value in saved.items():
if value is None:
os.environ.pop(key, None)
else:
os.environ[key] = value
return out.getvalue()
def test_json_output_is_never_colourised():
"""Rich colourises print_json and honours FORCE_COLOR.
Using it put ANSI escapes inside the JSON, so on any machine with colour
forced on the front end could not parse a single byte. Machine-readable
output has to be byte-exact.
"""
for env in (
{"FORCE_COLOR": "3", "TERM": "xterm-256color"},
{"FORCE_COLOR": "1"},
{"CLICOLOR_FORCE": "1"},
{"NO_COLOR": "1"},
{},
):
out = _run_json(
["plan", "--offline", "--json", "--no-security", "--no-hygiene", "--no-flatpak"], env
)
assert "\x1b" not in out, f"ANSI escape in JSON with env {env}: {out[:80]!r}"
parsed = json.loads(out) # must be valid on its own
assert isinstance(parsed, dict)
def test_json_output_parses_under_a_hostile_environment():
for env in (
{"TERM": "dumb"},
{"TERM": "xterm-256color", "FORCE_COLOR": "3"},
{"COLUMNS": "40"},
{"LANG": "C"},
):
out = _run_json(
["plan", "--offline", "--json", "--no-security", "--no-hygiene", "--no-flatpak"], env
)
json.loads(out)
def test_the_security_and_advisory_json_paths_are_plain_too():
"""Every subcommand that can emit JSON must go through the same writer."""
src = _code(cli)
# the call form, not the mention in _print_json's own docstring
assert _strip("console.print_json(") not in src, "console.print_json reintroduces the ANSI leak"
if __name__ == "__main__":
raise SystemExit(main())
# --------------------------------------------------------------------------
# consent
# --------------------------------------------------------------------------
def _run_with_phases(argv: list[str]) -> tuple[int, str]:
"""Run the CLI with the repo plan and AUR classifier stubbed to yield work.
Consent is what is under test, so the phases must be non-empty: with nothing
to do the command correctly returns before it ever asks.
"""
import contextlib
from updater import flatpak as flatpak_mod
from updater import repo as repo_mod
stub_plan = repo_mod.Plan(
updates=[
repo_mod.Update(
name="linux",
old="6.15-1",
new="6.16-1",
repo="core",
download=1,
installed_old=1,
installed_new=1,
desc="the kernel",
)
],
ignored=[],
)
saved = {
"plan": repo_mod.plan,
"classify": cli._classify_aur,
"flatpak": cli.flatpak.collect,
}
repo_mod.plan = lambda handle, *a, **k: stub_plan
cli._classify_aur = lambda installed, meta: ({"demo-bin": ("1-1", "2-1")}, {})
cli.flatpak.collect = lambda **_: flatpak_mod.FlatpakState(
available=True, reason="", remotes=[], installed=[], updates=[]
)
out = io.StringIO()
try:
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(io.StringIO()):
code = cli.main(argv)
finally:
repo_mod.plan = saved["plan"]
cli._classify_aur = saved["classify"]
cli.flatpak.collect = saved["flatpak"]
return code, out.getvalue()
def test_json_stream_without_yes_is_refused_rather_than_assumed():
"""A stream consumer owns its own dialog, so asking for the stream is not consent."""
code, out = _run_with_phases(["apply", "--json-stream", "--offline"])
events = [json.loads(l) for l in out.splitlines() if l.strip()]
assert events, "no protocol output at all"
assert code == cli.EXIT_DECLINED, f"exit {code}: {events}"
assert events[-1]["event"] == "done"
assert events[-1]["returncode"] == cli.EXIT_DECLINED
assert any(e["event"] == "error" and "--yes" in e.get("message", "") for e in events), events
def test_apply_never_runs_a_phase_without_consent():
"""The whole point: no phase may run when consent was not given."""
code, out = _run_with_phases(["apply", "--json-stream", "--offline"])
events = [json.loads(l) for l in out.splitlines() if l.strip()]
assert code != cli.EXIT_OK
assert not [e for e in events if e["event"] == "phase_start"], "a phase started without consent"
assert not [e for e in events if e["event"] == "phase_end"], "a phase ran without consent"
def test_an_unexpected_failure_is_not_a_traceback():
"""The top of the program owns turning a failure into an exit code.
A traceback tells somebody running an update nothing about what to do next.
"""
import contextlib
import io
from updater import cli as cli_mod
original = cli_mod.cmd_checks
def boom(run):
raise RuntimeError("something nobody anticipated")
cli_mod.cmd_checks = boom
err = io.StringIO()
try:
with contextlib.redirect_stderr(err):
code = cli_mod.main(["checks"])
finally:
cli_mod.cmd_checks = original
assert code == cli_mod.EXIT_FAILED, code
text = err.getvalue()
assert "RuntimeError" in text and "something nobody anticipated" in text, text
assert "UPDATER_DEBUG" in text, text
def test_a_failure_re_raises_for_debugging(monkeypatch=None):
"""UPDATER_DEBUG keeps the traceback, because then it is what you want."""
from updater import cli as cli_mod
source = _code(cli_mod.main)
assert _strip('os.environ.get("UPDATER_DEBUG")') in source, source
assert _strip("raise") in source, source
def test_an_audit_blocked_package_is_actually_skipped():
"""The plan and the prompt both promise it; the run has to do it.
The plan screen said "will be skipped", the prompt said "skipping N flagged
packages", and the skip set was built from --skip and IgnorePkg alone. A
package the audit refused was built with the reader told otherwise.
"""
source = _code(cli.cmd_apply)
assert _strip("set(run.blocked)") in source, source[:2000]
assert _strip("flagged") in source, source[:2000]
def test_the_checks_refresh_declares_what_it_covers():
"""`checks --json` recomputes machine-local rows only.
Replacing the whole list with it dropped "reboot needed" and "flatpak
remote unreachable" from the screen until the next restart, because those
rows are not things this command can recompute.
"""
source = _code(cli.cmd_checks)
assert _strip('"covers"') in source, source
def test_repo_advisories_are_objects_not_ids():
"""The overlay decodes one shape for both kinds of package."""
from updater import detail as detail_mod
source = _code(detail_mod.repo_package_detail)
assert _strip("advisory_detail(a)") in source, source[-900:]
def test_the_apply_refreshes_only_when_the_plan_did():
"""`-y` re-syncs, so the transaction can install a set the plan never saw."""
from updater import stream as stream_mod
assert "-Syu" in stream_mod.repo_command([], refresh=True)
assert "-Syu" not in stream_mod.repo_command([], refresh=False)
assert "-Su" in stream_mod.repo_command([], refresh=False)
def test_re_running_arch_audit_actually_reruns_it():
"""It served the cache the plan had just written, seconds earlier."""
source = _code(cli.cmd_security)
assert _strip("force=True") in source, source[:900]
def test_a_skipped_package_is_left_out_of_the_resume():
"""Skipping is a decision, not a failure.
It must not be counted with the failures, and it must not come back in the
resume command -- the reader said they wanted this one left.
"""
source = _code(cli._apply_aur_each)
assert _strip('runner, "skipped"') in source, source[-1200:]
assert _strip("passed_over") in source, source[-1200:]
apply_source = _code(cli.cmd_apply)
assert _strip("passed_over") in apply_source, apply_source[-1600:]
assert _strip("unfinished = [name for name, _ in failures] + remaining") in apply_source, (
"a skip leaked into the resume set"
)
def test_the_skip_handler_is_installed_only_where_it_can_be_acted_on():
"""An ordinary CLI run has nothing listening for the key.
Leaving SIGUSR1 at its default keeps an unexpected one a real stop.
"""
source = _code(cli.cmd_apply)
assert _strip("listen_for_skip()") in source, source[:1200]
assert _strip("opts.json_stream") in source, source[:1200]
def test_only_packages_that_build_weigh_anything():
"""A build runs for minutes; a download is over in about a second.
Counting them alike raced the bar to nearly full during the repository
phase, where every package is fast, and left the estimate resting on a mean
of zero-second work.
"""
import tempfile
from pathlib import Path as _P
from updater import stream as stream_mod
with tempfile.TemporaryDirectory() as td:
clone = _P(td) / "src-git"
clone.mkdir()
(clone / "PKGBUILD").write_text("pkgname=src-git\npkgver=1\nbuild() {\n make\n}\n")
plain = _P(td) / "bin-bin"
plain.mkdir()
(plain / "PKGBUILD").write_text(
'pkgname=bin-bin\npkgver=1\npackage() {\n install -Dm755 x "$pkgdir/x"\n}\n'
)
assert cli._package_weight("src-git", _P(td), phase="aur") == stream_mod.BUILD_WEIGHT
assert cli._package_weight("bin-bin", _P(td), phase="aur") == stream_mod.PLAIN_WEIGHT
# a repository package has no PKGBUILD here at all, and is fast anyway
assert cli._package_weight("linux", _P(td), phase="repo") == stream_mod.PLAIN_WEIGHT
def test_the_start_event_carries_work_as_well_as_packages():
from updater import stream as stream_mod
assert stream_mod.BUILD_WEIGHT > 1, "a build has to outweigh a download"
source = _code(cli.cmd_apply)
assert _strip("weight=weight_total") in source, source[:2000]
assert _strip("weights=") in source, source[:2000]