"""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]