diff --git a/archinstall/lib/error_recovery.py b/archinstall/lib/error_recovery.py new file mode 100644 index 0000000000..af61f73c8d --- /dev/null +++ b/archinstall/lib/error_recovery.py @@ -0,0 +1,248 @@ +""" +Interactive recovery from errors that would otherwise abort the installation. + +Rather than exiting on the first failed command, the installer halts and asks the +user how to proceed. Every prompt is built the same way: the caller supplies the +choices that are meaningful for its own failure, and this module appends the two +escape hatches that always apply -- resolve the problem by hand in a shell, or +stop the installation. + +The prompt is deliberately plain ``print``/``input`` and must only be used +outside of the Textual application, i.e. during the installation itself. +""" + +import os +import re +import select +import subprocess +import sys +import time +from dataclasses import dataclass + +from archinstall.lib.log import Font, debug, info, stylize, warn + +# Seconds to wait for an answer before falling back to the safe choice. +DEFAULT_TIMEOUT = 60.0 + +# Returned when the user fixed the problem by hand and asked to resume. +# Callers must handle this key in addition to their own: it means "run the failed +# operation again, unchanged". +RETRY = 'retry' + +_MANUAL = '__manual__' +_EXIT = '__exit__' +_VERSIONED = re.compile(r'(?P[^<>=]+)(?P[<>=].*)$') + + +class InstallationAbortedError(Exception): + """Raised when the user chooses to stop the installation from a recovery prompt.""" + + def __init__(self, reason: str, save_config: bool = False) -> None: + super().__init__(reason) + self.reason = reason + self.save_config = save_config + + +@dataclass(frozen=True) +class Recovery: + """One caller-defined way out of a failure. ``key`` is what :func:`prompt` returns.""" + + key: str + label: str + + +def package(spec: str) -> str: + """ + Render a package reference for use inside a :class:`Recovery` label. + + A version constraint, if the specification carries one, is coloured apart from + the name so that ``mesa>=25.0`` reads as a name plus a requirement. + """ + match = _VERSIONED.match(spec) + + if not match: + return stylize(spec, 'red') + + return stylize(match['name'], 'red') + stylize(match['version'], 'orange') + + +def prompt( + header: str, + options: list[Recovery], + *, + safe: str | None, + detail: str = '', + hint: str = '', + timeout: float = DEFAULT_TIMEOUT, + interactive: bool = True, +) -> str: + """ + Halt, describe the failure and let the user pick a way out. + + ``options`` are the caller's own choices; ``Resolve manually`` and ``Exit`` + are appended automatically. ``safe`` names the option to fall back on when + nobody answers within ``timeout`` seconds, or when the installer is running + unattended. Pass ``safe=None`` when no choice can be made safely without a + human: the prompt then waits indefinitely, and an unattended run aborts + instead of guessing. + + Returns one of the caller's keys or :data:`RETRY`, and raises + :class:`InstallationAbortedError` if the user asks to stop. + """ + if safe is not None and all(option.key != safe for option in options): + raise ValueError(f'Safe option {safe!r} is not among the offered choices') + + choices = [ + *options, + Recovery(_MANUAL, 'Resolve manually in a shell'), + Recovery(_EXIT, 'Exit the installation'), + ] + + if not interactive or not _can_prompt(): + if safe is None: + raise InstallationAbortedError(header) + + debug(f'Not interactive, recovering from "{header}" with the safe option: {safe}') + return safe + + while True: + _render(header, detail, choices) + + answer = _read_line(_question(choices, safe, timeout), timeout if safe is not None else None) + + if answer is None: + # Timed out, or stdin went away and we can no longer ask anything. + if safe is None: + raise InstallationAbortedError(header) + + warn(f'No answer within {int(timeout)}s, continuing with the safe option') + return safe + + if not answer.isdigit() or not 1 <= int(answer) <= len(choices): + print(f'Please enter a number between 1 and {len(choices)}.') + continue + + key = choices[int(answer) - 1].key + + if key == _MANUAL: + open_shell(hint) + + if confirm('Resume the installation now?', default=True): + return RETRY + elif key == _EXIT: + save = confirm('Save the installation configuration before exiting?', default=False) + raise InstallationAbortedError(header, save_config=save) + else: + return key + + +def open_shell(hint: str = '') -> None: + """Hand the terminal over to the user and block until they leave the shell.""" + shell = os.environ.get('SHELL') or '/bin/bash' + + print() + info(f'Pausing the installation and starting {shell}.') + + if hint: + print(hint) + + print(stylize("Type 'exit' or press Ctrl+D to return to the installer.", 'green')) + print() + + try: + subprocess.run([shell], check=False) + except (OSError, KeyboardInterrupt) as err: + warn(f'Could not run {shell}: {err}') + + print() + + +def ask_text(question: str, timeout: float = DEFAULT_TIMEOUT) -> str: + """Ask for a free-form answer. Returns an empty string on a timeout or on EOF.""" + return _read_line(question, timeout) or '' + + +def confirm(question: str, default: bool) -> bool: + """Ask a yes/no question. An empty answer, a timeout or EOF selects ``default``.""" + answer = _read_line(f'{question} {"(Y/n)" if default else "(y/N)"}: ', DEFAULT_TIMEOUT) + + if not answer: + return default + + return answer.lower() in ('y', 'yes') + + +def _question(choices: list[Recovery], safe: str | None, timeout: float) -> str: + prefix = f'Select an option [1-{len(choices)}]' + + if safe is None: + return f'{prefix}: ' + + index = next(i for i, choice in enumerate(choices, 1) if choice.key == safe) + return f'{prefix}, or wait {int(timeout)}s for {index}: ' + + +def _render(header: str, detail: str, choices: list[Recovery]) -> None: + print() + print(stylize(header, 'red', font=[Font.bold])) + + for line in detail.splitlines(): + print(f' {stylize(line, "gray")}') + + print() + + for index, choice in enumerate(choices, 1): + print(f' {stylize(str(index), "green")}) {choice.label}') + + print() + + +def _can_prompt() -> bool: + try: + return sys.stdin.isatty() + except AttributeError, OSError, ValueError: + return False + + +def _read_line(question: str, timeout: float | None) -> str | None: + """ + Read one line from stdin, returning ``None`` on timeout or EOF. + + The remaining time is shown up front rather than counted down live: a + redrawn countdown would overwrite whatever the user is halfway through + typing. + """ + sys.stdout.write(question) + sys.stdout.flush() + + deadline = None if timeout is None else time.monotonic() + timeout + + while deadline is None or (remaining := deadline - time.monotonic()) > 0: + try: + ready, _, _ = select.select([sys.stdin], [], [], None if deadline is None else min(1.0, remaining)) + except (OSError, ValueError) as err: + debug(f'Cannot wait on stdin: {err}') + return None + + if ready: + if line := sys.stdin.readline(): + return line.strip() + + print() + return None # EOF + + print() + return None + + +def report_dropped(packages: list[str]) -> None: + """Re-state everything that was skipped during recovery, so that it cannot pass unnoticed.""" + if not packages: + return + + warn('The following packages were skipped while recovering from installation errors:') + + for name in packages: + warn(f' - {name}') + + warn('The installed system may be incomplete. Check the packages above before relying on it.') diff --git a/archinstall/lib/installer.py b/archinstall/lib/installer.py index 11b89a9cad..1e8707d087 100644 --- a/archinstall/lib/installer.py +++ b/archinstall/lib/installer.py @@ -25,6 +25,7 @@ mount, swapon, ) +from archinstall.lib.error_recovery import InstallationAbortedError, report_dropped from archinstall.lib.exceptions import ( DiskError, HardwareIncompatibilityError, @@ -150,6 +151,12 @@ def __enter__(self) -> Self: def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> bool | None: if exc_type is not None: + if issubclass(exc_type, InstallationAbortedError): + # The user asked to stop from a recovery prompt. That is a choice, + # not a crash, so skip the bug report banner; main() reports it. + self.sync_log_to_install_medium() + return None + error(str(exc_value)) self.sync_log_to_install_medium() @@ -165,6 +172,8 @@ def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseExceptio info(tr('Syncing the system...')) os.sync() + report_dropped(self.pacman.dropped_packages) + if not (missing_steps := self.post_install_check()): msg = f'Installation completed without any errors.\nLog files temporarily available at {logger.directory}.\nYou may reboot when ready.\n' log(msg, fg='green') diff --git a/archinstall/lib/log.py b/archinstall/lib/log.py index 08e4844129..cb8b5e1aa6 100644 --- a/archinstall/lib/log.py +++ b/archinstall/lib/log.py @@ -146,6 +146,24 @@ def _stylize_output( return f'\033[{ansi}m{text}\033[0m' +def stylize( + text: str, + fg: str, + bg: str | None = None, + font: list[Font] | None = None, +) -> str: + """ + Colorize a string for printing outside of the logging helpers. + + Returns the text unchanged when the terminal does not support colors, so the + result is always safe to embed in a message or to write to a pipe. + """ + if not _supports_color(): + return text + + return _stylize_output(text, fg, bg, reset=False, font=font) + + def journal_log(message: str, level: int = logging.DEBUG) -> None: try: import systemd.journal # type: ignore[import-not-found] diff --git a/archinstall/lib/pacman/errors.py b/archinstall/lib/pacman/errors.py new file mode 100644 index 0000000000..ffecb4c7d9 --- /dev/null +++ b/archinstall/lib/pacman/errors.py @@ -0,0 +1,138 @@ +""" +Classification of pacman and pacstrap failures. + +Kept free of any I/O so that it can be exercised against recorded pacman output. +:func:`classify` never raises and never returns ``None``: anything it does not +recognise comes back as :attr:`FailureKind.UNKNOWN`, which callers can still +offer a generic retry for. +""" + +import re +from dataclasses import dataclass, field +from enum import StrEnum, auto + +_CONFLICT = re.compile(r'(?P\S+) and (?P\S+) are in conflict') +_MISSING_TARGET = re.compile(r'target not found:\s*(?P\S+)') +_FILE_EXISTS = re.compile(r'(?P[\w@.+-]+):\s+(?P/\S+) exists in filesystem') +_UNSATISFIED = re.compile(r"unable to satisfy dependency '(?P[^']+)' required by '?(?P[^'\s]+)'?") +_ERROR_LINE = re.compile(r'^error:\s*(?P.+)$') + +# Package names may contain dots, so only strip punctuation that pacman adds +# around them when it builds a sentence. +_TRAILING = '.,;:?!)\'"' + + +class FailureKind(StrEnum): + CONFLICT = auto() + """Two packages cannot be installed together.""" + + FILE_EXISTS = auto() + """A package wants to write files that are already on the target.""" + + MISSING_TARGET = auto() + """A requested package does not exist in any configured repository.""" + + UNSATISFIED_DEPENDENCY = auto() + """A package requires something no repository provides.""" + + UNKNOWN = auto() + """Anything else, including download and signature errors.""" + + +@dataclass(frozen=True) +class Failure: + kind: FailureKind + candidates: tuple[str, ...] = () + """Packages involved, in the order pacman named them. Droppable by the caller.""" + + paths: tuple[str, ...] = field(default=()) + """Conflicting files, only populated for :attr:`FailureKind.FILE_EXISTS`.""" + + summary: str = '' + """The offending output, verbatim, for display.""" + + +def classify(output: str) -> Failure: + """Turn raw pacman output into the most specific failure it describes.""" + lines = [line.strip() for line in output.splitlines() if line.strip()] + + for detect in (_conflict, _file_conflict, _missing_target, _unsatisfied): + if failure := detect(lines): + return failure + + return Failure(FailureKind.UNKNOWN, summary=_last_error(lines)) + + +def strip_version(name: str) -> str: + """ + Reduce a package specification to its bare name: ``foo>=2:1.2-3`` becomes ``foo``. + + Only version constraints are removed. A trailing version is deliberately left + alone, because package names such as ``dotnet-runtime-8.0`` end in one. + """ + return re.split(r'[<>=]', name, maxsplit=1)[0].strip() + + +def _conflict(lines: list[str]) -> Failure | None: + for line in lines: + if match := _CONFLICT.search(line): + pkg_a, pkg_b = _clean(match['pkg_a']), _clean(match['pkg_b']) + + if pkg_a and pkg_b: + return Failure(FailureKind.CONFLICT, (pkg_a, pkg_b), summary=line) + + return None + + +def _file_conflict(lines: list[str]) -> Failure | None: + packages: list[str] = [] + paths: list[str] = [] + + for line in lines: + if match := _FILE_EXISTS.search(line): + if (pkg := _clean(match['pkg'])) and pkg not in packages: + packages.append(pkg) + + paths.append(match['path']) + + if not packages: + return None + + shown = paths[:5] + summary = '\n'.join(shown) + + if len(paths) > len(shown): + summary += f'\n... and {len(paths) - len(shown)} more file(s)' + + return Failure(FailureKind.FILE_EXISTS, tuple(packages), tuple(paths), summary) + + +def _missing_target(lines: list[str]) -> Failure | None: + for line in lines: + if match := _MISSING_TARGET.search(line): + if pkg := _clean(match['pkg']): + return Failure(FailureKind.MISSING_TARGET, (pkg,), summary=line) + + return None + + +def _unsatisfied(lines: list[str]) -> Failure | None: + for line in lines: + if match := _UNSATISFIED.search(line): + if pkg := _clean(match['pkg']): + return Failure(FailureKind.UNSATISFIED_DEPENDENCY, (pkg,), summary=line) + + return None + + +def _last_error(lines: list[str]) -> str: + """The most recent ``error:`` line, which is usually the one that matters.""" + for line in reversed(lines): + if match := _ERROR_LINE.match(line): + return match['message'] + + return lines[-1] if lines else '' + + +def _clean(name: str) -> str: + return name.strip(_TRAILING) diff --git a/archinstall/lib/pacman/pacman.py b/archinstall/lib/pacman/pacman.py index 18477746ee..d0ae34116d 100644 --- a/archinstall/lib/pacman/pacman.py +++ b/archinstall/lib/pacman/pacman.py @@ -1,3 +1,4 @@ +import shlex import sys import time from collections.abc import Callable @@ -5,11 +6,22 @@ from typing import Any from archinstall.lib.command import SysCommand +from archinstall.lib.error_recovery import RETRY, Recovery, ask_text, package, prompt from archinstall.lib.exceptions import RequirementError, SysCallError from archinstall.lib.log import debug, error, info, warn +from archinstall.lib.pacman.errors import Failure, FailureKind, classify, strip_version from archinstall.lib.pathnames import PACMAN_CONF from archinstall.lib.plugins import plugins from archinstall.lib.translationhandler import tr +from archinstall.lib.utils.encoding import clear_vt100_escape_codes_from_str + +_HEADERS = { + FailureKind.CONFLICT: 'Package conflict', + FailureKind.FILE_EXISTS: 'Conflicting files on the target', + FailureKind.MISSING_TARGET: 'Package not found', + FailureKind.UNSATISFIED_DEPENDENCY: 'Unsatisfied dependency', + FailureKind.UNKNOWN: 'Package installation failed', +} class Pacman: @@ -17,12 +29,22 @@ def __init__(self, target: Path, silent: bool = False) -> None: self.synced = False self.silent = silent self.target = target + self.dropped_packages: list[str] = [] + """Packages skipped while recovering from a failure, reported once the installation ends.""" @staticmethod def run(args: str, default_cmd: str = 'pacman') -> SysCommand: """ A centralized function to call `pacman` from. - It also protects us from colliding with other running pacman sessions (if used locally). + """ + Pacman.wait_for_db_lock() + + return SysCommand(f'{default_cmd} {args}') + + @staticmethod + def wait_for_db_lock() -> None: + """ + Protect us from colliding with other running pacman sessions (if used locally). The grace period is set to 10 minutes before exiting hard if another pacman instance is running. """ pacman_db_lock = Path('/var/lib/pacman/db.lck') @@ -38,8 +60,6 @@ def run(args: str, default_cmd: str = 'pacman') -> SysCommand: error(tr('Pre-existing pacman lock never exited. Please clean up any existing pacman sessions before using archinstall.')) sys.exit(1) - return SysCommand(f'{default_cmd} {args}') - def ask(self, error_message: str, bail_message: str, func: Callable, *args: Any, **kwargs: Any) -> None: # type: ignore[type-arg] while True: try: @@ -95,6 +115,7 @@ def _reinit_keyring() -> None: def strap(self, packages: str | list[str]) -> None: self.sync() + if isinstance(packages, str): packages = [packages] @@ -103,12 +124,195 @@ def strap(self, packages: str | list[str]) -> None: if result := plugin.on_pacstrap(packages): packages = result - info(f'Installing packages: {packages}') + packages = list(dict.fromkeys(packages)) # preserve order, remove dups + + if not packages: + return + + overwrite: list[str] = [] + seen: set[str] = set() + + while packages: + self.wait_for_db_lock() + info(f'Installing packages: {packages}') + + command = f'pacstrap -C {PACMAN_CONF} -K {self.target} {" ".join(packages)} --noconfirm --needed' + command += ''.join(f' --overwrite {shlex.quote(path)}' for path in overwrite) + + try: + SysCommand(command, peek_output=True) + return + except SysCallError as err: + error('Could not strap in packages') + + if not self._recover(err, packages, overwrite, seen): + raise RequirementError( + 'Pacstrap failed. See /var/log/archinstall/install.log or above message for error details', + ) from err + + # Everything that was asked for got skipped along the way; the summary at + # the end of the installation lists what is missing. + warn('Continuing without the remaining packages') + + def _recover( + self, + err: SysCallError, + packages: list[str], + overwrite: list[str], + seen: set[str], + ) -> bool: + """ + Halt and ask the user how to deal with a failed pacstrap run. + + `packages` and `overwrite` are updated in place. Returns True when + something actually changed and retrying makes sense, and False when the + caller should give up so that the original error is raised. + """ + failure = classify(_error_output(err)) - self.ask( - 'Could not strap in packages', - 'Pacstrap failed. See /var/log/archinstall/install.log or above message for error details', - SysCommand, - f'pacstrap -C {PACMAN_CONF} -K {self.target} {" ".join(packages)} --noconfirm --needed', - peek_output=True, + # The same failure twice in a row means whatever we offered last time did + # not help, so stop offering it. + signature = f'{failure.kind}:{",".join(failure.candidates)}' + repeated = signature in seen + seen.add(signature) + + droppable = [name for name in failure.candidates if _requested(packages, name)] + options, safe = _recovery_options(failure, droppable, repeated) + + detail = failure.summary + + if not options: + detail += _explain_no_options(failure, droppable) + + choice = prompt( + _HEADERS[failure.kind], + options, + safe=safe, + detail=detail, + hint=f'The target root is mounted at {self.target}; use "arch-chroot {self.target}" to work inside it.', + interactive=not self.silent, ) + + if choice == RETRY: + return True + + if choice == 'overwrite': + overwrite.extend(failure.paths) + return True + + before = list(packages) + + # Every branch below is only ever offered when the packages it names are in + # `droppable`, so indexing into it is safe. + match choice: + case 'keep_first': + self._drop(packages, droppable[1]) + case 'keep_second': + self._drop(packages, droppable[0]) + case 'drop': + self._drop(packages, *droppable) + case 'rename': + self._rename(packages, droppable[0]) + case 'abandon': + self._drop(packages, *list(packages)) + + if packages == before: + warn('Nothing changed, so retrying would fail the same way') + return False + + return True + + def _drop(self, packages: list[str], *names: str) -> None: + targets = {strip_version(name) for name in names} + kept = [] + + for entry in packages: + if strip_version(entry) in targets: + warn(f'Skipping {entry}') + self.dropped_packages.append(entry) + else: + kept.append(entry) + + packages[:] = kept + + def _rename(self, packages: list[str], name: str) -> None: + replacement = ask_text(f'Enter the package name to use instead of {name}: ') + + if not replacement: + warn(f'No replacement given, keeping {name}') + return + + target = strip_version(name) + packages[:] = [replacement if strip_version(entry) == target else entry for entry in packages] + info(f'Replaced {name} with {replacement}') + + +def _recovery_options(failure: Failure, droppable: list[str], repeated: bool) -> tuple[list[Recovery], str | None]: + """ + Build the choices that make sense for a failure. + + The second element names the choice to fall back on when nobody answers in + time. It is None whenever no option can be taken safely without a human, in + which case the prompt waits instead of guessing. + """ + if repeated: + return [Recovery('abandon', 'Skip the remaining packages and continue the installation')], None + + first, *rest = failure.candidates or ('',) + second = rest[0] if rest else '' + + match failure.kind: + case FailureKind.CONFLICT if len(droppable) == 2: + return [ + Recovery('keep_first', f'Keep {package(first)} and drop {package(second)}'), + Recovery('keep_second', f'Keep {package(second)} and drop {package(first)}'), + Recovery('drop', f'Keep neither {package(first)} nor {package(second)} and continue'), + ], 'drop' + case FailureKind.CONFLICT | FailureKind.MISSING_TARGET | FailureKind.UNSATISFIED_DEPENDENCY if droppable: + options = [Recovery('drop', f'Drop {package(droppable[0])} and continue')] + + if failure.kind == FailureKind.MISSING_TARGET: + options.insert(0, Recovery('rename', f'Replace {package(first)} with another package name')) + + return options, 'drop' + case FailureKind.FILE_EXISTS: + options = [Recovery('overwrite', 'Overwrite the conflicting files and retry')] + + if droppable: + options.append(Recovery('drop', f'Drop {package(droppable[0])} and continue')) + + return options, 'drop' if droppable else None + case FailureKind.UNKNOWN: + return [ + Recovery(RETRY, 'Retry installing these packages'), + Recovery('abandon', 'Skip these packages and continue the installation'), + ], RETRY + case _: + return [], None + + +def _explain_no_options(failure: Failure, droppable: list[str]) -> str: + if failure.candidates and not droppable: + names = ', '.join(failure.candidates) + return f'\n{names} was pulled in as a dependency, so it cannot be removed from the requested package list.' + + return '' + + +def _requested(packages: list[str], name: str) -> bool: + """Whether a package was asked for directly, as opposed to being pulled in as a dependency.""" + target = strip_version(name) + return any(strip_version(entry) == target for entry in packages) + + +def _error_output(err: SysCallError) -> str: + """ + The command output to classify. + + `str(err)` is truncated to the last 500 characters, which routinely cuts off + the line naming the conflict, so prefer the full worker log. + """ + if err.worker_log: + return clear_vt100_escape_codes_from_str(err.worker_log.decode('utf-8', errors='replace')) + + return str(err) diff --git a/archinstall/main.py b/archinstall/main.py index 36885db1ba..cd1a2eac83 100644 --- a/archinstall/main.py +++ b/archinstall/main.py @@ -8,8 +8,9 @@ import traceback from pathlib import Path -from archinstall.lib.args import ArchConfigHandler, SubCommand +from archinstall.lib.args import DEFAULT_SAVE_PATH, ArchConfig, ArchConfigHandler, SubCommand from archinstall.lib.disk.utils import disk_layouts +from archinstall.lib.error_recovery import InstallationAbortedError from archinstall.lib.hardware import MemInfo, SysInfo, read_meminfo from archinstall.lib.log import debug, error, info, logger, share_install_log, warn from archinstall.lib.menu.helpers import Confirmation @@ -168,11 +169,26 @@ def run() -> int: mod_name = f'archinstall.scripts.{script}' # by loading the module we'll automatically run the script module = importlib.import_module(mod_name) - module.main(arch_config_handler) + + try: + module.main(arch_config_handler) + except InstallationAbortedError as aborted: + return _abort(aborted, arch_config_handler.config) return 0 +def _abort(aborted: InstallationAbortedError, config: ArchConfig) -> int: + """Stop after the user asked to leave a recovery prompt, keeping their configuration if wanted.""" + warn(f'Installation stopped: {aborted.reason}') + + if aborted.save_config: + config.save() + info(f'Configuration saved to {DEFAULT_SAVE_PATH}') + + return 1 + + def _error_message(exc: Exception) -> None: err = ''.join(traceback.format_exception(exc)) error(err) diff --git a/tests/test_error_recovery.py b/tests/test_error_recovery.py new file mode 100644 index 0000000000..4b4ca8c868 --- /dev/null +++ b/tests/test_error_recovery.py @@ -0,0 +1,130 @@ +import pytest + +from archinstall.lib import error_recovery +from archinstall.lib.error_recovery import RETRY, InstallationAbortedError, Recovery, prompt + +OPTIONS = [ + Recovery('keep_first', 'Keep the first one'), + Recovery('drop', 'Keep neither and continue'), +] + + +@pytest.fixture(autouse=True) +def _interactive(monkeypatch: pytest.MonkeyPatch) -> None: + """Pretend a terminal is attached; individual tests drive the answers themselves.""" + monkeypatch.setattr(error_recovery, '_can_prompt', lambda: True) + + +def _answers(monkeypatch: pytest.MonkeyPatch, *lines: str | None) -> None: + remaining = list(lines) + + def read_line(_question: str, _timeout: float | None) -> str | None: + return remaining.pop(0) + + monkeypatch.setattr(error_recovery, '_read_line', read_line) + + +def test_selecting_an_option_returns_its_key(monkeypatch: pytest.MonkeyPatch) -> None: + _answers(monkeypatch, '1') + + assert prompt('Package conflict', OPTIONS, safe='drop') == 'keep_first' + + +def test_invalid_answers_are_rejected_without_advancing(monkeypatch: pytest.MonkeyPatch) -> None: + _answers(monkeypatch, 'yes', '', '9', '-1', '2') + + assert prompt('Package conflict', OPTIONS, safe='drop') == 'drop' + + +def test_no_answer_falls_back_to_the_safe_option(monkeypatch: pytest.MonkeyPatch) -> None: + _answers(monkeypatch, None) + + assert prompt('Package conflict', OPTIONS, safe='drop') == 'drop' + + +def test_no_answer_and_no_safe_option_aborts(monkeypatch: pytest.MonkeyPatch) -> None: + """Without a safe choice the prompt waits, so a None here means stdin is gone.""" + _answers(monkeypatch, None) + + with pytest.raises(InstallationAbortedError): + prompt('Package conflict', OPTIONS, safe=None) + + +def test_unattended_runs_take_the_safe_option_without_asking(monkeypatch: pytest.MonkeyPatch) -> None: + def unreachable(_question: str, _timeout: float | None) -> str | None: + raise AssertionError('should not have prompted') + + monkeypatch.setattr(error_recovery, '_read_line', unreachable) + + assert prompt('Package conflict', OPTIONS, safe='drop', interactive=False) == 'drop' + + +def test_unattended_runs_abort_when_nothing_is_safe() -> None: + with pytest.raises(InstallationAbortedError): + prompt('Package conflict', OPTIONS, safe=None, interactive=False) + + +def test_the_escape_hatches_are_always_appended(monkeypatch: pytest.MonkeyPatch) -> None: + """Two caller options means 3 is the shell and 4 is exit.""" + _answers(monkeypatch, '4', 'n') + + with pytest.raises(InstallationAbortedError) as raised: + prompt('Package conflict', OPTIONS, safe='drop') + + assert raised.value.save_config is False + + +def test_exiting_can_request_the_configuration_to_be_saved(monkeypatch: pytest.MonkeyPatch) -> None: + _answers(monkeypatch, '4', 'y') + + with pytest.raises(InstallationAbortedError) as raised: + prompt('Package conflict', OPTIONS, safe='drop') + + assert raised.value.save_config is True + assert raised.value.reason == 'Package conflict' + + +def test_resolving_manually_and_resuming_retries_the_operation(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(error_recovery, 'open_shell', lambda _hint='': None) + _answers(monkeypatch, '3', 'y') + + assert prompt('Package conflict', OPTIONS, safe='drop') == RETRY + + +def test_declining_to_resume_shows_the_choices_again(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(error_recovery, 'open_shell', lambda _hint='': None) + _answers(monkeypatch, '3', 'n', '2') + + assert prompt('Package conflict', OPTIONS, safe='drop') == 'drop' + + +def test_a_safe_option_that_is_not_offered_is_a_programming_error() -> None: + with pytest.raises(ValueError, match='not among the offered choices'): + prompt('Package conflict', OPTIONS, safe='nonexistent') + + +def test_a_deliberate_exit_is_not_reported_as_a_crash(capsys: pytest.CaptureFixture[str]) -> None: + """ + The abort travels out through `with Installer(...)`. Choosing to stop is a + decision, so it must not print the "please file a bug" banner. + """ + from archinstall.lib.installer import Installer + + installer = Installer.__new__(Installer) + installer._helper_flags = {} + + propagated = installer.__exit__(InstallationAbortedError, InstallationAbortedError('Package conflict'), None) + + assert propagated is None, 'the abort has to reach main() to be reported there' + assert 'submit this issue' not in capsys.readouterr().out + + +def test_a_genuine_failure_still_asks_for_a_bug_report(capsys: pytest.CaptureFixture[str]) -> None: + from archinstall.lib.installer import Installer + + installer = Installer.__new__(Installer) + installer._helper_flags = {} + + installer.__exit__(RuntimeError, RuntimeError('boom'), None) + + assert 'submit this issue' in capsys.readouterr().out diff --git a/tests/test_pacman_errors.py b/tests/test_pacman_errors.py new file mode 100644 index 0000000000..d5dabe18b7 --- /dev/null +++ b/tests/test_pacman_errors.py @@ -0,0 +1,177 @@ +from archinstall.lib.pacman.errors import FailureKind, classify, strip_version +from archinstall.lib.pacman.pacman import _recovery_options + +CONFLICT = """\ +:: Synchronizing package databases... +resolving dependencies... +looking for conflicting packages... +:: pipewire-jack and jack2 are in conflict. Remove jack2? [y/N] +error: unresolvable package conflicts detected +error: failed to prepare transaction (conflicting dependencies) +""" + +CONFLICT_WITH_PROVIDER = """\ +:: iptables-nft and iptables are in conflict (iptables) +error: unresolvable package conflicts detected +""" + +MISSING_TARGET = """\ +:: Synchronizing package databases... +error: target not found: linux-zen-headerz +""" + +FILE_EXISTS = """\ +checking for file conflicts... +error: failed to commit transaction (conflicting files) +filesystem: /usr/lib/os-release exists in filesystem +glibc: /usr/lib/locale/locale-archive exists in filesystem (owned by glibc-locales) +Errors occurred, no packages were upgraded. +""" + +UNSATISFIED = """\ +resolving dependencies... +:: unable to satisfy dependency 'libfoo=2.0' required by bar +error: failed to prepare transaction (could not satisfy dependencies) +""" + +DOWNLOAD_FAILURE = """\ +error: failed retrieving file 'core.db' from mirror.example.org : Connection timed out +error: failed to synchronize all databases (unexpected error) +""" + + +def test_classify_conflict() -> None: + failure = classify(CONFLICT) + + assert failure.kind == FailureKind.CONFLICT + assert failure.candidates == ('pipewire-jack', 'jack2') + + +def test_classify_conflict_strips_trailing_punctuation() -> None: + """The package name is followed by '.' in 'are in conflict. Remove x?'.""" + assert classify(CONFLICT).candidates[1] == 'jack2' + + +def test_classify_conflict_with_provider_suffix() -> None: + failure = classify(CONFLICT_WITH_PROVIDER) + + assert failure.kind == FailureKind.CONFLICT + assert failure.candidates == ('iptables-nft', 'iptables') + + +def test_classify_missing_target() -> None: + failure = classify(MISSING_TARGET) + + assert failure.kind == FailureKind.MISSING_TARGET + assert failure.candidates == ('linux-zen-headerz',) + + +def test_classify_file_conflict_collects_every_path() -> None: + failure = classify(FILE_EXISTS) + + assert failure.kind == FailureKind.FILE_EXISTS + assert failure.candidates == ('filesystem', 'glibc') + assert failure.paths == ('/usr/lib/os-release', '/usr/lib/locale/locale-archive') + + +def test_classify_unsatisfied_dependency() -> None: + failure = classify(UNSATISFIED) + + assert failure.kind == FailureKind.UNSATISFIED_DEPENDENCY + assert failure.candidates == ('bar',) + + +def test_classify_unknown_keeps_the_last_error_line() -> None: + failure = classify(DOWNLOAD_FAILURE) + + assert failure.kind == FailureKind.UNKNOWN + assert failure.candidates == () + assert failure.summary == 'failed to synchronize all databases (unexpected error)' + + +def test_classify_conflict_wins_over_a_trailing_error_line() -> None: + """A conflict is more actionable than the generic error pacman ends with.""" + assert classify(CONFLICT + DOWNLOAD_FAILURE).kind == FailureKind.CONFLICT + + +def test_classify_empty_output() -> None: + failure = classify('') + + assert failure.kind == FailureKind.UNKNOWN + assert failure.summary == '' + + +def test_strip_version() -> None: + assert strip_version('foo') == 'foo' + assert strip_version('foo>=1.2') == 'foo' + assert strip_version('foo=1.2') == 'foo' + assert strip_version('foo>2:1.2-3') == 'foo' + assert strip_version('lib32-glibc') == 'lib32-glibc' + + +def test_strip_version_keeps_names_that_end_in_a_version() -> None: + """dotnet-runtime-8.0 and friends are package names, not versioned specs.""" + assert strip_version('dotnet-runtime-8.0') == 'dotnet-runtime-8.0' + assert strip_version('python2') == 'python2' + assert strip_version('sdl2-compat') == 'sdl2-compat' + + +def test_conflict_between_two_requested_packages_offers_all_three_choices() -> None: + failure = classify(CONFLICT) + options, safe = _recovery_options(failure, ['pipewire-jack', 'jack2'], repeated=False) + + assert [option.key for option in options] == ['keep_first', 'keep_second', 'drop'] + assert safe == 'drop' + + +def test_conflict_between_dependencies_offers_no_automatic_choice() -> None: + """ + Neither package was requested, so dropping one from the request list would be + a no-op and retrying would fail identically. Nothing may be picked on a timeout. + """ + failure = classify(CONFLICT) + options, safe = _recovery_options(failure, [], repeated=False) + + assert options == [] + assert safe is None + + +def test_conflict_with_one_requested_package_can_only_drop_that_one() -> None: + failure = classify(CONFLICT) + options, safe = _recovery_options(failure, ['jack2'], repeated=False) + + assert [option.key for option in options] == ['drop'] + assert safe == 'drop' + + +def test_missing_target_can_be_renamed_or_dropped() -> None: + failure = classify(MISSING_TARGET) + options, safe = _recovery_options(failure, ['linux-zen-headerz'], repeated=False) + + assert [option.key for option in options] == ['rename', 'drop'] + assert safe == 'drop' + + +def test_file_conflict_never_overwrites_on_a_timeout() -> None: + failure = classify(FILE_EXISTS) + options, safe = _recovery_options(failure, [], repeated=False) + + assert [option.key for option in options] == ['overwrite'] + assert safe is None + + +def test_unknown_failure_retries_once() -> None: + failure = classify(DOWNLOAD_FAILURE) + options, safe = _recovery_options(failure, [], repeated=False) + + assert [option.key for option in options] == ['retry', 'abandon'] + assert safe == 'retry' + + +def test_a_repeated_failure_stops_offering_the_same_way_out() -> None: + """Otherwise the timeout default would loop forever without a keypress.""" + failure = classify(CONFLICT) + options, safe = _recovery_options(failure, ['pipewire-jack', 'jack2'], repeated=True) + + assert [option.key for option in options] == ['abandon'] + assert safe is None