diff --git a/CHANGELOG.md b/CHANGELOG.md index cdc7352..8e2cbe6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,8 +7,29 @@ and versions are tracked in the repo-root `VERSION` file. ## [Unreleased] +### Added + +- Add `ConfigurationError` so consumer profiles can explicitly mark + user-correctable configuration messages as safe usage errors. + +### Changed + +- Normalize command returns, Click errors, aborts, interrupts, `SystemExit`, and + unexpected exceptions through one core outcome model and clean `run_app()` + process boundary. +- Treat plain profile-callback exceptions as private internal failures. Profiles + that used `ValueError` for expected configuration problems should raise + `ConfigurationError` instead. + ### Fixed +- Finalize core-owned run metadata for successful, failed, aborted, interrupted, + and unexpected command outcomes without letting secondary persistence + failures replace the command result. +- Roll back partially constructed command contexts without leaking handlers, + temporary directories, or incomplete run bundles. +- Preserve exception tracebacks in persistent logs and show them on stderr only + when debug output is enabled. - Make history persistence best-effort so secondary failures cannot mask the command outcome or skip cleanup, context reset, and logger shutdown. - Allow finished history records to omit `log_path` when file logging is diff --git a/README.md b/README.md index 6f7b2a0..7e0aa2a 100644 --- a/README.md +++ b/README.md @@ -275,6 +275,8 @@ command result meanings: prevented successful completion. - `ExitCode.USAGE_ERROR` (`2`): the command could not proceed because user input, configuration, or environment setup was invalid or incomplete. +- `ExitCode.INTERRUPTED` (`130`): the user interrupted the command with + Ctrl+C. Existing commands can keep returning integers. New code should prefer the named constants when it makes intent clearer: @@ -285,6 +287,37 @@ if ctx.project_root is None: return base_cli.ExitCode.USAGE_ERROR ``` +`run_app()` is the process boundary for production entry points. It preserves +Click's messages and exit codes for usage and application errors, reports an +explicit abort as `1`, and reports Ctrl+C during startup or +command execution as `130` without a traceback. After the command outcome has +settled, history, metadata, and cleanup are best-effort teardown: even a second +interrupt there cannot replace the primary result. + +An unexpected exception returns `1` with a stable, detail-free message. The run +ID and diagnostic-log path are included when context and file logging are +available. The traceback is kept in the persistent log when enabled and is +shown on stderr with an effective `--debug` setting. A failure before option +parsing can provide a traceback only when `--debug` is an unambiguous leading +flag; otherwise the message says that diagnostic context was unavailable. +Tests or embedding code that need the original exception can pass +`reraise_unexpected=True`. + +| Command result or exception | `outcome` | Exit code | Default message | +| --- | --- | ---: | --- | +| `None` or returned `0` | `success` | 0 | none | +| returned `2` | `usage_error` | 2 | none | +| another returned nonzero integer | `nonzero_return` | returned value | none | +| `click.UsageError` | `usage_error` | exception code | Click usage error | +| another `click.ClickException` | `click_error` | exception code | Click error | +| `click.Abort` | `aborted` | 1 | `Aborted!` | +| Ctrl+C | `interrupted` | 130 | `Interrupted.` | +| `SystemExit` | `system_exit` | normalized payload | string payload, if any | +| another unexpected exception | `unexpected_error` | 1 | stable internal-error message | + +For `SystemExit`, a missing payload becomes `0`, an integer payload is preserved, +and any other payload is printed and normalized to `1`. + ## Context `Context` is the object command code should pass around instead of rediscovering @@ -454,8 +487,26 @@ Windows uses `%LOCALAPPDATA%` (falling back to `~/AppData/Local`). Set `BASE_CLI_CACHE_DIR` to override the default on any platform. The generic profile does not prescribe a product-wide cache name or cleanup command. -Each invocation is a run bundle containing a private `run.json`, `logs/`, and -`tmp/`, while persistent component caches live in the bundle's cache directory. +Each lifecycle-owned invocation is a run bundle containing a private +`run.json`, `logs/`, and `tmp/`, while persistent component caches live in the +owner's cache directory. When persistence succeeds, `run.json` begins with +`status: "running"` and is finalized with `status`, `outcome`, `exit_code`, +`ended_at`, and `duration_ms`, including command failures and interruptions. +The stable outcome values are `success`, `usage_error`, `nonzero_return`, +`click_error`, `aborted`, `interrupted`, `system_exit`, and +`unexpected_error`. Terminal-write failures are warnings and cannot change the +process result; base-cli then removes a matching or corrupt owned record on a +best-effort basis so history data cannot masquerade as authoritative core data. + +Parsing errors, help, and version requests occur before the command lifecycle +owns a bundle and therefore do not create one. Neither do inherited runtimes, +`log_to_file=False`, or dry-run invocations; an explicit log path can still +receive diagnostics in the latter two modes. Context startup is transactional: +if directory creation, logger setup, or retention fails, base-cli closes +partially installed handlers and removes new bundle-local temp/log artifacts +and empty directories. Pre-existing content, persistent component caches, and +parent-runtime data are preserved. + On POSIX, base-cli enforces owner-only `0600`/`0700` modes. On Windows, the default user-local cache root relies on inherited user-profile ACLs; consumers using a custom cache root must provide the appropriate ACL themselves. diff --git a/docs/cache-ownership-and-layout.md b/docs/cache-ownership-and-layout.md index b9c1ebb..320a6be 100644 --- a/docs/cache-ownership-and-layout.md +++ b/docs/cache-ownership-and-layout.md @@ -25,6 +25,29 @@ Each invocation has a private run bundle containing: - `logs/` for diagnostic logs; and - `tmp/` for temporary command data. +The core lifecycle, rather than an optional history adapter, owns `run.json`. +Once command context construction succeeds, the file is written with +`status: "running"`. When persistence succeeds, the core writes a terminal +snapshot containing `status`, `outcome`, `exit_code`, `ended_at`, and +`duration_ms`. Terminal status is `ok` only for exit code zero; all other exit +codes use `error`. The outcome discriminator is one of `success`, +`usage_error`, `nonzero_return`, `click_error`, `aborted`, `interrupted`, +`system_exit`, or `unexpected_error`. + +History may enrich a matching record with consumer fields, but the core writes +the canonical lifecycle fields last. If terminal persistence fails, the +process keeps its primary result and the framework best-effort removes its +matching or corrupt record rather than leave history data or `running` state +looking authoritative. Writes are not yet promised to be atomic. + +The ownership boundary intentionally excludes parser failures, help and version +requests, inherited runtime bindings, `log_to_file=False`, and dry-run mode. +Those invocations do not create or finalize a bundle. If context construction +fails after creating artifacts, rollback closes partial logging handlers and +removes new bundle-local temp/log artifacts and empty directories. It does not +delete pre-existing content, persistent component caches, paths outside the +selected run root, or a parent runtime's metadata. + Persistent component caches live under the owner's `cache/components/` path. On POSIX systems, runtime directories are owner-only (`0700`) and runtime files are owner-only (`0600`). On Windows, the default `%LOCALAPPDATA%` root relies diff --git a/docs/consumer-profiles.md b/docs/consumer-profiles.md index eff306f..8936b10 100644 --- a/docs/consumer-profiles.md +++ b/docs/consumer-profiles.md @@ -80,6 +80,20 @@ translate internal entry-point names into user-facing labels. The generic default only replaces underscores with hyphens; it does not know any product's command aliases. +## Safe profile errors + +Plain exceptions from profile callbacks are treated as unexpected internal +errors: production output hides their details, while `--debug` exposes the +traceback after option parsing. This prevents a programming error or a private +value in a callback from becoming user-facing output by accident. + +For a user-correctable configuration problem whose message is safe to show, +raise `base_cli.ConfigurationError`; `run_app()` renders it as a Click usage +error with exit code `2`. A callback may raise `click.UsageError` or another +`click.ClickException` when it needs Click's standard rendering or a custom +exit code. Consumers that previously raised plain `ValueError` for expected +configuration failures should migrate those sites to `ConfigurationError`. + ## Consumer-owned adapters `App()` uses `CliProfile.generic()` when no profile is supplied. This keeps the diff --git a/lib/python/base_cli/__init__.py b/lib/python/base_cli/__init__.py index de2669d..76e6dbb 100644 --- a/lib/python/base_cli/__init__.py +++ b/lib/python/base_cli/__init__.py @@ -45,6 +45,7 @@ def _resolve_version() -> str: register_record_schema, ) from .context import Context, get_current_context +from .errors import ConfigurationError from .exit_codes import ExitCode from .inspection import inspection_envelope, render_inspection_json from .logging import configure_logger, log_critical, log_debug, log_error, log_info, log_warning @@ -66,6 +67,7 @@ def _resolve_version() -> str: "CliProfile", "CommandFilterNormalizer", "CommandProtocolError", + "ConfigurationError", "Context", "ExitCode", "FieldSpec", diff --git a/lib/python/base_cli/_lifecycle.py b/lib/python/base_cli/_lifecycle.py new file mode 100644 index 0000000..e6a6ffe --- /dev/null +++ b/lib/python/base_cli/_lifecycle.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime +from typing import Any + +from ._private_files import write_private_json +from .context import Context +from .exit_codes import ExitCode +from .history import format_timestamp + + +@dataclass(frozen=True) +class InvocationOutcome: + """Private normalized result of one lifecycle-owned invocation.""" + + kind: str + status: str + exit_code: int + + +@dataclass(frozen=True) +class RunRecorder: + """Write core-owned lifecycle snapshots for one Context.""" + + context: Context + started_at: datetime + started_monotonic_ns: int + + def start(self) -> None: + if self.context._run_metadata_path is None: + return + write_private_json( + self.context._run_metadata_path, + self._metadata(status="running"), + ) + + def finish( + self, + outcome: InvocationOutcome, + *, + ended_at: datetime, + ended_monotonic_ns: int, + ) -> None: + if self.context._run_metadata_path is None: + return + elapsed_ns = max(0, ended_monotonic_ns - self.started_monotonic_ns) + metadata = self._existing_metadata() + metadata.update(self._metadata(status=outcome.status)) + metadata.update( + { + "outcome": outcome.kind, + "exit_code": outcome.exit_code, + "ended_at": format_timestamp(ended_at), + "duration_ms": round(elapsed_ns / 1_000_000), + } + ) + write_private_json(self.context._run_metadata_path, metadata) + + def _existing_metadata(self) -> dict[str, Any]: + path = self.context._run_metadata_path + if path is None: + return {} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + return {} + if not isinstance(payload, dict) or payload.get("run_id") != self.context.run_id: + return {} + return dict(payload) + + def discard_owned_record(self) -> None: + """Remove our misleading record after terminal persistence fails.""" + path = self.context._run_metadata_path + if path is None: + return + try: + payload = json.loads(path.read_text(encoding="utf-8")) + if ( + isinstance(payload, dict) + and payload.get("run_id") == self.context.run_id + ): + path.unlink() + except (UnicodeDecodeError, json.JSONDecodeError): + try: + path.unlink() + except OSError: + pass + except OSError: + pass + + def _metadata(self, *, status: str) -> dict[str, Any]: + context = self.context + return { + "schema_version": 1, + "run_id": context.run_id, + "owner": context.runtime_owner, + "cli": context.cli_name, + "status": status, + "started_at": format_timestamp(self.started_at), + "project": context.project_name, + "project_root": str(context.project_root) if context.project_root else None, + "manifest": str(context.manifest_path) if context.manifest_path else None, + "workspace_root": str(context.workspace_root) if context.workspace_root else None, + } + + +def outcome_from_exit_code(exit_code: int) -> InvocationOutcome: + if exit_code == ExitCode.SUCCESS: + return InvocationOutcome("success", "ok", exit_code) + if exit_code == ExitCode.USAGE_ERROR: + return InvocationOutcome("usage_error", "error", exit_code) + if exit_code == ExitCode.INTERRUPTED: + return InvocationOutcome("interrupted", "error", exit_code) + return InvocationOutcome("nonzero_return", "error", exit_code) + + +def outcome_from_exception(click: Any, exc: BaseException) -> InvocationOutcome: + if isinstance(exc, KeyboardInterrupt): + return InvocationOutcome("interrupted", "error", ExitCode.INTERRUPTED) + if isinstance(exc, EOFError): + return InvocationOutcome("aborted", "error", ExitCode.FAILURE) + if isinstance(exc, click.Abort): + if isinstance(exc.__cause__, KeyboardInterrupt): + return InvocationOutcome("interrupted", "error", ExitCode.INTERRUPTED) + return InvocationOutcome("aborted", "error", ExitCode.FAILURE) + if isinstance(exc, click.exceptions.Exit): + exit_code = _click_exception_exit_code(exc) + if exit_code is None: + return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE) + return outcome_from_exit_code(exit_code) + if isinstance(exc, click.UsageError): + exit_code = _click_exception_exit_code(exc) + if exit_code is None: + return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE) + return InvocationOutcome("usage_error", _status_for_exit_code(exit_code), exit_code) + if isinstance(exc, click.ClickException): + exit_code = _click_exception_exit_code(exc) + if exit_code is None: + return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE) + return InvocationOutcome("click_error", _status_for_exit_code(exit_code), exit_code) + if isinstance(exc, SystemExit): + exit_code = system_exit_code(exc) + return InvocationOutcome("system_exit", _status_for_exit_code(exit_code), exit_code) + return InvocationOutcome("unexpected_error", "error", ExitCode.FAILURE) + + +def system_exit_code(exc: SystemExit) -> int: + if exc.code is None: + return ExitCode.SUCCESS + if isinstance(exc.code, int): + return int(exc.code) + return ExitCode.FAILURE + + +def _status_for_exit_code(exit_code: int) -> str: + return "ok" if exit_code == ExitCode.SUCCESS else "error" + + +def _click_exception_exit_code(exc: Any) -> int | None: + try: + return int(exc.exit_code) + except BaseException: # pylint: disable=broad-exception-caught + return None diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py index 52f46a4..1c94b9f 100644 --- a/lib/python/base_cli/_runtime.py +++ b/lib/python/base_cli/_runtime.py @@ -23,6 +23,10 @@ class RuntimeLayout: _LOG_INDEX_NAME = ".base-cli-log-index.json" +class RuntimeDirectoryError(RuntimeError): + """An actionable failure to create a framework-owned runtime directory.""" + + # pylint: disable=too-many-arguments def runtime_layout( cache_root: Path, @@ -62,7 +66,7 @@ def create_runtime_directory(path: Path, cache_root: Path) -> None: for directory in [path, *missing]: restrict_directory(directory) except OSError as exc: - raise RuntimeError(_runtime_directory_error(path, cache_root, exc)) from exc + raise RuntimeDirectoryError(_runtime_directory_error(path, cache_root, exc)) from exc def runtime_namespace_root(cache_root: Path, namespace: str) -> Path: diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index a5fb842..6d88da0 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -1,15 +1,29 @@ from __future__ import annotations import functools +import logging import os +import shutil import sys -from contextvars import ContextVar +import time +import traceback +from contextvars import ContextVar, Token +from dataclasses import dataclass +from datetime import datetime from pathlib import Path from typing import Any, Callable -from ._runtime import create_runtime_directory, prune_log_files +from ._lifecycle import ( + InvocationOutcome, + RunRecorder, + outcome_from_exception, + outcome_from_exit_code, + system_exit_code, +) from ._private_files import write_private_json -from .context import Context, reset_current_context, set_current_context +from ._runtime import RuntimeDirectoryError, create_runtime_directory, prune_log_files +from .context import Context, recover_current_context, reset_current_context, set_current_context +from .errors import ConfigurationError from .exit_codes import ExitCode from .history import utc_now from .logging import configure_logger, log_invocation @@ -26,18 +40,130 @@ _INVOCATION_ARGV: ContextVar[list[str] | None] = ContextVar("base_cli_invocation_argv", default=None) +@dataclass +class _InvocationState: + run_id: str | None = None + log_file: Path | None = None + debug: bool = False + quiet: bool = False + options_parsed: bool = False + + +_INVOCATION_STATE: ContextVar[_InvocationState | None] = ContextVar("base_cli_invocation_state", default=None) + + +def _reset_context_var(variable: ContextVar[Any], token: Any) -> None: + try: + variable.reset(token) + except BaseException: # pylint: disable=broad-exception-caught + try: + previous = token.old_value + variable.set(None if previous is Token.MISSING else previous) + except BaseException: # pylint: disable=broad-exception-caught + pass + + def _default_log_file(layout: Any, configured_log_file: Path | None) -> Path: return configured_log_file or layout.log_dir / "primary.log" -def _warn_lifecycle_failure(context: Context, message: str, exc: Exception) -> None: +def _warn_lifecycle_failure(context: Context, message: str, exc: BaseException) -> None: """Report a secondary lifecycle failure without breaking teardown.""" try: - context.log.warning("%s: %s", message, exc) - except Exception: # pylint: disable=broad-exception-caught + detail = str(exc) or type(exc).__name__ + context.log.warning("%s: %s", message, detail) + except BaseException: # pylint: disable=broad-exception-caught + pass + + +def _capture_invocation_context(context: Context) -> None: + state = _INVOCATION_STATE.get() + if state is None: + return + state.run_id = context.run_id + state.log_file = context.log_file + state.debug = context.debug + state.quiet = context.quiet + + +def _capture_standard_options(standard: dict[str, Any]) -> None: + state = _INVOCATION_STATE.get() + if state is None: + return + state.debug = bool(standard.get("debug")) + state.quiet = bool(standard.get("quiet")) + state.options_parsed = True + + +def _capture_effective_output_options(*, debug: bool, quiet: bool) -> None: + state = _INVOCATION_STATE.get() + if state is None: + return + state.debug = debug + state.quiet = quiet + + +def _record_unexpected_traceback(context: Context, outcome: InvocationOutcome) -> None: + if outcome.kind != "unexpected_error": + return + try: + context.log.debug("Unexpected command exception", exc_info=True) + except BaseException: # pylint: disable=broad-exception-caught pass +def _start_run_recorder(recorder: RunRecorder) -> None: + try: + recorder.start() + except Exception as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(recorder.context, "Run metadata start failed", exc) + + +def _finish_run_recorder( + recorder: RunRecorder, + outcome: InvocationOutcome, + *, + ended_at: datetime, + ended_monotonic_ns: int, +) -> None: + try: + recorder.finish( + outcome, + ended_at=ended_at, + ended_monotonic_ns=ended_monotonic_ns, + ) + except BaseException as exc: # pylint: disable=broad-exception-caught + path = recorder.context._run_metadata_path + _warn_lifecycle_failure( + recorder.context, + f"Run metadata finalization failed for '{path}'", + exc, + ) + _discard_owned_run_record(recorder) + + +def _discard_owned_run_record(recorder: RunRecorder) -> None: + try: + recorder.discard_owned_record() + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure( + recorder.context, + f"Run metadata recovery failed for '{recorder.context._run_metadata_path}'", + exc, + ) + + +def _reset_active_context(context: Context, token: Any) -> None: + try: + reset_current_context(token) + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(context, "Active context reset failed", exc) + try: + recover_current_context(token) + except BaseException: # pylint: disable=broad-exception-caught + pass + + def _require_click(): try: import click @@ -68,8 +194,8 @@ def __init__( self.log_to_file = log_to_file self.max_log_files = max_log_files # Standalone applications must not inherit a consumer's product - # conventions. Consumers with an existing integration should pass an - # Consumers with product-specific policies should pass an explicit profile. + # conventions. Consumers with product-specific policies should pass an + # explicit profile. self.profile = profile or CliProfile.generic() self._click_command = None self._command_func: Callable[..., Any] | None = None @@ -136,7 +262,12 @@ def _build_click_command(self) -> Any: group.add_command(click.command(*command_args, **command_kwargs)(wrapper)) return group - def _build_command_wrapper(self, click: Any, func: Callable[..., Any], include_version: bool) -> Callable[..., Any]: + def _build_command_wrapper( + self, + click: Any, + func: Callable[..., Any], + include_version: bool, + ) -> Callable[..., Any]: sensitive_options = set(getattr(func, "__base_cli_sensitive_options__", set())) dry_run_parameter = getattr(func, "__base_cli_dry_run_parameter__", "dry_run") @@ -147,15 +278,31 @@ def wrapper(**kwargs: Any): _pop_standard_options(kwargs), ) _validate_standard_options(click, standard) - try: - context = self._create_context(standard, sensitive_options, dry_run=bool(kwargs.get(dry_run_parameter))) - except (RuntimeError, ValueError) as exc: - raise click.ClickException(str(exc)) from exc - token = set_current_context(context) + _capture_standard_options(standard) started_at = utc_now() - exit_code = ExitCode.SUCCESS - invocation_argv = _current_invocation_argv() + started_monotonic_ns = time.monotonic_ns() + context: Context | None = None + recorder: RunRecorder | None = None + outcome = outcome_from_exit_code(ExitCode.SUCCESS) + invocation_argv: list[str] = [] + token = None try: + try: + context = self._create_context( + standard, + sensitive_options, + dry_run=bool(kwargs.get(dry_run_parameter)), + ) + except ConfigurationError as exc: + raise click.UsageError(str(exc)) from exc + except RuntimeDirectoryError as exc: + raise click.ClickException(str(exc)) from exc + + recorder = RunRecorder(context, started_at, started_monotonic_ns) + token = set_current_context(context) + _capture_invocation_context(context) + invocation_argv = _current_invocation_argv() + _start_run_recorder(recorder) log_invocation(context.log, invocation_argv, sensitive_options) if context.project_root is not None: context.log.debug("project_root=%s", context.project_root) @@ -166,31 +313,55 @@ def wrapper(**kwargs: Any): exit_code = _normalize_command_result(result) except TypeError as exc: raise click.ClickException(str(exc)) from exc + outcome = outcome_from_exit_code(exit_code) return result - except Exception: - exit_code = ExitCode.FAILURE + except BaseException as exc: + if context is not None: + outcome = outcome_from_exception(click, exc) + _record_unexpected_traceback(context, outcome) raise finally: - try: - if self.profile.history_writer is not None: - try: + if context is not None: + try: + ended_at = utc_now() + ended_monotonic_ns = time.monotonic_ns() + except BaseException as exc: # pylint: disable=broad-exception-caught + ended_at = started_at + ended_monotonic_ns = started_monotonic_ns + _warn_lifecycle_failure(context, "Terminal clock capture failed", exc) + + try: + if self.profile.history_writer is not None: self.profile.history_writer( context, invocation_argv, sensitive_options, started_at, - exit_code, + outcome.exit_code, ) - except Exception as exc: # pylint: disable=broad-exception-caught - _warn_lifecycle_failure(context, "History finalization failed", exc) - finally: - try: + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(context, "History finalization failed", exc) + + if recorder is None: try: - context.cleanup() - except Exception as exc: # pylint: disable=broad-exception-caught - _warn_lifecycle_failure(context, "Lifecycle cleanup failed", exc) + recorder = RunRecorder(context, started_at, started_monotonic_ns) + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(context, "Run recorder construction failed", exc) + if recorder is not None: + _finish_run_recorder( + recorder, + outcome, + ended_at=ended_at, + ended_monotonic_ns=ended_monotonic_ns, + ) + + try: + context.cleanup() + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(context, "Lifecycle cleanup failed", exc) finally: - reset_current_context(token) + if token is not None: + _reset_active_context(context, token) for kind, param_decls, attrs in getattr(func, "__base_cli_param_specs__", []): if kind == "option": @@ -213,6 +384,7 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], debug = bool(standard.get("debug") or str(config.get("log_level", "")).lower() == "debug") quiet = bool(standard.get("quiet")) keep_temp = bool(standard.get("keep_temp") or config.get("keep_temp")) + _capture_effective_output_options(debug=debug, quiet=quiet) runtime = self.profile.resolve_runtime(self.name, project) cache_root = runtime.cache_root @@ -225,57 +397,26 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], log_file = Path(standard["log_file"]).expanduser() if standard.get("log_file") else None uses_default_log_file = log_file is None - if dry_run or not self.log_to_file: - if log_file is not None: - create_runtime_directory(log_file.parent, cache_root) - else: - for directory in (layout.log_dir, layout.cache_dir, layout.temp_dir): - create_runtime_directory(directory, cache_root) - if log_file is None: - log_file = _default_log_file(layout, runtime.primary_log_file) - create_runtime_directory(log_file.parent, cache_root) - if inherited_path is None and not dry_run and self.log_to_file: - create_runtime_directory(layout.owner_root, cache_root) - create_runtime_directory(layout.run_root, cache_root) - try: - run_metadata = { - "run_id": run_id, - "owner": runtime_owner, - "cli": self.name, - "status": "running", - "started_at": utc_now().isoformat(timespec="seconds").replace("+00:00", "Z"), - "project": selected_project_name, - "project_root": str(selected_project_root) if selected_project_root else None, - "manifest": str(manifest_path) if manifest_path else None, - "workspace_root": str(workspace_root) if workspace_root else None, - } - run_metadata_path = layout.run_root / "run.json" - write_private_json(run_metadata_path, run_metadata) - except OSError: - pass - if runtime.write_identity and selected_project_root is not None and not dry_run and self.log_to_file: - try: - create_runtime_directory(layout.owner_root, cache_root) - identity_path = layout.owner_root / "identity.json" - if not identity_path.exists(): - write_private_json( - identity_path, - { - "schema_version": 1, - "project": selected_project_name, - "project_root": str(selected_project_root), - "manifest": str(manifest_path) if manifest_path is not None else None, - "checkout_id": layout.owner_root.name, - }, - ) - except OSError: - pass - logger = configure_logger(self.name, log_file, debug, quiet=quiet) - logger.debug("cli=%s run_id=%s environment=%s", self.name, run_id, environment) - if self.max_log_files is not None and uses_default_log_file and log_file is not None: - prune_log_files(layout.owner_root / "runs", log_file, self.max_log_files, logger) - - return Context( + if not dry_run and self.log_to_file and log_file is None: + log_file = _default_log_file(layout, runtime.primary_log_file) + + owns_run_metadata = inherited_path is None and not dry_run and self.log_to_file + run_metadata_path = layout.run_root / "run.json" if owns_run_metadata else None + run_root_was_new = not layout.run_root.exists() + temp_dir_was_new = not layout.temp_dir.exists() + rollback_empty_directories = tuple( + directory + for directory in ( + layout.temp_dir.parent, + layout.temp_dir.parent.parent, + layout.log_dir, + layout.run_root, + ) + if not directory.exists() + ) + log_file_existed = log_file.exists() if log_file is not None else False + logger = logging.getLogger(f"base_cli.{self.name}") + context = Context( cli_name=self.name, run_id=run_id, runtime_owner=runtime_owner, @@ -303,9 +444,121 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], history_scope=runtime.history_scope, history_parent_run_id=runtime.history_parent_run_id, ) + context._run_metadata_path = run_metadata_path + + logger_activation_started = False + try: + if owns_run_metadata: + create_runtime_directory(layout.run_root, cache_root) + if dry_run or not self.log_to_file: + if log_file is not None: + create_runtime_directory(log_file.parent, cache_root) + else: + for directory in (layout.log_dir, layout.cache_dir, layout.temp_dir): + create_runtime_directory(directory, cache_root) + if log_file is not None: + create_runtime_directory(log_file.parent, cache_root) + + logger_activation_started = True + try: + context.log = configure_logger(self.name, log_file, debug, quiet=quiet) + except OSError as exc: + target = f"persistent log file '{log_file}'" if log_file is not None else "stderr logging" + raise RuntimeDirectoryError(f"Unable to configure {target}: {exc}") from exc + context.log.debug("cli=%s run_id=%s environment=%s", self.name, run_id, environment) + if self.max_log_files is not None and uses_default_log_file and log_file is not None: + prune_log_files(layout.owner_root / "runs", log_file, self.max_log_files, context.log) + + if runtime.write_identity and selected_project_root is not None and not dry_run and self.log_to_file: + try: + create_runtime_directory(layout.owner_root, cache_root) + identity_path = layout.owner_root / "identity.json" + if not identity_path.exists(): + write_private_json( + identity_path, + { + "schema_version": 1, + "project": selected_project_name, + "project_root": str(selected_project_root), + "manifest": str(manifest_path) if manifest_path is not None else None, + "checkout_id": layout.owner_root.name, + }, + ) + except OSError: + pass + return context + except BaseException: + remove_owned_log = bool( + uses_default_log_file + and log_file is not None + and not log_file_existed + and run_root_was_new + and _path_is_within(log_file, layout.run_root) + ) + _rollback_context_creation( + context, + logger_activation_started=logger_activation_started, + remove_owned_log=remove_owned_log, + remove_new_temp=( + temp_dir_was_new + and _path_is_within(layout.temp_dir, layout.run_root, strict=True) + ), + empty_directories=rollback_empty_directories, + ) + raise + + +def _rollback_context_creation( + context: Context, + *, + logger_activation_started: bool, + remove_owned_log: bool, + remove_new_temp: bool, + empty_directories: tuple[Path, ...], +) -> None: + if logger_activation_started: + keep_temp = context.keep_temp + context.keep_temp = True + try: + try: + context.cleanup() + except BaseException: # pylint: disable=broad-exception-caught + pass + finally: + context.keep_temp = keep_temp + if remove_new_temp: + _remove_new_temp_directory(context.temp_dir) -def run_app(app: App, argv: list[str] | None = None) -> int: + if remove_owned_log and context.log_file is not None: + try: + context.log_file.unlink() + except BaseException: # pylint: disable=broad-exception-caught + pass + for directory in sorted(set(empty_directories), key=lambda path: len(path.parts), reverse=True): + try: + directory.rmdir() + except BaseException: # pylint: disable=broad-exception-caught + pass + + +def _remove_new_temp_directory(temp_dir: Path) -> None: + try: + if temp_dir.exists(): + shutil.rmtree(temp_dir) + except BaseException: # pylint: disable=broad-exception-caught + pass + + +def _path_is_within(path: Path, root: Path, *, strict: bool = False) -> bool: + try: + relative = path.resolve().relative_to(root.resolve()) + except BaseException: # pylint: disable=broad-exception-caught + return False + return not strict or relative != Path(".") + + +def run_app(app: App, argv: list[str] | None = None, *, reraise_unexpected: bool = False) -> int: """Run an :class:`App` and return its normalized process exit code.""" try: @@ -316,26 +569,74 @@ def run_app(app: App, argv: list[str] | None = None) -> int: explicit_argv = argv is not None args = list(sys.argv[1:] if argv is None else argv) + leading_debug, leading_quiet = _leading_output_flags(args) + state = _InvocationState(debug=leading_debug, quiet=leading_quiet) + state_token = _INVOCATION_STATE.set(state) try: - _reject_equals_option_values(click, args) - display_command = app.profile.display_command() - invocation_argv = _effective_invocation_argv(app, args, explicit_argv, display_command) - invocation_token = _INVOCATION_ARGV.set(invocation_argv) try: - if display_command: - result = app.click_command.main(args=args, prog_name=display_command, standalone_mode=False) + _reject_equals_option_values(click, args) + display_command = app.profile.display_command() + invocation_argv = _effective_invocation_argv(app, args, explicit_argv, display_command) + invocation_token = _INVOCATION_ARGV.set(invocation_argv) + try: + if display_command: + result = app.click_command.main(args=args, prog_name=display_command, standalone_mode=False) + else: + result = app.click_command.main(args=args, standalone_mode=False) + finally: + _reset_context_var(_INVOCATION_ARGV, invocation_token) + except click.Abort as exc: + outcome = outcome_from_exception(click, exc) + if outcome.kind == "interrupted": + print("Interrupted.", file=sys.stderr) else: - result = app.click_command.main(args=args, standalone_mode=False) - finally: - _INVOCATION_ARGV.reset(invocation_token) - except click.ClickException as exc: - exc.show() - return int(exc.exit_code) - try: - return _normalize_command_result(result) - except TypeError as exc: - print(f"ERROR: {exc}", file=sys.stderr) - return ExitCode.FAILURE + print("Aborted!", file=sys.stderr) + return outcome.exit_code + except click.ClickException as exc: + outcome = outcome_from_exception(click, exc) + if outcome.kind == "unexpected_error": + if reraise_unexpected: + raise + _show_unexpected_error(state, exc) + return outcome.exit_code + exc.show() + return outcome.exit_code + except KeyboardInterrupt: + print("Interrupted.", file=sys.stderr) + return ExitCode.INTERRUPTED + except SystemExit as exc: + if exc.code is not None and not isinstance(exc.code, int): + print(str(exc.code), file=sys.stderr) + return system_exit_code(exc) + except Exception as exc: + if reraise_unexpected: + raise + _show_unexpected_error(state, exc) + return ExitCode.FAILURE + + try: + return _normalize_command_result(result) + except TypeError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return ExitCode.FAILURE + finally: + _reset_context_var(_INVOCATION_STATE, state_token) + + +def _show_unexpected_error(state: _InvocationState, exc: Exception) -> None: + print("Error: Unexpected internal error.", file=sys.stderr) + if state.run_id is not None: + print(f"Run ID: {state.run_id}", file=sys.stderr) + if state.log_file is not None: + print(f"Diagnostic log: {state.log_file}", file=sys.stderr) + traceback_visible = state.debug and not state.quiet + if traceback_visible and state.run_id is None: + traceback.print_exception(type(exc), exc, exc.__traceback__, file=sys.stderr) + elif not traceback_visible: + if state.options_parsed: + print("Re-run with --debug for a traceback.", file=sys.stderr) + else: + print("Diagnostic context was unavailable before option parsing completed.", file=sys.stderr) def _normalize_command_result(result: Any) -> int: @@ -349,6 +650,19 @@ def _normalize_command_result(result: Any) -> int: ) +def _leading_output_flags(argv: list[str]) -> tuple[bool, bool]: + debug = False + quiet = False + for token in argv: + if token == "--debug": + debug = True + elif token in ("--quiet", "-q"): + quiet = True + else: + break + return debug, quiet + + def _effective_invocation_argv( app: App, args: list[str], diff --git a/lib/python/base_cli/config.py b/lib/python/base_cli/config.py index 058a701..2b51394 100644 --- a/lib/python/base_cli/config.py +++ b/lib/python/base_cli/config.py @@ -4,6 +4,7 @@ from typing import Any from ._dependencies import require_yaml +from .errors import ConfigurationError __all__ = [ @@ -18,11 +19,15 @@ def load_yaml_file(path: Path) -> dict[str, Any]: yaml = require_yaml("PyYAML is required to load the explicit CLI configuration file.") try: - data = yaml.safe_load(path.read_text(encoding="utf-8")) + contents = path.read_text(encoding="utf-8") + except OSError as exc: + raise ConfigurationError(f"Unable to read config file '{path}': {exc}") from exc + try: + data = yaml.safe_load(contents) except yaml.YAMLError as exc: - raise ValueError(f"Config file '{path}' contains invalid YAML: {exc}") from exc + raise ConfigurationError(f"Config file '{path}' contains invalid YAML: {exc}") from exc if data is None: return {} if not isinstance(data, dict): - raise ValueError(f"Config file '{path}' must contain a YAML mapping.") + raise ConfigurationError(f"Config file '{path}' must contain a YAML mapping.") return data diff --git a/lib/python/base_cli/context.py b/lib/python/base_cli/context.py index 20d169c..1efdd7f 100644 --- a/lib/python/base_cli/context.py +++ b/lib/python/base_cli/context.py @@ -49,6 +49,7 @@ class Context: runtime_owner: str = "default" owner_root: Path | None = None run_root: Path | None = None + _run_metadata_path: Path | None = field(default=None, init=False, repr=False, compare=False) def on_cleanup(self, hook: Callable[[], None]) -> None: self.cleanup_hooks.append(hook) @@ -62,14 +63,14 @@ def bind_project(self, project_name: str | None, project_root: Path, manifest_pa def _warn_cleanup_failure(self, message: str, *args: object) -> None: try: self.log.warning(message, *args) - except Exception: # pylint: disable=broad-exception-caught + except BaseException: # pylint: disable=broad-exception-caught pass def cleanup(self) -> None: for hook in self.cleanup_hooks: try: hook() - except Exception as exc: # pylint: disable=broad-exception-caught + except BaseException as exc: # pylint: disable=broad-exception-caught self._warn_cleanup_failure("Cleanup hook failed: %s", exc) if not self.keep_temp and self.temp_dir.exists(): try: @@ -79,18 +80,25 @@ def cleanup(self) -> None: parent.rmdir() except OSError: break - except OSError as exc: + except BaseException as exc: # pylint: disable=broad-exception-caught self._warn_cleanup_failure("Temp directory cleanup failed for '%s': %s", self.temp_dir, exc) for handler in list(self.log.handlers): try: handler.flush() - except Exception as exc: # pylint: disable=broad-exception-caught + except BaseException as exc: # pylint: disable=broad-exception-caught self._warn_cleanup_failure("Log handler flush failed: %s", exc) try: handler.close() - except Exception as exc: # pylint: disable=broad-exception-caught + except BaseException as exc: # pylint: disable=broad-exception-caught self._warn_cleanup_failure("Log handler close failed: %s", exc) - self.log.removeHandler(handler) + try: + self.log.removeHandler(handler) + except BaseException as exc: # pylint: disable=broad-exception-caught + self._warn_cleanup_failure("Log handler removal failed: %s", exc) + try: + self.log.handlers.remove(handler) + except BaseException: # pylint: disable=broad-exception-caught + pass def set_current_context(context: Context | None) -> contextvars.Token[Context | None]: @@ -98,7 +106,15 @@ def set_current_context(context: Context | None) -> contextvars.Token[Context | def reset_current_context(token: contextvars.Token[Context | None]) -> None: - _current_context.reset(token) + try: + _current_context.reset(token) + except BaseException: # pylint: disable=broad-exception-caught + recover_current_context(token) + + +def recover_current_context(token: contextvars.Token[Context | None]) -> None: + previous = token.old_value + _current_context.set(None if previous is contextvars.Token.MISSING else previous) def get_current_context() -> Context: diff --git a/lib/python/base_cli/errors.py b/lib/python/base_cli/errors.py new file mode 100644 index 0000000..119f3d5 --- /dev/null +++ b/lib/python/base_cli/errors.py @@ -0,0 +1,5 @@ +from __future__ import annotations + + +class ConfigurationError(ValueError): + """A user-correctable configuration error that is safe to show.""" diff --git a/lib/python/base_cli/exit_codes.py b/lib/python/base_cli/exit_codes.py index 75d989b..9ffe9ae 100644 --- a/lib/python/base_cli/exit_codes.py +++ b/lib/python/base_cli/exit_codes.py @@ -7,3 +7,4 @@ class ExitCode: SUCCESS = 0 FAILURE = 1 USAGE_ERROR = 2 + INTERRUPTED = 130 diff --git a/lib/python/base_cli/logging.py b/lib/python/base_cli/logging.py index 416e8ca..52b7105 100644 --- a/lib/python/base_cli/logging.py +++ b/lib/python/base_cli/logging.py @@ -116,6 +116,12 @@ def format(self, record: logging.LogRecord) -> str: source = _source_path(record) level = _level_name(record) line = f"{timestamp} {level:<7} {source}:{record.lineno} {record.getMessage()}" + if record.exc_info: + if not record.exc_text: + record.exc_text = self.formatException(record.exc_info) + line = f"{line}\n{record.exc_text}" + if record.stack_info: + line = f"{line}\n{self.formatStack(record.stack_info)}" if not self.use_color: return line color = _LEVEL_COLORS.get(record.levelno) diff --git a/tests/test_app_lifecycle.py b/tests/test_app_lifecycle.py index fc34839..d86e1e5 100644 --- a/tests/test_app_lifecycle.py +++ b/tests/test_app_lifecycle.py @@ -6,8 +6,11 @@ import unittest from dataclasses import replace from pathlib import Path +from unittest import mock import base_cli +import base_cli.app as app_module +import base_cli.context as context_module from base_cli.testing import invoke @@ -128,6 +131,93 @@ def main(ctx: base_cli.Context) -> None: base_cli.get_current_context() self.assertIn("History finalization failed: history unavailable", result.stderr) + def test_non_os_temp_cleanup_failure_still_closes_handlers(self) -> None: + app = base_cli.App(name="cleanup-runtime-failure") + seen: dict[str, object] = {} + + @app.command() + def main(ctx: base_cli.Context) -> None: + seen["logger"] = ctx.log + + with tempfile.TemporaryDirectory() as tmpdir: + with mock.patch.object( + context_module.shutil, + "rmtree", + side_effect=RuntimeError("cleanup implementation failed"), + ): + result = invoke(app, [], home=Path(tmpdir)) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("Temp directory cleanup failed", result.stderr) + self.assertEqual(seen["logger"].handlers, []) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_active_context_reset_interruption_uses_direct_recovery(self) -> None: + app = base_cli.App(name="context-reset-interrupt") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir, mock.patch.object( + app_module, + "reset_current_context", + side_effect=KeyboardInterrupt(), + ): + result = invoke(app, [], home=Path(tmpdir)) + + self.assertEqual(result.exit_code, 0, result.output) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_handler_removal_interruption_uses_direct_detach_fallback(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + logger = logging.Logger("isolated-handler-removal") + logger.addHandler(logging.NullHandler()) + logger.addHandler(logging.NullHandler()) + logger.removeHandler = mock.Mock(side_effect=KeyboardInterrupt()) + context = base_cli.Context( + cli_name="handler-removal-interrupt", + run_id="run-1", + state_dir=root / "state", + log_dir=root / "logs", + cache_dir=root / "cache", + temp_dir=root / "tmp", + log_file=None, + config={}, + environment="dev", + debug=False, + keep_temp=False, + log=logger, + ) + + context.cleanup() + + self.assertEqual(logger.handlers, []) + + def test_context_var_reset_helper_restores_previous_value_after_interrupt(self) -> None: + class Token: + MISSING = object() + old_value = "parent" + + class InterruptedVariable: + def __init__(self) -> None: + self.restored: object | None = None + + def reset(self, _token: object) -> None: + raise KeyboardInterrupt() + + def set(self, value: object) -> None: + self.restored = value + + variable = InterruptedVariable() + + app_module._reset_context_var(variable, Token()) + + self.assertEqual(variable.restored, "parent") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_app_run.py b/tests/test_app_run.py index e01a142..b92d485 100644 --- a/tests/test_app_run.py +++ b/tests/test_app_run.py @@ -18,11 +18,85 @@ def generic_app(**kwargs: object) -> base_cli.App: class RunAppTests(unittest.TestCase): + @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") + def test_malformed_click_exit_code_before_context_is_an_unexpected_error(self) -> None: + import click + + class MalformedExit(click.ClickException): + exit_code = object() + + profile = replace( + base_cli.CliProfile.generic(), + display_command=lambda: (_ for _ in ()).throw( + MalformedExit("private pre-context detail") + ), + ) + app = base_cli.App(name="malformed-pre-context", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + stderr = io.StringIO() + with redirect_stderr(stderr): + status = base_cli.run_app(app, []) + + output = stderr.getvalue() + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", output) + self.assertIn("Diagnostic context was unavailable", output) + self.assertNotIn("private pre-context detail", output) + + @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") + def test_profile_programming_errors_use_the_unexpected_error_boundary(self) -> None: + callbacks = ( + ("discover_project", RuntimeError), + ("load_user_config", ValueError), + ("resolve_workspace_root", RuntimeError), + ("load_config", ValueError), + ("resolve_runtime", RuntimeError), + ("display_command", ValueError), + ) + for field_name, error_type in callbacks: + with self.subTest(field=field_name), tempfile.TemporaryDirectory() as tmpdir: + detail = f"private {field_name} detail" + + def fail_callback(*_args: object) -> object: + raise error_type(detail) + + profile = replace(base_cli.CliProfile.generic(), **{field_name: fail_callback}) + app = base_cli.App(name=f"profile-error-{field_name}", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + home = Path(tmpdir) + stderr = io.StringIO() + with mock.patch.dict( + os.environ, + { + "HOME": str(home), + "BASE_CLI_CACHE_DIR": str(home / ".cache"), + }, + ), redirect_stderr(stderr): + status = base_cli.run_app(app, []) + + output = stderr.getvalue() + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", output) + if field_name == "display_command": + self.assertIn("Diagnostic context was unavailable", output) + else: + self.assertIn("Re-run with --debug for a traceback.", output) + self.assertNotIn(detail, output) + self.assertNotIn("Traceback", output) + @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") def test_run_app_reports_config_errors_without_traceback(self) -> None: profile = base_cli.CliProfile.generic( load_config=lambda _project, _explicit: (_ for _ in ()).throw( - ValueError("workspace must be a mapping when provided.") + base_cli.ConfigurationError("workspace must be a mapping when provided.") ) ) app = base_cli.App(profile=profile, name="bad-config", log_to_file=False) @@ -45,13 +119,37 @@ def main(ctx: base_cli.Context) -> None: ), redirect_stderr(stderr): status = base_cli.run_app(app, []) - self.assertEqual(status, 1) + self.assertEqual(status, 2) self.assertEqual(seen, {}) self.assertIn("workspace must be a mapping", stderr.getvalue()) self.assertNotIn("Traceback", stderr.getvalue()) @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") - def test_run_app_preserves_unexpected_command_exceptions(self) -> None: + def test_generic_invalid_yaml_is_a_safe_usage_error(self) -> None: + app = base_cli.App(name="invalid-yaml", log_to_file=False) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + config = home / "invalid.yml" + config.write_text("broken: [", encoding="utf-8") + stderr = io.StringIO() + with mock.patch.dict( + os.environ, + {"HOME": str(home), "BASE_CLI_CACHE_DIR": str(home / ".cache")}, + ), redirect_stderr(stderr): + status = base_cli.run_app(app, ["--config", str(config)]) + + self.assertEqual(status, 2) + self.assertIn("contains invalid YAML", stderr.getvalue()) + self.assertNotIn("Traceback", stderr.getvalue()) + + @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") + def test_run_app_can_reraise_unexpected_command_exceptions(self) -> None: app = base_cli.App(name="boom", log_to_file=False) @app.command() @@ -69,7 +167,7 @@ def main(ctx: base_cli.Context) -> None: }, ): with self.assertRaisesRegex(RuntimeError, "boom"): - base_cli.run_app(app, []) + base_cli.run_app(app, [], reraise_unexpected=True) @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") def test_run_app_reports_invalid_command_return_values(self) -> None: diff --git a/tests/test_app_run_metadata.py b/tests/test_app_run_metadata.py new file mode 100644 index 0000000..421d00b --- /dev/null +++ b/tests/test_app_run_metadata.py @@ -0,0 +1,931 @@ +from __future__ import annotations + +import importlib.util +import io +import json +import logging +import os +import tempfile +import unittest +from contextlib import redirect_stderr +from dataclasses import replace +from datetime import datetime, timezone +from pathlib import Path +from unittest import mock + +import base_cli +import base_cli._lifecycle as lifecycle_module +import base_cli.app as app_module +from base_cli._lifecycle import RunRecorder +from base_cli._runtime import runtime_layout + + +def _run(app: base_cli.App, home: Path, args: list[str] | None = None) -> tuple[int, str]: + stderr = io.StringIO() + with mock.patch.dict( + os.environ, + { + "HOME": str(home), + "BASE_CLI_CACHE_DIR": str(home / "cache"), + }, + ), redirect_stderr(stderr): + status = base_cli.run_app(app, args or []) + return status, stderr.getvalue() + + +def _metadata_files(home: Path) -> list[Path]: + return sorted((home / "cache").glob("**/run.json")) + + +def _load_only_metadata(test: unittest.TestCase, home: Path) -> tuple[Path, dict[str, object]]: + paths = _metadata_files(home) + test.assertEqual(len(paths), 1, paths) + payload = json.loads(paths[0].read_text(encoding="utf-8")) + test.assertIsInstance(payload, dict) + return paths[0], payload + + +def _assert_terminal_metadata( + test: unittest.TestCase, + payload: dict[str, object], + *, + status: str, + outcome: str, + exit_code: int, +) -> None: + test.assertEqual(payload["schema_version"], 1) + test.assertEqual(payload["status"], status) + test.assertEqual(payload["outcome"], outcome) + test.assertEqual(payload["exit_code"], exit_code) + test.assertIsInstance(payload["run_id"], str) + test.assertIsInstance(payload["owner"], str) + test.assertIsInstance(payload["cli"], str) + started_text = str(payload["started_at"]) + ended_text = str(payload["ended_at"]) + test.assertTrue(started_text.endswith("Z"), started_text) + test.assertTrue(ended_text.endswith("Z"), ended_text) + started_at = datetime.fromisoformat(started_text.replace("Z", "+00:00")) + ended_at = datetime.fromisoformat(ended_text.replace("Z", "+00:00")) + test.assertEqual(started_at.utcoffset(), timezone.utc.utcoffset(started_at)) + test.assertEqual(ended_at.utcoffset(), timezone.utc.utcoffset(ended_at)) + test.assertGreaterEqual(ended_at, started_at) + test.assertIs(type(payload["duration_ms"]), int) + test.assertGreaterEqual(payload["duration_ms"], 0) + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class AppRunMetadataTests(unittest.TestCase): + def test_normal_returns_finalize_core_owned_metadata(self) -> None: + cases = ( + ("none", None, 0, "ok", "success"), + ("zero", 0, 0, "ok", "success"), + ("usage", 2, 2, "error", "usage_error"), + ("nonzero", 7, 7, "error", "nonzero_return"), + ) + for name, returned, expected_code, expected_status, expected_outcome in cases: + with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir: + app = base_cli.App(name=f"metadata-{name}") + + @app.command() + def main(ctx: base_cli.Context) -> int | None: + del ctx + return returned + + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, expected_code) + self.assertEqual(stderr, "") + _assert_terminal_metadata( + self, + metadata, + status=expected_status, + outcome=expected_outcome, + exit_code=expected_code, + ) + + def test_command_usage_error_preserves_click_rendering_and_exit_code(self) -> None: + import click + + app = base_cli.App(name="metadata-usage-error") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise click.UsageError("choose a valid target") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 2) + self.assertIn("Usage:", stderr) + self.assertIn("Error: choose a valid target", stderr) + _assert_terminal_metadata(self, metadata, status="error", outcome="usage_error", exit_code=2) + + def test_click_exception_preserves_custom_exit_code(self) -> None: + import click + + class Unavailable(click.ClickException): + exit_code = 78 + + app = base_cli.App(name="metadata-click-error") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise Unavailable("service unavailable") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 78) + self.assertIn("Error: service unavailable", stderr) + _assert_terminal_metadata(self, metadata, status="error", outcome="click_error", exit_code=78) + + def test_zero_code_click_exception_keeps_code_and_status_consistent(self) -> None: + import click + + class InformationalExit(click.ClickException): + exit_code = 0 + + app = base_cli.App(name="metadata-zero-click-error") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise InformationalExit("informational stop") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertIn("Error: informational stop", stderr) + _assert_terminal_metadata(self, metadata, status="ok", outcome="click_error", exit_code=0) + + def test_malformed_click_exit_code_is_an_unexpected_error(self) -> None: + import click + + class MalformedExit(click.ClickException): + exit_code = "not-an-exit-code" + + app = base_cli.App(name="metadata-malformed-click-error") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise MalformedExit("private malformed exception detail") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", stderr) + self.assertNotIn("private malformed exception detail", stderr) + _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1) + + def test_abort_and_keyboard_interrupt_have_distinct_outcomes(self) -> None: + import click + + cases = ( + ("abort", click.Abort(), 1, "aborted", "Aborted!"), + ("interrupt", KeyboardInterrupt(), 130, "interrupted", "Interrupted."), + ) + for name, raised, expected_code, expected_outcome, expected_message in cases: + with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir: + app = base_cli.App(name=f"metadata-{name}") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise raised + + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, expected_code) + self.assertIn(expected_message, stderr) + self.assertNotIn("Traceback", stderr) + _assert_terminal_metadata( + self, + metadata, + status="error", + outcome=expected_outcome, + exit_code=expected_code, + ) + + def test_explicit_click_and_system_exits_are_normalized(self) -> None: + import click + + cases = ( + ("click", click.exceptions.Exit(9), 9, "error", "nonzero_return", ""), + ("system-none", SystemExit(None), 0, "ok", "system_exit", ""), + ("system-success", SystemExit(0), 0, "ok", "system_exit", ""), + ("system-failure", SystemExit(5), 5, "error", "system_exit", ""), + ("system-message", SystemExit("exit detail"), 1, "error", "system_exit", "exit detail\n"), + ) + for name, raised, expected_code, expected_status, expected_outcome, expected_stderr in cases: + with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir: + app = base_cli.App(name=f"metadata-{name}") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise raised + + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, expected_code) + self.assertEqual(stderr, expected_stderr) + _assert_terminal_metadata( + self, + metadata, + status=expected_status, + outcome=expected_outcome, + exit_code=expected_code, + ) + + def test_unexpected_error_is_clean_but_persists_traceback(self) -> None: + app = base_cli.App(name="metadata-unexpected") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise RuntimeError("private failure detail") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + metadata_path, metadata = _load_only_metadata(self, home) + log_text = (metadata_path.parent / "logs" / "primary.log").read_text(encoding="utf-8") + + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", stderr) + self.assertIn(f"Run ID: {metadata['run_id']}", stderr) + self.assertIn("Diagnostic log:", stderr) + self.assertIn("Re-run with --debug for a traceback.", stderr) + self.assertNotIn("private failure detail", stderr) + self.assertNotIn("Traceback", stderr) + self.assertIn("Traceback", log_text) + self.assertIn("RuntimeError: private failure detail", log_text) + _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1) + + def test_debug_mirrors_unexpected_traceback_to_stderr(self) -> None: + app = base_cli.App(name="metadata-debug") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise RuntimeError("debug failure detail") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home, ["--debug"]) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 1) + self.assertIn("Traceback", stderr) + self.assertIn("RuntimeError: debug failure detail", stderr) + self.assertNotIn("Re-run with --debug", stderr) + _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1) + + def test_unexpected_error_without_file_logging_reports_only_available_diagnostics(self) -> None: + app = base_cli.App(name="metadata-no-file-error", log_to_file=False) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise RuntimeError("no-file private detail") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", stderr) + self.assertIn("Run ID:", stderr) + self.assertNotIn("Diagnostic log:", stderr) + self.assertNotIn("no-file private detail", stderr) + self.assertIn("Re-run with --debug for a traceback.", stderr) + self.assertEqual(_metadata_files(home), []) + + def test_debug_shows_traceback_for_failure_before_context_activation(self) -> None: + def fail_discovery(_cwd: Path) -> base_cli.ProjectInfo | None: + raise RuntimeError("pre-context private detail") + + profile = base_cli.CliProfile.generic(discover_project=fail_discovery) + app = base_cli.App(name="metadata-pre-context", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home, ["--debug"]) + + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", stderr) + self.assertIn("Traceback", stderr) + self.assertIn("RuntimeError: pre-context private detail", stderr) + self.assertNotIn("Re-run with --debug", stderr) + self.assertNotIn("Run ID:", stderr) + self.assertNotIn("Diagnostic log:", stderr) + self.assertEqual(_metadata_files(home), []) + + def test_leading_debug_shows_traceback_for_failure_before_click_parsing(self) -> None: + def fail_display_command() -> str | None: + raise RuntimeError("pre-parser private detail") + + profile = replace(base_cli.CliProfile.generic(), display_command=fail_display_command) + app = base_cli.App(name="metadata-pre-parser", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home, ["--debug"]) + + self.assertEqual(status, 1) + self.assertIn("Traceback", stderr) + self.assertIn("RuntimeError: pre-parser private detail", stderr) + self.assertNotIn("Diagnostic context was unavailable", stderr) + + def test_debug_in_an_option_value_position_never_exposes_pre_parser_failure(self) -> None: + def fail_display_command() -> str | None: + raise RuntimeError("value-position private detail") + + profile = replace(base_cli.CliProfile.generic(), display_command=fail_display_command) + app = base_cli.App(name="metadata-value-position", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home, ["--config", "--debug"]) + + self.assertEqual(status, 1) + self.assertNotIn("Traceback", stderr) + self.assertNotIn("value-position private detail", stderr) + self.assertIn("Diagnostic context was unavailable", stderr) + + def test_debug_literal_after_option_terminator_does_not_expose_traceback(self) -> None: + def fail_discovery(_cwd: Path) -> base_cli.ProjectInfo | None: + raise KeyError("literal-debug private detail") + + profile = base_cli.CliProfile.generic(discover_project=fail_discovery) + app = base_cli.App(name="metadata-literal-debug", profile=profile) + + @app.command() + @base_cli.argument("value") + def main(ctx: base_cli.Context, value: str) -> None: + del ctx, value + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home, ["--", "--debug"]) + + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", stderr) + self.assertNotIn("Traceback", stderr) + self.assertNotIn("literal-debug private detail", stderr) + self.assertIn("Re-run with --debug for a traceback.", stderr) + + def test_config_derived_debug_applies_to_later_startup_failure(self) -> None: + def fail_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + raise KeyError("configured-debug private detail") + + profile = replace( + base_cli.CliProfile.generic( + load_config=lambda _project, _explicit: {"log_level": "debug"}, + ), + resolve_runtime=fail_runtime, + ) + app = base_cli.App(name="metadata-configured-debug-startup", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + + self.assertEqual(status, 1) + self.assertIn("Traceback", stderr) + self.assertIn("KeyError: 'configured-debug private detail'", stderr) + self.assertNotIn("Re-run with --debug", stderr) + + def test_config_debug_with_quiet_keeps_traceback_out_of_stderr_and_shows_hint(self) -> None: + profile = base_cli.CliProfile.generic( + load_config=lambda _project, _explicit: {"log_level": "debug"}, + ) + app = base_cli.App(name="metadata-config-debug-quiet", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise RuntimeError("quiet private detail") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home, ["--quiet"]) + metadata_path, metadata = _load_only_metadata(self, home) + log_text = (metadata_path.parent / "logs" / "primary.log").read_text(encoding="utf-8") + + self.assertEqual(status, 1) + self.assertNotIn("Traceback", stderr) + self.assertNotIn("quiet private detail", stderr) + self.assertIn("Re-run with --debug for a traceback.", stderr) + self.assertIn("RuntimeError: quiet private detail", log_text) + _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1) + + def test_traceback_logging_interruption_cannot_replace_primary_exception(self) -> None: + app = base_cli.App(name="metadata-traceback-interrupt") + + @app.command() + def main(ctx: base_cli.Context) -> None: + ctx.log.debug = mock.Mock(side_effect=KeyboardInterrupt()) + raise RuntimeError("primary private detail") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 1) + self.assertIn("Error: Unexpected internal error.", stderr) + self.assertNotIn("Interrupted.", stderr) + _assert_terminal_metadata(self, metadata, status="error", outcome="unexpected_error", exit_code=1) + + def test_history_writer_cannot_override_core_terminal_outcome(self) -> None: + observed_codes: list[int] = [] + + def history_writer( + ctx: base_cli.Context, + _argv: list[str], + _sensitive: set[str], + _started: object, + exit_code: int, + ) -> None: + observed_codes.append(exit_code) + assert ctx.run_root is not None + (ctx.run_root / "run.json").write_text( + json.dumps( + { + "run_id": ctx.run_id, + "status": "error", + "exit_code": 99, + "command": "metadata-history-authority", + "custom_history_field": "preserved", + } + ), + encoding="utf-8", + ) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + app = base_cli.App(name="metadata-history-authority", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, _ = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertEqual(observed_codes, [0]) + self.assertEqual(metadata["command"], "metadata-history-authority") + self.assertEqual(metadata["custom_history_field"], "preserved") + _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0) + + def test_history_failure_does_not_block_core_finalization(self) -> None: + def fail_history(*_args: object) -> None: + raise OSError("history unavailable") + + profile = replace(base_cli.CliProfile.generic(), history_writer=fail_history) + app = base_cli.App(name="metadata-history-failure", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertIn("History finalization failed: history unavailable", stderr) + _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0) + + def test_command_duration_excludes_history_writer_latency(self) -> None: + observed: list[None] = [] + + def history_writer(*_args: object) -> None: + observed.append(None) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + app = base_cli.App(name="metadata-duration", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object( + app_module.time, + "monotonic_ns", + side_effect=(1_000_000_000, 1_025_600_000), + ): + status, _ = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertEqual(observed, [None]) + self.assertEqual(metadata["duration_ms"], 26) + + def test_terminal_write_failure_preserves_primary_outcome_and_discards_running_marker(self) -> None: + cases = (("success", False, 0), ("failure", True, 1)) + for name, fail_command, expected_code in cases: + with self.subTest(name=name), tempfile.TemporaryDirectory() as tmpdir: + app = base_cli.App(name=f"metadata-finalize-{name}") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + if fail_command: + raise RuntimeError("command failure") + + home = Path(tmpdir) + with mock.patch.object(RunRecorder, "finish", side_effect=RuntimeError("metadata unavailable")): + status, stderr = _run(app, home) + + self.assertEqual(status, expected_code) + self.assertIn("Run metadata finalization failed", stderr) + self.assertEqual(_metadata_files(home), []) + self.assertEqual(logging.getLogger(f"base_cli.metadata-finalize-{name}").handlers, []) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_metadata_recovery_failure_cannot_replace_command_outcome(self) -> None: + app = base_cli.App(name="metadata-recovery-failure") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with ( + mock.patch.object(RunRecorder, "finish", side_effect=RuntimeError("finish unavailable")), + mock.patch.object( + RunRecorder, + "discard_owned_record", + side_effect=RuntimeError("recovery unavailable"), + ), + ): + status, stderr = _run(app, home) + + self.assertEqual(status, 0) + self.assertIn("Run metadata finalization failed", stderr) + self.assertIn("Run metadata recovery failed", stderr) + self.assertEqual(logging.getLogger("base_cli.metadata-recovery-failure").handlers, []) + + def test_finish_failure_removes_matching_history_owned_terminal_fields(self) -> None: + def history_writer( + ctx: base_cli.Context, + _argv: list[str], + _sensitive: set[str], + _started: object, + _exit_code: int, + ) -> None: + assert ctx.run_root is not None + (ctx.run_root / "run.json").write_text( + json.dumps({"run_id": ctx.run_id, "status": "error", "exit_code": 99}), + encoding="utf-8", + ) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + app = base_cli.App(name="metadata-history-finish-failure", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object(RunRecorder, "finish", side_effect=OSError("finish unavailable")): + status, stderr = _run(app, home) + metadata_files = _metadata_files(home) + + self.assertEqual(status, 0) + self.assertIn("Run metadata finalization failed", stderr) + self.assertEqual(metadata_files, []) + + def test_partial_terminal_write_is_removed_without_masking_success(self) -> None: + app = base_cli.App(name="metadata-partial-write") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + real_write = lifecycle_module.write_private_json + calls = 0 + + def fail_second_write(path: Path, value: dict[str, object]) -> None: + nonlocal calls + calls += 1 + if calls == 2: + path.write_text("{", encoding="utf-8") + raise OSError("partial terminal write") + real_write(path, value) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object(lifecycle_module, "write_private_json", side_effect=fail_second_write): + status, stderr = _run(app, home) + metadata_files = _metadata_files(home) + + self.assertEqual(status, 0) + self.assertEqual(calls, 2) + self.assertIn("Run metadata finalization failed", stderr) + self.assertEqual(metadata_files, []) + + def test_terminal_merge_rejects_metadata_from_a_different_run(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + layout = runtime_layout(cache_root, "metadata-stale-merge", "new-run") + layout.run_root.mkdir(parents=True) + metadata_path = layout.run_root / "run.json" + metadata_path.write_text( + json.dumps({"run_id": "old-run", "status": "ok", "stale_field": "do not copy"}), + encoding="utf-8", + ) + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="custom-owner", + project_root=None, + project_name=None, + inherited_path=None, + history_parent_run_id=None, + run_id="new-run", + ) + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="metadata-stale-merge", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with mock.patch.object(RunRecorder, "start", return_value=None): + status, _ = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertEqual(metadata["run_id"], "new-run") + self.assertEqual(metadata["owner"], "custom-owner") + self.assertNotIn("stale_field", metadata) + _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0) + + def test_start_write_failure_can_recover_with_terminal_snapshot(self) -> None: + app = base_cli.App(name="metadata-start-failure") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object(RunRecorder, "start", side_effect=RuntimeError("metadata unavailable")): + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertIn("Run metadata start failed: metadata unavailable", stderr) + _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0) + + def test_interrupt_during_metadata_start_still_finalizes_and_tears_down(self) -> None: + app = base_cli.App(name="metadata-start-interrupt") + called: list[None] = [] + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + called.append(None) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object(RunRecorder, "start", side_effect=KeyboardInterrupt()): + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 130) + self.assertEqual(called, []) + self.assertIn("Interrupted.", stderr) + _assert_terminal_metadata(self, metadata, status="error", outcome="interrupted", exit_code=130) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_interrupt_during_history_cannot_replace_settled_command_outcome(self) -> None: + def interrupt_history(*_args: object) -> None: + raise KeyboardInterrupt() + + profile = replace(base_cli.CliProfile.generic(), history_writer=interrupt_history) + app = base_cli.App(name="metadata-history-interrupt", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertIn("History finalization failed: KeyboardInterrupt", stderr) + _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0) + self.assertEqual(logging.getLogger("base_cli.metadata-history-interrupt").handlers, []) + + def test_interrupt_during_terminal_write_cannot_replace_settled_command_outcome(self) -> None: + app = base_cli.App(name="metadata-finish-interrupt") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + calls = 0 + + def interrupt_finish(recorder: RunRecorder, *args: object, **kwargs: object) -> None: + nonlocal calls + del recorder, args, kwargs + calls += 1 + raise KeyboardInterrupt() + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object(RunRecorder, "finish", new=interrupt_finish): + status, stderr = _run(app, home) + metadata_files = _metadata_files(home) + + self.assertEqual(status, 0) + self.assertEqual(calls, 1) + self.assertIn("Run metadata finalization failed", stderr) + self.assertEqual(metadata_files, []) + self.assertEqual(logging.getLogger("base_cli.metadata-finish-interrupt").handlers, []) + + def test_interrupt_during_cleanup_cannot_replace_settled_command_outcome(self) -> None: + app = base_cli.App(name="metadata-cleanup-interrupt") + + @app.command() + def main(ctx: base_cli.Context) -> None: + ctx.on_cleanup(lambda: (_ for _ in ()).throw(KeyboardInterrupt())) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, stderr = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertIn("Cleanup hook failed", stderr) + _assert_terminal_metadata(self, metadata, status="ok", outcome="success", exit_code=0) + self.assertEqual(logging.getLogger("base_cli.metadata-cleanup-interrupt").handlers, []) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_terminal_metadata_refreshes_project_binding(self) -> None: + app = base_cli.App(name="metadata-project-binding") + + @app.command() + def main(ctx: base_cli.Context) -> None: + ctx.bind_project("bound-project", Path("/tmp/bound-project"), Path("/tmp/bound-project/project.yml")) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + status, _ = _run(app, home) + _, metadata = _load_only_metadata(self, home) + + self.assertEqual(status, 0) + self.assertEqual(metadata["project"], "bound-project") + self.assertEqual(metadata["project_root"], str(Path("/tmp/bound-project").resolve())) + self.assertEqual(metadata["manifest"], str(Path("/tmp/bound-project/project.yml").resolve())) + + def test_parse_error_and_no_file_modes_do_not_own_run_metadata(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + parse_app = base_cli.App(name="metadata-parse-error") + + @parse_app.command() + @base_cli.option("--name", required=True) + def parse_main(ctx: base_cli.Context, name: str) -> None: + del ctx, name + + status, _ = _run(parse_app, home) + self.assertEqual(status, 2) + self.assertEqual(_metadata_files(home), []) + + informational_app = base_cli.App(name="metadata-informational", version="1.2.3") + + @informational_app.command() + def informational_main(ctx: base_cli.Context) -> None: + del ctx + + for args in (["--help"], ["--version"]): + status, _ = _run(informational_app, home, args) + self.assertEqual(status, 0) + self.assertEqual(_metadata_files(home), []) + + no_file_app = base_cli.App(name="metadata-no-file", log_to_file=False) + + @no_file_app.command() + def no_file_main(ctx: base_cli.Context) -> None: + del ctx + + status, _ = _run(no_file_app, home) + self.assertEqual(status, 0) + self.assertEqual(_metadata_files(home), []) + + dry_run_app = base_cli.App(name="metadata-dry-run") + + @dry_run_app.command() + @base_cli.option("--dry-run", is_flag=True, dry_run=True) + def dry_run_main(ctx: base_cli.Context, dry_run: bool) -> None: + self.assertTrue(ctx.dry_run) + self.assertTrue(dry_run) + + status, _ = _run(dry_run_app, home, ["--dry-run"]) + self.assertEqual(status, 0) + self.assertEqual(_metadata_files(home), []) + + def test_inherited_runtime_does_not_mutate_parent_metadata(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + parent_run = cache_root / "parent" / "runs" / "parent-run" + parent_run.mkdir(parents=True) + parent_metadata = parent_run / "run.json" + parent_payload = {"run_id": "parent-run", "status": "running", "custom": "parent-owned"} + parent_metadata.write_text(json.dumps(parent_payload), encoding="utf-8") + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=runtime_layout( + cache_root, + "metadata-inherited", + "child-run", + namespace="parent", + inherited_run_root=parent_run, + ), + application_home=None, + runtime_owner="parent", + project_root=None, + project_name=None, + inherited_path=parent_run, + history_parent_run_id="parent-run", + run_id="child-run", + ) + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="metadata-inherited", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + status, _ = _run(app, home) + + self.assertEqual(status, 0) + self.assertEqual(json.loads(parent_metadata.read_text(encoding="utf-8")), parent_payload) + self.assertEqual(_metadata_files(home), [parent_metadata]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_app_runtime_errors.py b/tests/test_app_runtime_errors.py index 2e2354d..5321ae5 100644 --- a/tests/test_app_runtime_errors.py +++ b/tests/test_app_runtime_errors.py @@ -73,3 +73,31 @@ def main(ctx: base_cli.Context) -> None: self.assertIn("Unable to create runtime directory", error) self.assertIn(str(cache_root / "cache-failure" / "runs"), error) self.assertNotIn("Traceback", error) + + @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") + def test_run_app_reports_framework_log_path_failure_without_traceback(self) -> None: + app = generic_app(name="log-open-failure") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command body should not run when logging setup fails") + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + not_a_directory = root / "not-a-directory" + not_a_directory.write_text("file", encoding="utf-8") + log_file = not_a_directory / "primary.log" + stderr = io.StringIO() + with mock.patch.dict( + os.environ, + {"HOME": str(home), "BASE_CLI_CACHE_DIR": str(root / "cache")}, + ), redirect_stderr(stderr): + exit_code = base_cli.run_app(app, ["--log-file", str(log_file)]) + + error = stderr.getvalue() + self.assertEqual(exit_code, 1) + self.assertIn(f"Unable to create runtime directory '{not_a_directory}'", error) + self.assertNotIn("Unexpected internal error", error) + self.assertNotIn("Traceback", error) diff --git a/tests/test_app_startup_transaction.py b/tests/test_app_startup_transaction.py new file mode 100644 index 0000000..b22ae74 --- /dev/null +++ b/tests/test_app_startup_transaction.py @@ -0,0 +1,243 @@ +from __future__ import annotations + +import importlib.util +import json +import logging +import os +import tempfile +import unittest +from dataclasses import replace +from pathlib import Path +from unittest import mock + +import base_cli +import base_cli.app as app_module +from base_cli._runtime import runtime_layout + + +def _run(app: base_cli.App, home: Path) -> int: + with mock.patch.dict( + os.environ, + { + "HOME": str(home), + "BASE_CLI_CACHE_DIR": str(home / "cache"), + }, + ): + return base_cli.run_app(app, []) + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class AppStartupTransactionTests(unittest.TestCase): + def test_retention_failure_rolls_back_new_bundle_and_logger(self) -> None: + app = base_cli.App(name="startup-retention", max_log_files=1) + called: list[None] = [] + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + called.append(None) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + preserved = home / "cache" / "startup-retention" / "cache" / "existing.txt" + preserved.parent.mkdir(parents=True) + preserved.write_text("keep", encoding="utf-8") + + with mock.patch.object(app_module, "prune_log_files", side_effect=RuntimeError("retention unavailable")): + status = _run(app, home) + + self.assertEqual(status, 1) + self.assertEqual(called, []) + self.assertEqual(list((home / "cache").glob("**/run.json")), []) + self.assertEqual(list((home / "cache").glob("**/primary.log")), []) + self.assertEqual(logging.getLogger("base_cli.startup-retention").handlers, []) + self.assertEqual(preserved.read_text(encoding="utf-8"), "keep") + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_partial_logger_failure_closes_handlers_and_removes_new_bundle(self) -> None: + app = base_cli.App(name="startup-logger") + original_configure_logger = app_module.configure_logger + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + def fail_after_logger_setup(*args: object, **kwargs: object) -> logging.Logger: + original_configure_logger(*args, **kwargs) + raise OSError("logger unavailable") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object(app_module, "configure_logger", side_effect=fail_after_logger_setup): + status = _run(app, home) + + self.assertEqual(status, 1) + self.assertEqual(list((home / "cache").glob("**/run.json")), []) + self.assertEqual(list((home / "cache").glob("**/primary.log")), []) + self.assertEqual(logging.getLogger("base_cli.startup-logger").handlers, []) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_inherited_startup_failure_never_finalizes_or_deletes_parent_bundle(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + parent_run = cache_root / "parent" / "runs" / "parent-run" + parent_run.mkdir(parents=True) + parent_metadata = parent_run / "run.json" + parent_payload = {"run_id": "parent-run", "status": "running"} + parent_metadata.write_text(json.dumps(parent_payload), encoding="utf-8") + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=runtime_layout( + cache_root, + "startup-inherited", + "child-run", + namespace="parent", + inherited_run_root=parent_run, + ), + application_home=None, + runtime_owner="parent", + project_root=None, + project_name=None, + inherited_path=parent_run, + history_parent_run_id="parent-run", + run_id="child-run", + ) + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="startup-inherited", max_log_files=1, profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + with mock.patch.object(app_module, "prune_log_files", side_effect=RuntimeError("retention unavailable")): + status = _run(app, home) + + self.assertEqual(status, 1) + self.assertEqual(json.loads(parent_metadata.read_text(encoding="utf-8")), parent_payload) + self.assertTrue(parent_run.is_dir()) + self.assertFalse((parent_run / "tmp" / "startup-inherited" / "child-run").exists()) + self.assertEqual(logging.getLogger("base_cli.startup-inherited").handlers, []) + + def test_startup_rollback_preserves_preexisting_temp_content(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + layout = runtime_layout(cache_root, "startup-preexisting", "fixed-run") + layout.temp_dir.mkdir(parents=True) + marker = layout.temp_dir / "existing.txt" + marker.write_text("keep", encoding="utf-8") + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="startup-preexisting", + project_root=None, + project_name=None, + inherited_path=None, + history_parent_run_id=None, + run_id="fixed-run", + ) + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="startup-preexisting", max_log_files=1, profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + with mock.patch.object(app_module, "prune_log_files", side_effect=RuntimeError("retention unavailable")): + status = _run(app, home) + + self.assertEqual(status, 1) + self.assertEqual(marker.read_text(encoding="utf-8"), "keep") + self.assertEqual(logging.getLogger("base_cli.startup-preexisting").handlers, []) + + def test_startup_rollback_never_recursively_deletes_temp_outside_run_root(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + external_temp = root / "profile-selected-external-temp" + layout = replace( + runtime_layout(cache_root, "startup-contained", "fixed-run"), + temp_dir=external_temp, + ) + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="startup-contained", + project_root=None, + project_name=None, + inherited_path=None, + history_parent_run_id=None, + run_id="fixed-run", + ) + + marker = external_temp / "created-during-startup.txt" + + def fail_retention(*_args: object) -> None: + marker.write_text("preserve", encoding="utf-8") + raise RuntimeError("retention unavailable") + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="startup-contained", max_log_files=1, profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + with mock.patch.object(app_module, "prune_log_files", side_effect=fail_retention): + status = _run(app, home) + + self.assertEqual(status, 1) + self.assertEqual(marker.read_text(encoding="utf-8"), "preserve") + self.assertEqual(logging.getLogger("base_cli.startup-contained").handlers, []) + + def test_rollback_cleanup_runtime_error_cannot_mask_startup_failure(self) -> None: + app = base_cli.App(name="startup-rollback-runtime", max_log_files=1) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.object( + app_module, + "prune_log_files", + side_effect=RuntimeError("primary startup failure"), + ), mock.patch.object( + app_module.shutil, + "rmtree", + side_effect=RuntimeError("secondary rollback failure"), + ): + with self.assertRaisesRegex(RuntimeError, "primary startup failure"): + with mock.patch.dict( + os.environ, + {"HOME": str(home), "BASE_CLI_CACHE_DIR": str(home / "cache")}, + ): + base_cli.run_app(app, [], reraise_unexpected=True) + + self.assertEqual(logging.getLogger("base_cli.startup-rollback-runtime").handlers, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_logging.py b/tests/test_logging.py index 3a7d52f..e097b2c 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -37,6 +37,20 @@ def test_configure_logger_accepts_custom_formatter(self) -> None: self.assertEqual(stream.getvalue().strip(), "INFO:hello formatter") + def test_base_formatter_includes_exception_tracebacks(self) -> None: + stream = io.StringIO() + logger = base_cli.configure_logger("exception-traceback", None, debug=True, stream=stream) + + try: + raise RuntimeError("diagnostic detail") + except RuntimeError: + logger.debug("unexpected command exception", exc_info=True) + + output = stream.getvalue() + self.assertIn("unexpected command exception", output) + self.assertIn("Traceback", output) + self.assertIn("RuntimeError: diagnostic detail", output) + def test_configure_logger_defaults_to_stderr_and_base_formatter(self) -> None: stream = io.StringIO() diff --git a/tests/test_public_api.py b/tests/test_public_api.py index e52394c..a00e1f7 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -32,6 +32,7 @@ def test_version_resolution_ignores_unrelated_ancestor_version_files(self) -> No def test_facade_exports_supported_modules_functions_and_types(self) -> None: expected = { "CommandProtocolError", + "ConfigurationError", "command_filters", "command_matches", "command_protocol", @@ -47,6 +48,7 @@ def test_facade_exports_supported_modules_functions_and_types(self) -> None: self.assertFalse(hasattr(base_cli, "UserConfig")) self.assertIs(base_cli.command_filters, command_filters) self.assertIs(base_cli.command_protocol, command_protocol) + self.assertTrue(issubclass(base_cli.ConfigurationError, ValueError)) def test_module_all_surfaces_are_explicit(self) -> None: self.assertEqual(