diff --git a/CHANGELOG.md b/CHANGELOG.md index 42f46dd..d6043d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and versions are tracked in the repo-root `VERSION` file. ### Added +- Add `App.attach()` and `base_cli.attach()` for applying one lifecycle to + existing nested, aliased, chained, and lazy Click command trees while + preserving their native callbacks, contexts, and result values. +- Add public `sensitive_parameters` attachment policy for existing and lazy + Click parameters with domain-specific secret names. - Add `ConfigurationError` so consumer profiles can explicitly mark user-correctable configuration messages as safe usage errors. diff --git a/README.md b/README.md index d999abc..1942dde 100644 --- a/README.md +++ b/README.md @@ -230,6 +230,115 @@ hyphens and conventional `_command`, `_cmd`, `_group`, and `_grp` suffixes are removed. Pass an explicit subcommand name when a different public spelling is required. +### Attach an existing Click application + +An established Click command tree can adopt the lifecycle without rebuilding +its commands around `App`: + +```python +import base_cli +import click + + +@click.group() +@click.pass_context +def cli(click_ctx: click.Context) -> None: + click_ctx.ensure_object(dict) + + +@cli.command() +def status() -> None: + ctx = base_cli.get_current_context() + ctx.log.info("checking status") + + +cli = base_cli.attach(cli) + + +if __name__ == "__main__": + raise SystemExit(base_cli.run_app(cli)) +``` + +`attach()` returns the same Click command object. Existing callbacks, +parameters, command and alias names, help text, context settings, result +callbacks, and `click.Context.obj` values keep their Click semantics. The +lifecycle wraps the whole selected invocation once, so nested groups and lazy +`get_command()` implementations are supported without listing or importing +unselected commands. A normal Click callback may keep returning any value its +parent result callback expects; attached commands do not adopt the stricter +`None`-or-integer return contract used by `App.command()` callbacks. +Lifecycle options are installed only on the attached root and should precede +the first subcommand. Existing option declarations, including a vendor-owned +`--version`, are retained instead of being duplicated. Matching lifecycle +declarations must expose one compatible scalar value; callbacks on those Click +parameters are preserved, and their parsed results are validated before +lifecycle startup. + +Pass destination names or option aliases through `sensitive_parameters` when +an existing or lazily loaded Click parameter has a domain-specific name that +does not look secret: + +```python +cli = base_cli.attach( + cli, + sensitive_parameters={"access_code", "--credential"}, +) +``` + +These names are applied to the selected lazy path without enumerating other +commands. Click password prompts configured with `hide_input=True` are treated +as sensitive automatically. + +The module helper creates a generic `App`. Use an explicit app with the same +canonical name as the Click root when the CLI has consumer-specific runtime or +configuration policies: + +```python +lifecycle = base_cli.App(name=cli.name, profile=my_profile) +lifecycle.attach(cli) +``` + +Factories can create application state and services after Click has parsed the +root parameters and before any existing group, command, or result callback +runs: + +```python +def make_application_context(ctx: base_cli.Context) -> ApplicationContext: + return ApplicationContext(environment=ctx.environment) + + +def make_services(ctx: base_cli.Context) -> Services: + services = Services(ctx.config) + ctx.on_cleanup(services.close) + return services + + +cli = base_cli.attach( + cli, + context_factory=make_application_context, + service_factory=make_services, +) +``` + +Their results are available as `ctx.application_context` and `ctx.services`. +The factories receive the active `base_cli.Context`, may register cleanup hooks, +and never replace the existing Click context object. `get_current_context()` is +valid in group callbacks, leaf callbacks, result callbacks, and factory-created +helpers for the duration of the attached invocation. Root Click parameter +callbacks run during initial parsing, before the attached lifecycle is active; +descendant parameter callbacks run inside it as Click dispatches the selected +path. Any root resources registered during pre-parse retain that early entry +timing but close inside the Base lifecycle: the deliberate order is enter +pre-parse resource, enter Base lifecycle, exit pre-parse resource, exit Base +lifecycle. Resources and close hooks registered by either factory also exit +before Base cleanup, so failures are reflected in history and run metadata +while `get_current_context()` is still valid. + +Attach only the highest Click root that should share a lifecycle. A separately +attached child or a native `base_cli.App` command selected beneath an attached +root is rejected before its callbacks run, preventing duplicate lifecycle +boundaries. + ## Options And Arguments `base_cli.option` and `base_cli.argument` mirror Click's decorators: @@ -381,6 +490,10 @@ Important fields include: - `ctx.log_file`: the run's shared `logs/primary.log`, or `None` when persistent logging is disabled. - `ctx.config`: merged configuration dictionary. +- `ctx.application_context`: optional application state returned by an + attachment's `context_factory`, or `None`. +- `ctx.services`: optional services returned by an attachment's + `service_factory`, or `None`. - `ctx.user_config`: opaque consumer-owned user configuration returned by the profile, or `None` for the generic default. - `ctx.environment`: active environment, defaulting to `dev`. diff --git a/lib/python/base_cli/__init__.py b/lib/python/base_cli/__init__.py index cceab32..40e39ac 100644 --- a/lib/python/base_cli/__init__.py +++ b/lib/python/base_cli/__init__.py @@ -34,6 +34,7 @@ def _resolve_version() -> str: from .app import ( App, argument, + attach, command, delegated_display_command, get_command_app, @@ -91,6 +92,7 @@ def _resolve_version() -> str: "render_inspection_json", "testing", "argument", + "attach", "command", "configure_logger", "delegated_display_command", diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index b761b94..9158717 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -7,6 +7,7 @@ import sys import time import traceback +from collections.abc import Iterable from contextvars import ContextVar, Token from dataclasses import dataclass from datetime import datetime @@ -38,14 +39,27 @@ normalize_cli_name, ) from .profile import CliProfile -from .redaction import RedactionPlan, compile_redaction_plan, parameter_name_from_decls, redact_argv +from .redaction import REDACTED, RedactionPlan, compile_redaction_plan, parameter_name_from_decls, redact_argv _STANDARD_OPTION_KEYS = ("debug", "quiet", "environment", "config", "keep_temp", "log_file") _GROUP_STANDARD_OPTIONS_KEY = "base_cli_standard_options" +_ATTACHED_STANDARD_OPTIONS_KEY = object() DISPLAY_COMMAND_ENV = "BASE_CLI_DISPLAY_COMMAND" _INVOCATION_ARGV: ContextVar[list[str] | None] = ContextVar("base_cli_invocation_argv", default=None) +_INVOCATION_MAIN_BYPASS: ContextVar[Any | None] = ContextVar( + "base_cli_invocation_main_bypass", + default=None, +) _COMMAND_APP_ATTRIBUTE = "__base_cli_command_app__" _COMMAND_APP_LOCK = RLock() +_CLICK_ATTACHMENT_ATTRIBUTE = "__base_cli_attachment__" +_CLICK_INSTRUMENTED_ATTRIBUTE = "__base_cli_lifecycle_instrumented__" +_CLICK_MAIN_INSTRUMENTED_ATTRIBUTE = "__base_cli_main_instrumented__" +_CLICK_ORIGINAL_INVOKE_ATTRIBUTE = "__base_cli_original_invoke__" +_CLICK_ORIGINAL_RESOLVE_ATTRIBUTE = "__base_cli_original_resolve__" +_CLICK_ORIGINAL_MAIN_ATTRIBUTE = "__base_cli_original_main__" +_CLICK_APP_OWNER_ATTRIBUTE = "__base_cli_app_owner__" +_CLICK_ATTACHMENT_LOCK = RLock() _REGISTRATION_OPEN = "open" _REGISTRATION_MATERIALIZING = "materializing" _REGISTRATION_FROZEN = "frozen" @@ -54,11 +68,13 @@ @dataclass class _InvocationState: + owner_app: Any = None run_id: str | None = None log_file: Path | None = None debug: bool = False quiet: bool = False options_parsed: bool = False + attached_completion: bool = False @dataclass(frozen=True) @@ -69,7 +85,118 @@ class _SubcommandRegistration: name: str +@dataclass(frozen=True) +class _ClickAttachment: + app: Any + command: Any + context_factory: Callable[[Context], Any] | None + service_factory: Callable[[Context], Any] | None + sensitive_parameters: frozenset[str] + standard_bindings: dict[str, str] + + +class _AttachedInvocation: + """One attachment invocation whose schema is completed lazily.""" + + def __init__( + self, + attachment: _ClickAttachment, + root_click_context: Any, + context: Context, + recorder: RunRecorder, + ) -> None: + self.attachment = attachment + self.root_click_context = root_click_context + self.context = context + self.recorder = recorder + self.redaction_plan = RedactionPlan() + self.invocation_argv: list[str] = [] + self.started = False + self._resolved_children: dict[int, list[tuple[str, Any, Any]]] = {} + self._resolution_parents: dict[int, Any] = {} + self._has_chain = bool(getattr(attachment.command, "chain", False)) + self._selected_boundary_seen = False + + def note_resolution( + self, + parent_context: Any, + command_name: str, + child_command: Any, + ) -> None: + if getattr(getattr(parent_context, "command", None), "chain", False): + self._has_chain = True + self._resolution_parents[id(parent_context)] = parent_context + self._resolved_children.setdefault(id(parent_context), []).append( + (command_name, child_command, None) + ) + + def note_child_context(self, child_context: Any) -> None: + parent = getattr(child_context, "parent", None) + if parent is None: + return + resolutions = self._resolved_children.get(id(parent), []) + for index in range(len(resolutions) - 1, -1, -1): + name, command, recorded_context = resolutions[index] + if recorded_context is None and command is getattr(child_context, "command", None): + resolutions[index] = (name, command, child_context) + break + + def start( + self, + selected_context: Any | None = None, + *, + force: bool = False, + ) -> None: + if self.started: + return + if selected_context is not None: + self._selected_boundary_seen = True + if self._has_chain and not force: + # Click resolves all chain members before invoking the first one. + # Wait until root teardown so every selected command can contribute + # its sensitive option names to the conservative chain scan. + return + # Mark first so a schema failure cannot trigger a second logging + # attempt during teardown and mask the original exception. + self.started = True + opaque_teardown = force and not self._selected_boundary_seen + if opaque_teardown: + self.redaction_plan = RedactionPlan() + elif self._has_chain: + self.redaction_plan = compile_redaction_plan( + self.attachment.command, + self.attachment.sensitive_parameters, + selected_paths=_selected_click_paths( + self.root_click_context, + self._resolved_children, + self._resolution_parents, + ), + ) + else: + selected_path = _selected_click_path( + self.root_click_context, + selected_context, + self._resolved_children, + ) + self.redaction_plan = compile_redaction_plan( + self.attachment.command, + self.attachment.sensitive_parameters, + selected_path=selected_path, + ) + raw_argv = _current_invocation_argv() + self.invocation_argv = ( + [raw_argv[0], *([REDACTED] * (len(raw_argv) - 1))] + if opaque_teardown and raw_argv + else redact_argv(raw_argv, self.redaction_plan) + ) + log_invocation(self.context.log, self.invocation_argv, None) + + _INVOCATION_STATE: ContextVar[_InvocationState | None] = ContextVar("base_cli_invocation_state", default=None) +_ATTACHED_INVOCATION: ContextVar[_AttachedInvocation | None] = ContextVar( + "base_cli_attached_invocation", + default=None, +) def _reset_context_var(variable: ContextVar[Any], token: Any) -> None: @@ -96,9 +223,9 @@ def _warn_lifecycle_failure(context: Context, message: str, exc: BaseException) pass -def _capture_invocation_context(context: Context) -> None: +def _capture_invocation_context(context: Context, owner_app: App) -> None: state = _INVOCATION_STATE.get() - if state is None: + if state is None or state.owner_app is not owner_app: return state.run_id = context.run_id state.log_file = context.log_file @@ -106,18 +233,23 @@ def _capture_invocation_context(context: Context) -> None: state.quiet = context.quiet -def _capture_standard_options(standard: dict[str, Any]) -> None: +def _capture_standard_options(standard: dict[str, Any], owner_app: App) -> None: state = _INVOCATION_STATE.get() - if state is None: + if state is None or state.owner_app is not owner_app: 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: +def _capture_effective_output_options( + *, + owner_app: App, + debug: bool, + quiet: bool, +) -> None: state = _INVOCATION_STATE.get() - if state is None: + if state is None or state.owner_app is not owner_app: return state.debug = debug state.quiet = quiet @@ -284,6 +416,7 @@ def __init__( self._command_kwargs: dict[str, Any] = {} self._subcommands: list[_SubcommandRegistration] = [] self._subcommand_names: set[str] = set() + self._attached_command: Any | None = None @property def name(self) -> str: @@ -390,6 +523,154 @@ def decorator(func: Callable[..., Any]): return decorator + def attach( + self, + command: Any, + *, + context_factory: Callable[[Context], Any] | None = None, + service_factory: Callable[[Context], Any] | None = None, + sensitive_parameters: Iterable[str] = (), + ) -> Any: + """Attach this app's lifecycle to an existing Click command tree. + + The same command object is returned rather than copied. Click continues + to own callbacks, contexts, aliases, and lazy command resolution while + base-cli extends its root parameters and adds one lifecycle boundary. + """ + + click = _require_click() + if not isinstance(command, click.Command): + raise TypeError("App.attach() requires a click.Command instance.") + if context_factory is not None and not callable(context_factory): + raise TypeError("context_factory must be callable or None.") + if service_factory is not None and not callable(service_factory): + raise TypeError("service_factory must be callable or None.") + normalized_sensitive_parameters = _normalize_sensitive_parameters( + sensitive_parameters + ) + + with _CLICK_ATTACHMENT_LOCK, self._registration_lock: + existing = getattr(command, _CLICK_ATTACHMENT_ATTRIBUTE, None) + if isinstance(existing, _ClickAttachment): + if ( + existing.app is self + and existing.command is command + and existing.context_factory is context_factory + and existing.service_factory is service_factory + and existing.sensitive_parameters == normalized_sensitive_parameters + and self._attached_command is command + and self._click_command is command + and self._registration_state == _REGISTRATION_FROZEN + ): + return command + raise RuntimeError( + f"Click command '{getattr(command, 'name', None) or ''}' " + "is already attached to a base_cli.App." + ) + native_owner = getattr(command, _CLICK_APP_OWNER_ATTRIBUTE, None) + if isinstance(native_owner, App): + raise RuntimeError( + f"Click command '{getattr(command, 'name', None) or ''}' " + "already belongs to a native base_cli.App and cannot be attached." + ) + self._ensure_registration_open() + if self._command_func is not None or self._subcommands: + raise RuntimeError( + f"App '{self.name}' already has registered commands and cannot " + "attach an existing Click tree." + ) + if self._attached_command is not None: + raise RuntimeError( + f"App '{self.name}' is already attached to a Click command." + ) + command_name = getattr(command, "name", None) + if not isinstance(command_name, str) or not command_name: + raise RuntimeError("App.attach() requires a named Click command.") + if command_name != self.name: + raise RuntimeError( + f"App '{self.name}' is the authoritative command name; " + f"the attached Click command cannot use '{command_name}'." + ) + + added_parameters: list[Any] = [] + command_was_instrumented = bool( + getattr(command, _CLICK_INSTRUMENTED_ATTRIBUTE, False) + ) + main_was_instrumented = bool( + getattr(command, _CLICK_MAIN_INSTRUMENTED_ATTRIBUTE, False) + ) + missing_marker = object() + previous_marker = getattr( + command, + _CLICK_ATTACHMENT_ATTRIBUTE, + missing_marker, + ) + previous_redaction_plan = self._redaction_plan + previous_attached_command = self._attached_command + previous_click_command = self._click_command + previous_registration_state = self._registration_state + try: + self._registration_state = _REGISTRATION_MATERIALIZING + standard_bindings = _add_attached_standard_options( + click, + command, + version=self.version, + added_parameters=added_parameters, + ) + redaction_plan = compile_redaction_plan( + command, + normalized_sensitive_parameters, + selected_path=(), + ) + attachment = _ClickAttachment( + app=self, + command=command, + context_factory=context_factory, + service_factory=service_factory, + sensitive_parameters=normalized_sensitive_parameters, + standard_bindings=standard_bindings, + ) + _instrument_attached_click_command(click, command) + _instrument_attached_click_main(command) + self._redaction_plan = redaction_plan + self._attached_command = command + self._click_command = command + self._registration_state = _REGISTRATION_FROZEN + # Publish ownership last. Invoke wrappers synchronize on this + # lock, so neither the marker nor partial App state can become + # observable before every attachment invariant is established. + setattr(command, _CLICK_ATTACHMENT_ATTRIBUTE, attachment) + except BaseException: + if previous_marker is missing_marker: + try: + delattr(command, _CLICK_ATTACHMENT_ATTRIBUTE) + except (AttributeError, TypeError): + pass + else: + try: + setattr(command, _CLICK_ATTACHMENT_ATTRIBUTE, previous_marker) + except (AttributeError, TypeError): + pass + if not main_was_instrumented: + _restore_attached_click_main(command) + if not command_was_instrumented: + _restore_attached_click_command(command) + for parameter in added_parameters: + try: + command.params.remove(parameter) + except (AttributeError, ValueError): + pass + object.__setattr__(self, "_redaction_plan", previous_redaction_plan) + object.__setattr__(self, "_attached_command", previous_attached_command) + object.__setattr__(self, "_click_command", previous_click_command) + object.__setattr__( + self, + "_registration_state", + previous_registration_state, + ) + raise + return command + def __call__(self, *args: Any, **kwargs: Any) -> Any: if len(args) < 2 and "prog_name" not in kwargs: kwargs["prog_name"] = self.profile.display_command() or self.name @@ -441,10 +722,12 @@ def _build_click_command(self) -> Any: command_kwargs, )(wrapper) _require_materialized_command_name(command, self.name, self.name) + setattr(command, _CLICK_APP_OWNER_ATTRIBUTE, self) return command group_wrapper = _decorate_standard_options(click, _build_group_wrapper(click), self.version) group = click.group(name=self.name, help=self.help)(group_wrapper) + setattr(group, _CLICK_APP_OWNER_ATTRIBUTE, self) for registration in self._subcommands: wrapper = self._build_command_wrapper(click, registration.func, include_version=False) command = _click_command_decorator( @@ -454,6 +737,7 @@ def _build_click_command(self) -> Any: registration.kwargs, )(wrapper) _require_materialized_command_name(command, registration.name, self.name) + setattr(command, _CLICK_APP_OWNER_ATTRIBUTE, self) # Supplying the canonical name explicitly also prevents a custom # Command implementation from changing the group key between the # validation above and Click's registration step. @@ -473,12 +757,17 @@ def _build_command_wrapper( @functools.wraps(func) def wrapper(**kwargs: Any): + if _ATTACHED_INVOCATION.get() is not None: + raise RuntimeError( + f"base_cli command '{self.name}' cannot run inside an attached " + "Click tree because that would create a second lifecycle." + ) standard = _merge_standard_options( _group_standard_options(click), _pop_standard_options(kwargs), ) _validate_standard_options(click, standard) - _capture_standard_options(standard) + _capture_standard_options(standard, self) started_at = utc_now() started_monotonic_ns = time.monotonic_ns() context: Context | None = None @@ -502,7 +791,7 @@ def wrapper(**kwargs: Any): recorder = RunRecorder(context, started_at, started_monotonic_ns) token = set_current_context(context) - _capture_invocation_context(context) + _capture_invocation_context(context, self) invocation_argv = redact_argv(_current_invocation_argv(), redaction_plan) _start_run_recorder(recorder) log_invocation(context.log, invocation_argv, None) @@ -591,7 +880,11 @@ def _create_context(self, standard: dict[str, Any], dry_run: bool = False) -> Co 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) + _capture_effective_output_options( + owner_app=self, + debug=debug, + quiet=quiet, + ) runtime = self.profile.resolve_runtime(self.name, project) cache_root = runtime.cache_root @@ -719,9 +1012,809 @@ def _rollback_context_creation( pass -def get_command_app(command_func: Callable[..., Any]) -> App: - """Return the isolated :class:`App` owned by ``@base_cli.command``.""" +class _AttachedLifecycleResource: + """Lifecycle resource retained by Click's root Context exit stack.""" + + def __init__( + self, + click: Any, + attachment: _ClickAttachment, + click_context: Any, + standard: dict[str, Any], + ) -> None: + self.click = click + self.attachment = attachment + self.click_context = click_context + self.standard = standard + self.started_at = utc_now() + self.started_monotonic_ns = time.monotonic_ns() + self.context: Context | None = None + self.invocation: _AttachedInvocation | None = None + self.context_token: Any = None + self.invocation_token: Any = None + self.original_click_exit: Callable[..., Any] | None = None + self.click_exit_wrapper: Callable[..., Any] | None = None + self.outcome = outcome_from_exit_code(ExitCode.SUCCESS) + self._closed = False + + def __enter__(self) -> _AttachedLifecycleResource: + try: + try: + context = self.attachment.app._create_context( # pylint: disable=protected-access + self.standard, + dry_run=False, + ) + except ConfigurationError as exc: + raise self.click.UsageError(str(exc)) from exc + except RuntimeDirectoryError as exc: + raise self.click.ClickException(str(exc)) from exc + + self.context = context + self.context_token = set_current_context(context) + _capture_invocation_context(context, self.attachment.app) + recorder = RunRecorder(context, self.started_at, self.started_monotonic_ns) + self.invocation = _AttachedInvocation( + self.attachment, + self.click_context, + context, + recorder, + ) + self.invocation_token = _ATTACHED_INVOCATION.set(self.invocation) + _start_run_recorder(recorder) + + original_click_exit = self.click_context.exit + + @functools.wraps(original_click_exit) + def lifecycle_aware_exit(code: int = 0) -> Any: + # Context.exit() closes the Context before it raises. When it + # is called from a close hook, that recursive close can unwind + # this resource with no exception information, so capture the + # terminal outcome before delegating. + self.record_exception(self.click.exceptions.Exit(code)) + return original_click_exit(code) + + self.original_click_exit = original_click_exit + self.click_exit_wrapper = lifecycle_aware_exit + self.click_context.exit = lifecycle_aware_exit + return self + except BaseException as exc: + if self.context is not None: + self.outcome = outcome_from_exception(self.click, exc) + _record_unexpected_traceback(self.context, self.outcome) + self._finalize() + raise + + def initialize_factories(self) -> None: + """Run extension factories after Click retains this resource. + + A factory may use Click's ``with_resource`` or ``call_on_close`` APIs. + Running it only after our own ``__exit__`` is on Click's stack keeps + those resources inside the base-cli lifecycle boundary. + """ + + context = self.context + if context is None: + raise RuntimeError("The attached lifecycle has not been entered.") + try: + if self.attachment.context_factory is not None: + context.application_context = self.attachment.context_factory(context) + if self.attachment.service_factory is not None: + context.services = self.attachment.service_factory(context) + except ConfigurationError as exc: + error = self.click.UsageError(str(exc)) + self.record_exception(error) + raise error from exc + except RuntimeDirectoryError as exc: + error = self.click.ClickException(str(exc)) + self.record_exception(error) + raise error from exc + except BaseException as exc: + self.record_exception(exc) + raise + + def record_result(self, _result: Any) -> None: + # Arbitrary Click return values are application data, not process exit + # codes. A normally completed attached tree is always successful. + self.outcome = outcome_from_exit_code(ExitCode.SUCCESS) + state = _INVOCATION_STATE.get() + if state is not None and state.owner_app is self.attachment.app: + state.attached_completion = True + + def record_exception(self, exc: BaseException) -> None: + state = _INVOCATION_STATE.get() + if state is not None and state.owner_app is self.attachment.app: + state.attached_completion = False + if self.context is not None: + self.outcome = outcome_from_exception(self.click, exc) + _record_unexpected_traceback(self.context, self.outcome) + + def __exit__( + self, + _exc_type: Any, + exc_value: Any, + _traceback: Any, + ) -> None: + # Click 8.1 closes Context resources without forwarding exception + # details, so the invoke wrapper records the outcome explicitly. + if exc_value is not None and self.outcome.kind == "success": + self.record_exception(exc_value) + self._finalize() + + def _finalize(self) -> None: + if self._closed: + return + self._closed = True + context = self.context + invocation = self.invocation + if context is None: + return + + if invocation is None: + try: + context.cleanup() + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(context, "Lifecycle cleanup failed", exc) + finally: + if self.context_token is not None: + _reset_active_context(context, self.context_token) + return + + try: + if not invocation.started: + invocation.start(force=True) + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(context, "Invocation redaction or logging failed", exc) + + try: + ended_at = utc_now() + ended_monotonic_ns = time.monotonic_ns() + except BaseException as exc: # pylint: disable=broad-exception-caught + ended_at = self.started_at + ended_monotonic_ns = self.started_monotonic_ns + _warn_lifecycle_failure(context, "Terminal clock capture failed", exc) + + try: + if self.attachment.app.profile.history_writer is not None: + self.attachment.app.profile.history_writer( + context, + invocation.invocation_argv, + set(invocation.redaction_plan), + self.started_at, + self.outcome.exit_code, + ) + except BaseException as exc: # pylint: disable=broad-exception-caught + _warn_lifecycle_failure(context, "History finalization failed", exc) + + _finish_run_recorder( + invocation.recorder, + self.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: + if ( + self.original_click_exit is not None + and getattr(self.click_context, "exit", None) is self.click_exit_wrapper + ): + try: + self.click_context.exit = self.original_click_exit + except (AttributeError, TypeError): + pass + if self.invocation_token is not None: + _reset_context_var(_ATTACHED_INVOCATION, self.invocation_token) + if self.context_token is not None: + _reset_active_context(context, self.context_token) + + +def _normalize_sensitive_parameters(values: Iterable[str]) -> frozenset[str]: + if isinstance(values, str): + values = (values,) + try: + normalized = frozenset(values) + except TypeError as exc: + raise TypeError("sensitive_parameters must be an iterable of strings.") from exc + if not all(isinstance(value, str) and value for value in normalized): + raise TypeError("sensitive_parameters must contain only non-empty strings.") + return normalized + + +def _add_attached_standard_options( + click: Any, + command: Any, + *, + version: str | None, + added_parameters: list[Any], +) -> dict[str, str]: + option_specs: tuple[tuple[str, tuple[str, ...], dict[str, Any]], ...] = ( + ( + "log_file", + ("--log-file",), + { + "type": click.Path(dir_okay=False), + "help": "Override the persistent log file.", + }, + ), + ( + "keep_temp", + ("--keep-temp",), + { + "is_flag": True, + "default": None, + "help": "Preserve this run's temp directory.", + }, + ), + ( + "config", + ("--config",), + { + "type": _explicit_config_path_type(click), + "help": "Load an additional config file.", + }, + ), + ("environment", ("--environment",), {"help": "Set the CLI environment."}), + ( + "debug", + ("--debug",), + { + "is_flag": True, + "default": None, + "help": "Enable DEBUG logging on the user-facing stream.", + }, + ), + ( + "quiet", + ("--quiet", "-q"), + { + "is_flag": True, + "default": None, + "help": "Suppress INFO logs on the user-facing stream.", + }, + ), + ) + parameters = getattr(command, "params", None) + if not isinstance(parameters, list): + raise TypeError("Attached Click commands must expose a mutable params list.") + existing_options = [ + parameter + for parameter in parameters + if getattr(parameter, "param_type_name", None) == "option" + ] + context_settings = dict(getattr(command, "context_settings", None) or {}) + token_normalize_func = context_settings.get("token_normalize_func") + used_declarations = { + _normalize_attached_option_declaration( + str(declaration), + token_normalize_func, + ) + for parameter in existing_options + for declaration in ( + *tuple(getattr(parameter, "opts", ())), + *tuple(getattr(parameter, "secondary_opts", ())), + ) + } + bindings: dict[str, str] = {} + bound_existing_parameters: dict[int, str] = {} + + for key, declarations, attrs in option_specs: + normalized_primary = _normalize_attached_option_declaration( + declarations[0], + token_normalize_func, + ) + existing = next( + ( + parameter + for parameter in existing_options + if normalized_primary + in { + _normalize_attached_option_declaration( + str(declaration), + token_normalize_func, + ) + for declaration in ( + *tuple(getattr(parameter, "opts", ())), + *tuple(getattr(parameter, "secondary_opts", ())), + ) + } + ), + None, + ) + if existing is not None: + previous_key = bound_existing_parameters.get(id(existing)) + if previous_key is not None: + raise RuntimeError( + f"Existing Click option combines lifecycle aliases " + f"'{previous_key}' and '{key}' in one parameter; each " + "base-cli lifecycle option must use a distinct parameter." + ) + expected_flag = key in {"debug", "quiet", "keep_temp"} + is_flag = bool( + getattr(existing, "is_flag", False) + or getattr(existing, "count", False) + ) + positive_declarations = { + _normalize_attached_option_declaration( + str(declaration), + token_normalize_func, + ) + for declaration in tuple(getattr(existing, "opts", ())) + } + secondary_declarations = { + _normalize_attached_option_declaration( + str(declaration), + token_normalize_func, + ) + for declaration in tuple(getattr(existing, "secondary_opts", ())) + } + incompatible = ( + is_flag != expected_flag + or not getattr(existing, "expose_value", True) + or bool(getattr(existing, "multiple", False)) + or getattr(existing, "nargs", 1) != 1 + or normalized_primary in secondary_declarations + or ( + expected_flag + and ( + normalized_primary not in positive_declarations + or not bool(getattr(existing, "flag_value", False)) + ) + ) + ) + if incompatible: + raise RuntimeError( + f"Existing '{declarations[0]}' option is incompatible with " + "the base-cli lifecycle option of the same name." + ) + bound_existing_parameters[id(existing)] = key + parameter_name = getattr(existing, "name", None) + if parameter_name: + bindings[key] = str(parameter_name) + continue + + available = tuple( + declaration + for declaration in declarations + if _normalize_attached_option_declaration( + declaration, + token_normalize_func, + ) + not in used_declarations + ) + if not available: + continue + + def capture(click_context: Any, _parameter: Any, value: Any, *, option_key: str = key) -> Any: + values = click_context.meta.setdefault(_ATTACHED_STANDARD_OPTIONS_KEY, {}) + values[option_key] = value + return value + + option_attrs = dict(attrs) + auto_envvar_prefix = context_settings.get("auto_envvar_prefix") + if isinstance(auto_envvar_prefix, str) and auto_envvar_prefix: + option_attrs.setdefault( + "envvar", + f"{auto_envvar_prefix}_{key.upper()}", + ) + option_attrs.update(callback=capture, expose_value=False) + parameter = click.Option( + [*available, f"_base_cli_{key}"], + **option_attrs, + ) + parameters.append(parameter) + added_parameters.append(parameter) + existing_options.append(parameter) + used_declarations.update( + _normalize_attached_option_declaration( + declaration, + token_normalize_func, + ) + for declaration in available + ) + + if ( + version is not None + and _normalize_attached_option_declaration("--version", token_normalize_func) + not in used_declarations + ): + def version_parameter_source() -> None: + return None + + decorated = click.version_option( + version, + "--version", + "_base_cli_version", + )(version_parameter_source) + click_parameters = list(getattr(decorated, "__click_params__", ())) + if not click_parameters: + raise RuntimeError("Click did not create the requested version option.") + parameter = click_parameters[-1] + parameters.append(parameter) + added_parameters.append(parameter) + + return bindings + + +def _normalize_attached_option_declaration( + declaration: str, + normalize: Callable[[str], str] | None, +) -> str: + if normalize is None: + return declaration + first = declaration[:1] + if not first or first.isalnum() or first == "_": + return declaration + prefix = declaration[:2] if declaration[1:2] == first else first + return f"{prefix}{normalize(declaration[len(prefix):])}" + + +def _attached_standard_options( + click_context: Any, + attachment: _ClickAttachment, +) -> dict[str, Any]: + captured = getattr(click_context, "meta", {}).get( + _ATTACHED_STANDARD_OPTIONS_KEY, + {}, + ) + standard: dict[str, Any] = {} + params = getattr(click_context, "params", {}) + for key in _STANDARD_OPTION_KEYS: + parameter_name = attachment.standard_bindings.get(key) + if parameter_name: + standard[key] = params.get(parameter_name) + else: + standard[key] = captured.get(key) + return standard + + +def _validate_attached_standard_values(click: Any, standard: dict[str, Any]) -> None: + for key in ("config", "log_file"): + value = standard.get(key) + if value is None: + continue + try: + raw_path = os.fspath(value) + except TypeError: + raw_path = None + if not isinstance(raw_path, str): + declaration = "--config" if key == "config" else "--log-file" + raise click.UsageError( + f"Existing '{declaration}' option produced an incompatible value; " + "expected a string or path-like object." + ) + environment = standard.get("environment") + if environment is not None and not isinstance(environment, str): + raise click.UsageError( + "Existing '--environment' option produced an incompatible value; " + "expected a string." + ) + + +def _selected_click_path( + root_context: Any, + selected_context: Any | None, + resolved_children: dict[int, list[tuple[str, Any, Any]]], +) -> tuple[tuple[str, Any], ...]: + if selected_context is not None: + contexts: list[Any] = [] + current = selected_context + while current is not None: + contexts.append(current) + if current is root_context: + contexts.reverse() + selected: list[tuple[str, Any]] = [] + for parent, child in zip(contexts, contexts[1:]): + resolutions = resolved_children.get(id(parent), []) + recorded = next( + ( + resolution + for resolution in reversed(resolutions) + if resolution[2] is child + ), + None, + ) + invoked_name = ( + recorded[0] + if recorded is not None + else getattr(child, "info_name", None) + or getattr(child.command, "name", "") + ) + selected.append((str(invoked_name), child.command)) + return tuple(selected) + current = getattr(current, "parent", None) + + path: list[tuple[str, Any]] = [] + parent = root_context + seen: set[int] = set() + while id(parent) not in seen: + seen.add(id(parent)) + resolutions = resolved_children.get(id(parent), []) + if not resolutions: + break + name, command, child_context = resolutions[-1] + path.append((str(name), command)) + if child_context is None: + break + parent = child_context + return tuple(path) + + +def _selected_click_paths( + root_context: Any, + resolved_children: dict[int, list[tuple[str, Any, Any]]], + resolution_parents: dict[int, Any], +) -> tuple[tuple[tuple[str, Any], ...], ...]: + paths: list[tuple[tuple[str, Any], ...]] = [] + seen: set[tuple[tuple[str, int], ...]] = set() + for parent_identity, resolutions in resolved_children.items(): + for name, command, child_context in resolutions: + if child_context is not None: + path = _selected_click_path( + root_context, + child_context, + resolved_children, + ) + else: + parent_context = resolution_parents.get(parent_identity) + parent_path = ( + _selected_click_path( + root_context, + parent_context, + resolved_children, + ) + if parent_context is not None + else () + ) + path = (*parent_path, (name, command)) + identity = tuple((name, id(command)) for name, command in path) + if path and identity not in seen: + paths.append(path) + seen.add(identity) + if not paths: + fallback = _selected_click_path(root_context, None, resolved_children) + if fallback: + paths.append(fallback) + return tuple(paths) + + +def _click_command_has_pending_children(click_context: Any, command: Any) -> bool: + if not callable(getattr(command, "resolve_command", None)): + return False + protected = getattr(click_context, "_protected_args", None) + if protected is None: + protected = getattr(click_context, "protected_args", ()) + return bool(protected or getattr(click_context, "args", ())) + + +def _with_attached_lifecycle_resource( + click_context: Any, + resource: _AttachedLifecycleResource, +) -> None: + # Parameter callbacks can register close hooks while Click parses the root + # context, before Command.invoke gives us a lifecycle boundary. Move those + # already-entered resources into a nested ExitStack so they unwind while + # the base-cli Context is still active and can influence the final outcome. + exit_stack = getattr(click_context, "_exit_stack", None) + pop_all = getattr(exit_stack, "pop_all", None) + if not callable(pop_all): + click_context.with_resource(resource) + resource.initialize_factories() + return + earlier_resources = pop_all() + try: + click_context.with_resource(resource) + finally: + click_context.with_resource(earlier_resources) + resource.initialize_factories() + + +def _instrument_attached_click_command(click: Any, command: Any) -> None: + with _CLICK_ATTACHMENT_LOCK: + if getattr(command, _CLICK_INSTRUMENTED_ATTRIBUTE, False): + return + original_invoke = command.invoke + original_resolve = getattr(command, "resolve_command", None) + + @functools.wraps(original_invoke) + def invoke(click_context: Any) -> Any: + active = _ATTACHED_INVOCATION.get() + with _CLICK_ATTACHMENT_LOCK: + attachment = getattr(command, _CLICK_ATTACHMENT_ATTRIBUTE, None) + if ( + active is not None + and isinstance(attachment, _ClickAttachment) + and attachment is not active.attachment + ): + raise RuntimeError( + f"Click command '{getattr(command, 'name', None) or ''}' " + "is attached to a different base_cli.App and cannot be nested " + "inside another attached tree." + ) + if active is None and isinstance(attachment, _ClickAttachment): + standard = _attached_standard_options(click_context, attachment) + _validate_attached_standard_values(click, standard) + _validate_standard_options(click, standard) + _capture_standard_options(standard, attachment.app) + resource = _AttachedLifecycleResource( + click, + attachment, + click_context, + standard, + ) + _with_attached_lifecycle_resource(click_context, resource) + if not _click_command_has_pending_children(click_context, command): + if resource.invocation is not None: + resource.invocation.start(click_context) + try: + result = original_invoke(click_context) + except BaseException as exc: + resource.record_exception(exc) + raise + resource.record_result(result) + return result + + if active is not None: + active.note_child_context(click_context) + if not _click_command_has_pending_children(click_context, command): + active.start(click_context) + return original_invoke(click_context) + + try: + setattr(command, _CLICK_ORIGINAL_INVOKE_ATTRIBUTE, original_invoke) + setattr(command, _CLICK_ORIGINAL_RESOLVE_ATTRIBUTE, original_resolve) + command.invoke = invoke + + if callable(original_resolve): + + @functools.wraps(original_resolve) + def resolve_command(click_context: Any, args: list[str]) -> Any: + invoked_name = str(args[0]) if args else "" + command_name, child, remaining = original_resolve(click_context, args) + if child is not None: + active = _ATTACHED_INVOCATION.get() + with _CLICK_ATTACHMENT_LOCK: + child_attachment = getattr( + child, + _CLICK_ATTACHMENT_ATTRIBUTE, + None, + ) + child_owner = getattr(child, _CLICK_APP_OWNER_ATTRIBUTE, None) + if ( + active is not None + and isinstance(child_attachment, _ClickAttachment) + and child_attachment is not active.attachment + ): + raise RuntimeError( + f"Click command '{getattr(child, 'name', None) or ''}' " + "is attached to a different base_cli.App and cannot be nested " + "inside another attached tree." + ) + if active is not None and isinstance(child_owner, App): + raise RuntimeError( + f"Click command '{getattr(child, 'name', None) or ''}' " + "already belongs to a native base_cli.App and cannot be nested " + "inside an attached tree because that would create a second lifecycle." + ) + _instrument_attached_click_command(click, child) + if active is not None: + active.note_resolution( + click_context, + invoked_name or str(command_name), + child, + ) + return command_name, child, remaining + + command.resolve_command = resolve_command + + setattr(command, _CLICK_INSTRUMENTED_ATTRIBUTE, True) + except BaseException: + _restore_attached_click_command(command) + raise + + +def _restore_attached_click_command(command: Any) -> None: + original_invoke = getattr(command, _CLICK_ORIGINAL_INVOKE_ATTRIBUTE, None) + original_resolve = getattr(command, _CLICK_ORIGINAL_RESOLVE_ATTRIBUTE, None) + if original_invoke is not None: + try: + command.invoke = original_invoke + except (AttributeError, TypeError): + pass + if callable(original_resolve): + try: + command.resolve_command = original_resolve + except (AttributeError, TypeError): + pass + for attribute in ( + _CLICK_INSTRUMENTED_ATTRIBUTE, + _CLICK_ORIGINAL_INVOKE_ATTRIBUTE, + _CLICK_ORIGINAL_RESOLVE_ATTRIBUTE, + ): + try: + delattr(command, attribute) + except (AttributeError, TypeError): + pass + + +def _instrument_attached_click_main(command: Any) -> None: + if getattr(command, _CLICK_MAIN_INSTRUMENTED_ATTRIBUTE, False): + return + original_main = command.main + + @functools.wraps(original_main) + def main(*args: Any, **kwargs: Any) -> Any: + if _INVOCATION_MAIN_BYPASS.get() is command: + bypass_token = _INVOCATION_MAIN_BYPASS.set(None) + try: + return original_main(*args, **kwargs) + finally: + _reset_context_var(_INVOCATION_MAIN_BYPASS, bypass_token) + explicit_args = kwargs.get("args", args[0] if args else None) + prog_name = kwargs.get("prog_name", args[1] if len(args) > 1 else None) + if explicit_args is None: + invocation_argv = list(sys.argv) + else: + materialized_args = list(explicit_args) + invocation_argv = [ + prog_name or getattr(command, "name", None) or "cli", + *materialized_args, + ] + if "args" in kwargs or not args: + kwargs = {**kwargs, "args": materialized_args} + else: + args = (materialized_args, *args[1:]) + token = _INVOCATION_ARGV.set(invocation_argv) + try: + return original_main(*args, **kwargs) + finally: + _reset_context_var(_INVOCATION_ARGV, token) + + try: + setattr(command, _CLICK_ORIGINAL_MAIN_ATTRIBUTE, original_main) + command.main = main + setattr(command, _CLICK_MAIN_INSTRUMENTED_ATTRIBUTE, True) + except BaseException: + _restore_attached_click_main(command) + raise + + +def _restore_attached_click_main(command: Any) -> None: + original_main = getattr(command, _CLICK_ORIGINAL_MAIN_ATTRIBUTE, None) + if original_main is not None: + try: + command.main = original_main + except (AttributeError, TypeError): + pass + for attribute in ( + _CLICK_MAIN_INSTRUMENTED_ATTRIBUTE, + _CLICK_ORIGINAL_MAIN_ATTRIBUTE, + ): + try: + delattr(command, attribute) + except (AttributeError, TypeError): + pass + +def get_command_app(command_func: Callable[..., Any]) -> App: + """Return the :class:`App` owning a registered function or attached tree.""" + + with _CLICK_ATTACHMENT_LOCK: + attachment = getattr(command_func, _CLICK_ATTACHMENT_ATTRIBUTE, None) + if ( + isinstance(attachment, _ClickAttachment) + and attachment.command is command_func + and isinstance(attachment.app, App) + ): + owner = attachment.app + with owner._registration_lock: # pylint: disable=protected-access + if ( + owner._attached_command is command_func # pylint: disable=protected-access + and owner._click_command is command_func # pylint: disable=protected-access + and owner._registration_state == _REGISTRATION_FROZEN # pylint: disable=protected-access + ): + return owner with _COMMAND_APP_LOCK: owner = getattr(command_func, _COMMAND_APP_ATTRIBUTE, None) if isinstance(owner, App): @@ -729,7 +1822,67 @@ def get_command_app(command_func: Callable[..., Any]) -> App: if owner._command_func is command_func: # pylint: disable=protected-access return owner raise TypeError( - "Expected a base_cli.App or a function registered with @base_cli.command()." + "Expected a base_cli.App, an attached Click command, or a function " + "registered with @base_cli.command()." + ) + + +def attach( + command: Any, + *, + app: App | None = None, + context_factory: Callable[[Context], Any] | None = None, + service_factory: Callable[[Context], Any] | None = None, + sensitive_parameters: Iterable[str] = (), + **app_kwargs: Any, +) -> Any: + """Attach lifecycle middleware and return the same Click command object. + + Attachment ownership, factories, and sensitivity policy are immutable; + repeating the same attachment (or omitting its existing policy through + this helper) is idempotent. + """ + + normalized_sensitive_parameters = _normalize_sensitive_parameters( + sensitive_parameters + ) + existing = getattr(command, _CLICK_ATTACHMENT_ATTRIBUTE, None) + if app is not None and app_kwargs: + unexpected = ", ".join(sorted(app_kwargs)) + raise TypeError( + f"App constructor arguments cannot be used with app= ({unexpected})." + ) + if isinstance(existing, _ClickAttachment) and ( + app is None or app is existing.app + ): + existing_app = get_command_app(command) + if app_kwargs: + unexpected = ", ".join(sorted(app_kwargs)) + raise TypeError( + f"Click command is already attached; app arguments cannot be changed ({unexpected})." + ) + if not normalized_sensitive_parameters: + normalized_sensitive_parameters = existing.sensitive_parameters + if ( + context_factory is None + and service_factory is None + and normalized_sensitive_parameters == existing.sensitive_parameters + ): + return command + app = existing_app + if app is None: + command_name = getattr(command, "name", None) + if not isinstance(command_name, str) or not command_name: + raise TypeError("attach() requires a named Click command.") + name = app_kwargs.pop("name", None) or command_name + app = App(name=name, **app_kwargs) + if not isinstance(app, App): + raise TypeError("app must be a base_cli.App instance or None.") + return app.attach( + command, + context_factory=context_factory, + service_factory=service_factory, + sensitive_parameters=normalized_sensitive_parameters, ) @@ -739,7 +1892,7 @@ def run_app( *, reraise_unexpected: bool = False, ) -> int: - """Run an :class:`App` or registered command and return its process exit code.""" + """Run an App, registered command, or attached Click tree and return its status.""" if not isinstance(app, App): app = get_command_app(app) @@ -753,19 +1906,28 @@ def run_app( 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 = _InvocationState( + owner_app=app, + debug=leading_debug, + quiet=leading_quiet, + ) state_token = _INVOCATION_STATE.set(state) try: try: display_command = app.profile.display_command() invocation_argv = _effective_invocation_argv(app, args, explicit_argv, display_command) + command = app.click_command invocation_token = _INVOCATION_ARGV.set(invocation_argv) try: - result = app.click_command.main( - args=args, - prog_name=display_command or app.name, - standalone_mode=False, - ) + bypass_token = _INVOCATION_MAIN_BYPASS.set(command) + try: + result = command.main( + args=args, + prog_name=display_command or app.name, + standalone_mode=False, + ) + finally: + _reset_context_var(_INVOCATION_MAIN_BYPASS, bypass_token) finally: _reset_context_var(_INVOCATION_ARGV, invocation_token) except click.Abort as exc: @@ -798,6 +1960,8 @@ def run_app( return ExitCode.FAILURE try: + if state.attached_completion: + return ExitCode.SUCCESS return _normalize_command_result(result) except TypeError as exc: print(f"ERROR: {exc}", file=sys.stderr) diff --git a/lib/python/base_cli/context.py b/lib/python/base_cli/context.py index 3897713..039a010 100644 --- a/lib/python/base_cli/context.py +++ b/lib/python/base_cli/context.py @@ -51,6 +51,8 @@ class Context: runtime_owner: str = "default" owner_root: Path | None = None run_root: Path | None = None + application_context: Any = field(default=None, repr=False, compare=False) + services: Any = field(default=None, repr=False, compare=False) _run_metadata_path: Path | None = field(default=None, init=False, repr=False, compare=False) _owns_temp_dir: bool = field(default=False, init=False, repr=False, compare=False) _owned_temp_identity: tuple[int, int] | None = field(default=None, init=False, repr=False, compare=False) diff --git a/lib/python/base_cli/redaction.py b/lib/python/base_cli/redaction.py index 660589b..66504fc 100644 --- a/lib/python/base_cli/redaction.py +++ b/lib/python/base_cli/redaction.py @@ -35,6 +35,7 @@ class _CommandSpec: allow_interspersed_args: bool ignore_unknown_options: bool token_normalize_func: Callable[[str], str] | None + chain: bool @dataclass(frozen=True) @@ -97,7 +98,13 @@ def parameter_name_from_decls(param_decls: tuple[str, ...]) -> str: return possible_names[0][1].replace("-", "_").lower() -def compile_redaction_plan(command: Any, sensitive_names: Iterable[str] = ()) -> RedactionPlan: +def compile_redaction_plan( + command: Any, + sensitive_names: Iterable[str] = (), + *, + selected_path: Sequence[tuple[str, Any]] | None = None, + selected_paths: Sequence[Sequence[tuple[str, Any]]] | None = None, +) -> RedactionPlan: """Compile a recursively Click-aware redaction plan for ``command``. Parameters may be marked with the private ``_base_cli_sensitive`` flag. @@ -106,6 +113,15 @@ def compile_redaction_plan(command: Any, sensitive_names: Iterable[str] = ()) -> protected automatically as defense in depth. """ + if selected_path is not None and selected_paths is not None: + raise TypeError("selected_path and selected_paths are mutually exclusive.") + effective_paths = ( + (tuple(selected_path),) + if selected_path is not None + else tuple(tuple(path) for path in selected_paths) + if selected_paths is not None + else None + ) explicit = _sensitive_forms(sensitive_names) compatible_names: set[str] = set(sensitive_names) root = _compile_command( @@ -114,6 +130,8 @@ def compile_redaction_plan(command: Any, sensitive_names: Iterable[str] = ()) -> compatible_names, active=set(), inherited_token_normalize_func=None, + selected_paths=effective_paths, + force_no_interspersed=False, ) return RedactionPlan(compatible_names, root=root) @@ -141,6 +159,8 @@ def _compile_command( *, active: set[int], inherited_token_normalize_func: Callable[[str], str] | None, + selected_paths: tuple[tuple[tuple[str, Any], ...], ...] | None, + force_no_interspersed: bool, ) -> _CommandSpec: identity = id(command) if identity in active: @@ -152,6 +172,7 @@ def _compile_command( allow_interspersed_args=True, ignore_unknown_options=False, token_normalize_func=inherited_token_normalize_func, + chain=False, ) active.add(identity) try: @@ -176,10 +197,14 @@ def _compile_command( if is_option else () ) - sensitive = bool(getattr(parameter, "_base_cli_sensitive", False)) or _parameter_is_sensitive( - name, - raw_aliases, - explicit, + sensitive = ( + bool(getattr(parameter, "_base_cli_sensitive", False)) + or bool(getattr(parameter, "hide_input", False)) + or _parameter_is_sensitive( + name, + raw_aliases, + explicit, + ) ) nargs = _parameter_nargs(parameter) if raw_aliases: @@ -201,21 +226,51 @@ def _compile_command( compatible_names.add(name) subcommands: dict[str, _CommandSpec] = {} - raw_subcommands = getattr(command, "commands", None) - if isinstance(raw_subcommands, Mapping): - for command_name, child in raw_subcommands.items(): - if child is not None: - subcommands[str(command_name)] = _compile_command( - child, - explicit, - compatible_names, - active=active, - inherited_token_normalize_func=token_normalize_func, + chain = bool(getattr(command, "chain", False)) + if selected_paths is not None: + selected_children: dict[str, tuple[Any, list[tuple[tuple[str, Any], ...]]]] = {} + for path in selected_paths: + if not path: + continue + command_name, child = path[0] + key = str(command_name) + if key not in selected_children: + selected_children[key] = (child, []) + elif selected_children[key][0] is not child: + raise RuntimeError( + f"Selected command name '{key}' resolved to multiple Click commands." ) + selected_children[key][1].append(path[1:]) + for command_name, (child, child_paths) in selected_children.items(): + subcommands[command_name] = _compile_command( + child, + explicit, + compatible_names, + active=active, + inherited_token_normalize_func=token_normalize_func, + selected_paths=tuple(child_paths), + force_no_interspersed=chain, + ) + else: + raw_subcommands = getattr(command, "commands", None) + if isinstance(raw_subcommands, Mapping): + for command_name, child in raw_subcommands.items(): + if child is not None: + subcommands[str(command_name)] = _compile_command( + child, + explicit, + compatible_names, + active=active, + inherited_token_normalize_func=token_normalize_func, + selected_paths=None, + force_no_interspersed=chain, + ) allow_interspersed_args = context_settings.get("allow_interspersed_args") if allow_interspersed_args is None: allow_interspersed_args = getattr(command, "allow_interspersed_args", True) + if force_no_interspersed: + allow_interspersed_args = False ignore_unknown_options = context_settings.get("ignore_unknown_options") if ignore_unknown_options is None: ignore_unknown_options = getattr(command, "ignore_unknown_options", False) @@ -227,6 +282,7 @@ def _compile_command( allow_interspersed_args=bool(allow_interspersed_args), ignore_unknown_options=bool(ignore_unknown_options), token_normalize_func=token_normalize_func, + chain=chain, ) finally: active.remove(identity) @@ -315,7 +371,7 @@ def _redact_command( token_indices: list[int], command: _CommandSpec, result: list[str], -) -> None: +) -> list[int]: positional_indices = _scan_options(argv, token_indices, command, result) argument_spans, extra_indices = _allocate_argument_spans(positional_indices, command.arguments) for argument, span in zip(command.arguments, argument_spans): @@ -324,14 +380,21 @@ def _redact_command( result[index] = REDACTED if not command.subcommands or not extra_indices: - return - command_index = extra_indices[0] - command_name = argv[command_index] - child = command.subcommands.get(command_name) - if child is None and command.token_normalize_func is not None: - child = command.subcommands.get(command.token_normalize_func(command_name)) - if child is not None: - _redact_command(argv, extra_indices[1:], child, result) + return extra_indices + + remaining = extra_indices + while remaining: + command_index = remaining[0] + command_name = argv[command_index] + child = command.subcommands.get(command_name) + if child is None and command.token_normalize_func is not None: + child = command.subcommands.get(command.token_normalize_func(command_name)) + if child is None: + return remaining + remaining = _redact_command(argv, remaining[1:], child, result) + if not command.chain: + return remaining + return remaining def _scan_options( diff --git a/tests/test_click_tree_attachment.py b/tests/test_click_tree_attachment.py new file mode 100644 index 0000000..44a5bca --- /dev/null +++ b/tests/test_click_tree_attachment.py @@ -0,0 +1,1728 @@ +from __future__ import annotations + +import importlib.util +import json +import os +import tempfile +import unittest +from dataclasses import replace +from pathlib import Path +from typing import Any +from unittest import mock + +import base_cli +from base_cli.testing import invoke + + +def _option_count(command: Any, declaration: str) -> int: + return sum( + declaration in tuple(getattr(parameter, "opts", ())) + for parameter in command.params + if getattr(parameter, "param_type_name", None) == "option" + ) + + +def _all_output(result: Any) -> str: + output = result.output + try: + stderr = result.stderr + except ValueError: + stderr = "" + if stderr and stderr not in output: + return f"{output}{stderr}" + return output + + +class _CountingApp(base_cli.App): + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.context_create_count = 0 + self.context_cleanup_count = 0 + self.created_contexts: list[base_cli.Context] = [] + + def _create_context( + self, + standard: dict[str, Any], + dry_run: bool = False, + ) -> base_cli.Context: + self.context_create_count += 1 + context = super()._create_context(standard, dry_run=dry_run) + self.created_contexts.append(context) + original_cleanup = context.cleanup + + def count_cleanup() -> None: + self.context_cleanup_count += 1 + original_cleanup() + + context.cleanup = count_cleanup # type: ignore[method-assign] + return context + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class ClickTreeAttachmentTests(unittest.TestCase): + def test_prebuilt_single_command_preserves_click_contract_and_lifecycle(self) -> None: + import click + + seen: dict[str, Any] = {} + cleanup_calls: list[None] = [] + context_settings = { + "help_option_names": ["-h", "--help"], + "token_normalize_func": str.casefold, + } + + @click.command( + name="vendor-sync", + help="Synchronize a vendor workspace.", + epilog="Provided by the vendor package.", + context_settings=context_settings, + ) + @click.version_option("9.8.7", prog_name="vendor-sync") + @click.option( + "-m", + "--mode", + type=click.Choice(["fast", "safe"], case_sensitive=False), + required=True, + help="Select the vendor mode.", + ) + @click.argument("target") + @click.pass_context + def vendor_sync(click_context: Any, mode: str, target: str) -> None: + context = base_cli.get_current_context() + context.on_cleanup(lambda: cleanup_calls.append(None)) + seen.update( + click_context=click_context, + context=context, + mode=mode, + target=target, + ) + + original_parameters = tuple(vendor_sync.params) + original_parameter_state = [ + ( + parameter, + parameter.name, + tuple(getattr(parameter, "opts", ())), + getattr(parameter, "help", None), + getattr(parameter, "required", False), + ) + for parameter in original_parameters + ] + original_context_settings = dict(vendor_sync.context_settings or {}) + app = _CountingApp( + name="vendor-sync", + version="1.2.3", + log_to_file=False, + ) + + attached = app.attach(vendor_sync) + + self.assertIs(attached, vendor_sync) + self.assertIs(app.click_command, vendor_sync) + self.assertEqual(vendor_sync.name, "vendor-sync") + self.assertEqual(vendor_sync.help, "Synchronize a vendor workspace.") + self.assertEqual(vendor_sync.epilog, "Provided by the vendor package.") + self.assertEqual(vendor_sync.context_settings, original_context_settings) + for parameter, name, declarations, help_text, required in original_parameter_state: + self.assertTrue(any(candidate is parameter for candidate in vendor_sync.params)) + self.assertEqual(parameter.name, name) + self.assertEqual(tuple(getattr(parameter, "opts", ())), declarations) + self.assertEqual(getattr(parameter, "help", None), help_text) + self.assertEqual(getattr(parameter, "required", False), required) + self.assertEqual(_option_count(vendor_sync, "--version"), 1) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + result = invoke( + app, + ["--MODE", "FAST", "workspace"], + home=home, + ) + help_result = invoke(app, ["-h"], home=home) + version_result = invoke(app, ["--version"], home=home) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(seen["mode"], "fast") + self.assertEqual(seen["target"], "workspace") + self.assertIs(seen["click_context"].command, vendor_sync) + self.assertEqual(seen["context"].cli_name, "vendor-sync") + self.assertEqual(cleanup_calls, [None]) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + self.assertEqual(help_result.exit_code, 0, help_result.output) + self.assertIn("Synchronize a vendor workspace.", help_result.output) + self.assertIn("Provided by the vendor package.", help_result.output) + self.assertIn("Select the vendor mode.", help_result.output) + self.assertEqual(version_result.exit_code, 0, version_result.output) + self.assertIn("vendor-sync, version 9.8.7", version_result.output) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_nested_groups_preserve_dispatch_and_arbitrary_result_callbacks(self) -> None: + import click + + events: list[tuple[str, Any]] = [] + active_run_ids: list[tuple[str, str]] = [] + + @click.group(name="workspace", help="Manage workspaces.") + def root() -> None: + active_run_ids.append(("root", base_cli.get_current_context().run_id)) + events.append(("root", None)) + + @root.group(name="admin", help="Administrative commands.") + def admin() -> None: + active_run_ids.append(("admin", base_cli.get_current_context().run_id)) + events.append(("admin", None)) + + @admin.command(name="deploy", help="Deploy a target.") + @click.option("--target", required=True) + def deploy(target: str) -> dict[str, str]: + context = base_cli.get_current_context() + active_run_ids.append(("deploy", context.run_id)) + events.append(("deploy", (target, context.run_id))) + return {"target": target} + + @admin.result_callback() + def finish_admin(result: dict[str, str]) -> tuple[str, dict[str, str]]: + active_run_ids.append(("admin-result", base_cli.get_current_context().run_id)) + events.append(("admin-result", result)) + return ("admin", result) + + @root.result_callback() + def finish_root(result: tuple[str, dict[str, str]]) -> int: + active_run_ids.append(("root-result", base_cli.get_current_context().run_id)) + events.append(("root-result", result)) + return 0 + + root_callback = root.callback + admin_callback = admin.callback + deploy_parameters = tuple(deploy.params) + app = _CountingApp(name="workspace", log_to_file=False) + + self.assertIs(app.attach(root), root) + self.assertIs(app.click_command, root) + self.assertIs(root.commands["admin"], admin) + self.assertIs(admin.commands["deploy"], deploy) + self.assertIs(root.callback, root_callback) + self.assertIs(admin.callback, admin_callback) + self.assertTrue(all( + any(candidate is parameter for candidate in deploy.params) + for parameter in deploy_parameters + )) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + ["--environment", "stage", "admin", "deploy", "--target", "prod"], + home=Path(tmpdir), + ) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + [name for name, _value in events], + ["root", "admin", "deploy", "admin-result", "root-result"], + ) + self.assertEqual(events[2][1][0], "prod") + self.assertEqual(events[3], ("admin-result", {"target": "prod"})) + self.assertEqual(events[4], ("root-result", ("admin", {"target": "prod"}))) + self.assertEqual( + [name for name, _run_id in active_run_ids], + ["root", "admin", "deploy", "admin-result", "root-result"], + ) + self.assertEqual(len({run_id for _name, run_id in active_run_ids}), 1) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_aliases_sharing_one_command_are_instrumented_once(self) -> None: + import click + + invocations: list[tuple[str, str]] = [] + cleanup_calls: list[str] = [] + + @click.command(name="synchronize") + @click.pass_context + def synchronize(click_context: Any) -> None: + context = base_cli.get_current_context() + alias = str(click_context.info_name) + invocations.append((alias, context.run_id)) + context.on_cleanup(lambda: cleanup_calls.append(alias)) + + root = click.Group(name="aliases") + root.add_command(synchronize, name="sync") + root.add_command(synchronize, name="ship") + app = _CountingApp(name="aliases", log_to_file=False) + + app.attach(root) + callback_after_attachment = synchronize.callback + + self.assertIs(root.commands["sync"], root.commands["ship"]) + self.assertIs(app.click_command, root) + self.assertIs(synchronize.callback, callback_after_attachment) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + sync_result = invoke(app, ["sync"], home=home) + ship_result = invoke(app, ["ship"], home=home) + + self.assertEqual(sync_result.exit_code, 0, sync_result.output) + self.assertEqual(ship_result.exit_code, 0, ship_result.output) + self.assertEqual([alias for alias, _run_id in invocations], ["sync", "ship"]) + self.assertNotEqual(invocations[0][1], invocations[1][1]) + self.assertEqual(cleanup_calls, ["sync", "ship"]) + self.assertEqual(app.context_create_count, 2) + self.assertEqual(app.context_cleanup_count, 2) + self.assertIs(synchronize.callback, callback_after_attachment) + + def test_standard_options_are_added_only_once_at_the_root(self) -> None: + import click + + seen: dict[str, Any] = {} + + @click.group(name="standard-options") + @click.option( + "--debug", + is_flag=True, + help="Enable the vendor's debug behavior.", + ) + def root(debug: bool) -> None: + seen["vendor_debug"] = debug + context = base_cli.get_current_context() + seen["root_base_debug"] = context.debug + seen["root_environment"] = context.environment + + @root.group(name="nested") + def nested() -> None: + pass + + @nested.command(name="status") + def status() -> None: + seen["base_debug"] = base_cli.get_current_context().debug + + debug_parameter = next( + parameter for parameter in root.params if "--debug" in getattr(parameter, "opts", ()) + ) + app = _CountingApp( + name="standard-options", + version="3.2.1", + log_to_file=False, + ) + + app.attach(root) + self.assertIs(app.click_command, root) + self.assertIs(app.click_command, root) + + for declaration in ( + "--debug", + "--quiet", + "--environment", + "--config", + "--keep-temp", + "--log-file", + "--version", + ): + self.assertEqual( + _option_count(root, declaration), + 1, + f"expected one root option for {declaration}", + ) + self.assertTrue(any(parameter is debug_parameter for parameter in root.params)) + self.assertEqual(debug_parameter.help, "Enable the vendor's debug behavior.") + for command in (nested, status): + for declaration in ( + "--debug", + "--quiet", + "--environment", + "--config", + "--keep-temp", + "--log-file", + "--version", + ): + self.assertEqual(_option_count(command, declaration), 0) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + result = invoke( + app, + ["--debug", "--environment", "stage", "nested", "status"], + home=home, + ) + help_result = invoke(app, ["--help"], home=home) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertTrue(seen["vendor_debug"]) + self.assertTrue(seen["root_base_debug"]) + self.assertEqual(seen["root_environment"], "stage") + self.assertTrue(seen["base_debug"]) + self.assertEqual(help_result.exit_code, 0, help_result.output) + for declaration in ( + "--debug", + "--quiet", + "--environment", + "--config", + "--keep-temp", + "--log-file", + ): + self.assertEqual(help_result.output.count(declaration), 1) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_attached_standard_options_do_not_collide_with_vendor_meta(self) -> None: + import click + + seen: list[tuple[str | None, bool, Any]] = [] + + def set_vendor_meta( + click_context: Any, + _parameter: Any, + value: str, + ) -> str: + click_context.meta["base_cli_standard_options"] = "vendor-owned" + return value + + @click.command(name="meta-safe") + @click.option("--vendor", callback=set_vendor_meta) + @click.pass_context + def command(click_context: Any, vendor: str | None) -> None: + context = base_cli.get_current_context() + seen.append( + ( + vendor, + context.debug, + click_context.meta["base_cli_standard_options"], + ) + ) + + app = _CountingApp(name="meta-safe", log_to_file=False) + app.attach(command) + + with tempfile.TemporaryDirectory() as tmpdir: + vendor_first = invoke( + app, + ["--vendor", "kept", "--debug"], + home=Path(tmpdir), + ) + base_first = invoke( + app, + ["--debug", "--vendor", "also-kept"], + home=Path(tmpdir), + ) + + self.assertEqual(vendor_first.exit_code, 0, vendor_first.output) + self.assertEqual(base_first.exit_code, 0, base_first.output) + self.assertEqual( + seen, + [ + ("kept", True, "vendor-owned"), + ("also-kept", True, "vendor-owned"), + ], + ) + self.assertEqual(app.context_create_count, 2) + self.assertEqual(app.context_cleanup_count, 2) + + def test_injected_version_uses_click_invocation_alias(self) -> None: + import click + from click.testing import CliRunner + + command = click.Command(name="canonical-name", callback=lambda: None) + app = _CountingApp( + name="canonical-name", + version="4.5.6", + log_to_file=False, + ) + app.attach(command) + + result = CliRunner().invoke( + command, + ["--version"], + prog_name="alias-bin", + ) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("alias-bin, version 4.5.6", result.output) + self.assertNotIn("canonical-name, version", result.output) + self.assertEqual(app.context_create_count, 0) + self.assertEqual(app.context_cleanup_count, 0) + + def test_semantically_inverted_standard_flags_are_rejected(self) -> None: + import click + + reversed_declaration = click.Command( + name="reversed-debug", + callback=lambda **_kwargs: None, + params=[click.Option(["--no-debug/--debug"], default=True)], + ) + false_flag_value = click.Command( + name="false-debug", + callback=lambda **_kwargs: None, + params=[ + click.Option( + ["--debug"], + is_flag=True, + flag_value=False, + default=True, + ) + ], + ) + + with self.assertRaisesRegex(RuntimeError, "--debug.*incompatible"): + base_cli.App(name="reversed-debug").attach(reversed_declaration) + with self.assertRaisesRegex(RuntimeError, "--debug.*incompatible"): + base_cli.App(name="false-debug").attach(false_flag_value) + + def test_one_vendor_parameter_cannot_supply_two_lifecycle_options(self) -> None: + import click + + command = click.Command( + name="ambiguous-options", + callback=lambda **_kwargs: None, + params=[ + click.Option( + ["--environment", "--config"], + type=str, + ) + ], + ) + + with self.assertRaisesRegex(RuntimeError, "combines lifecycle aliases"): + base_cli.App(name="ambiguous-options").attach(command) + + def test_attach_rejects_invalid_or_already_registered_inputs(self) -> None: + import click + + app = base_cli.App(name="invalid-attachment") + with self.assertRaisesRegex(TypeError, "click.Command"): + app.attach(object()) + + registered_app = base_cli.App(name="registered") + + @registered_app.command() + def registered(context: base_cli.Context) -> None: + del context + + external = click.Command(name="external", callback=lambda: None) + with self.assertRaisesRegex(RuntimeError, "registered commands|cannot attach"): + registered_app.attach(external) + + def test_attach_is_idempotent_only_for_the_same_app_and_factories(self) -> None: + import click + + command = click.Command(name="ownership", callback=lambda: None) + app = base_cli.App(name="ownership", log_to_file=False) + + def context_factory(context: base_cli.Context) -> object: + return context + + def service_factory(context: base_cli.Context) -> object: + return context + + self.assertIs( + app.attach( + command, + context_factory=context_factory, + service_factory=service_factory, + ), + command, + ) + self.assertIs( + app.attach( + command, + context_factory=context_factory, + service_factory=service_factory, + ), + command, + ) + + with self.assertRaisesRegex(RuntimeError, "already attached"): + app.attach(command, context_factory=lambda context: context) + with self.assertRaisesRegex(RuntimeError, "already attached"): + base_cli.App(name="other-owner").attach(command) + + def test_attach_publication_failure_rolls_back_and_can_retry(self) -> None: + import click + + class FailOnceApp(_CountingApp): + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.fail_attachment_publication = False + super().__init__(*args, **kwargs) + self.fail_attachment_publication = True + + def __setattr__(self, name: str, value: Any) -> None: + if ( + name == "_click_command" + and value is not None + and getattr(self, "fail_attachment_publication", False) + ): + object.__setattr__(self, "fail_attachment_publication", False) + raise RuntimeError("simulated publication failure") + super().__setattr__(name, value) + + command = click.Command( + name="retry-attachment", + callback=lambda: None, + params=[click.Option(["--vendor"])], + ) + original_parameters = tuple(command.params) + original_invoke = command.invoke + original_main = command.main + app = FailOnceApp(name="retry-attachment", log_to_file=False) + + with self.assertRaisesRegex(RuntimeError, "publication failure"): + app.attach(command) + + self.assertEqual(tuple(command.params), original_parameters) + self.assertEqual(command.invoke, original_invoke) + self.assertEqual(command.main, original_main) + self.assertFalse(hasattr(command, "__base_cli_attachment__")) + self.assertFalse(hasattr(command, "__base_cli_lifecycle_instrumented__")) + self.assertFalse(hasattr(command, "__base_cli_main_instrumented__")) + self.assertIsNone(app._attached_command) # pylint: disable=protected-access + self.assertIsNone(app._click_command) # pylint: disable=protected-access + + self.assertIs(app.attach(command), command) + self.assertIs(app.click_command, command) + + def test_factory_failure_still_cleans_and_resets_the_active_context(self) -> None: + import click + + class FactoryFailure(RuntimeError): + pass + + cleanup_calls: list[base_cli.Context] = [] + callback_calls: list[None] = [] + failure = FactoryFailure("factory unavailable") + + @click.command(name="factory-failure") + def command() -> None: + callback_calls.append(None) + + def context_factory(context: base_cli.Context) -> object: + self.assertIs(base_cli.get_current_context(), context) + context.on_cleanup(lambda: cleanup_calls.append(context)) + raise failure + + app = _CountingApp(name="factory-failure", log_to_file=False) + app.attach(command, context_factory=context_factory) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + [], + home=Path(tmpdir), + reraise_unexpected=True, + ) + + self.assertEqual(result.exit_code, 1) + self.assertIs(result.exception, failure) + self.assertEqual(callback_calls, []) + self.assertEqual(len(cleanup_calls), 1) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_factory_registered_click_resources_close_inside_lifecycle(self) -> None: + import click + + failure = RuntimeError("factory close failed") + history_exit_codes: list[int] = [] + events: list[str] = [] + test_case = self + + class OrderedApp(_CountingApp): + def _create_context( + self, + standard: dict[str, Any], + dry_run: bool = False, + ) -> base_cli.Context: + events.append("lifecycle-enter") + context = super()._create_context(standard, dry_run=dry_run) + original_cleanup = context.cleanup + + def ordered_cleanup() -> None: + events.append("lifecycle-exit") + original_cleanup() + + context.cleanup = ordered_cleanup # type: ignore[method-assign] + return context + + class FactoryResource: + def __enter__(self) -> FactoryResource: + test_case.assertIsNotNone(base_cli.get_current_context()) + events.append("factory-resource-enter") + return self + + def __exit__( + self, + exc_type: Any, + exc_value: Any, + traceback: Any, + ) -> None: + del exc_type, traceback + test_case.assertIsNotNone(base_cli.get_current_context()) + test_case.assertIs(exc_value, failure) + events.append("factory-resource-exit") + + def history_writer( + _context: base_cli.Context, + _argv: list[str], + _sensitive: set[str], + _started_at: Any, + exit_code: int, + ) -> None: + history_exit_codes.append(exit_code) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + + def context_factory(context: base_cli.Context) -> object: + self.assertIs(base_cli.get_current_context(), context) + click.get_current_context().with_resource(FactoryResource()) + events.append("context-factory") + return object() + + def service_factory(context: base_cli.Context) -> object: + self.assertIsNotNone(context.application_context) + + def fail_close() -> None: + self.assertIs(base_cli.get_current_context(), context) + events.append("factory-close-hook") + raise failure + + click.get_current_context().call_on_close(fail_close) + events.append("service-factory") + return object() + + @click.command(name="factory-close") + def command() -> None: + events.append("callback") + + app = OrderedApp(name="factory-close", profile=profile) + app.attach( + command, + context_factory=context_factory, + service_factory=service_factory, + ) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + [], + home=Path(tmpdir), + reraise_unexpected=True, + ) + metadata_path = app.created_contexts[0].run_root / "run.json" + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + + self.assertEqual(result.exit_code, 1) + self.assertIs(result.exception, failure) + self.assertEqual(history_exit_codes, [1]) + self.assertEqual(metadata["status"], "error") + self.assertEqual(metadata["outcome"], "unexpected_error") + self.assertEqual(metadata["exit_code"], 1) + self.assertEqual( + events, + [ + "lifecycle-enter", + "factory-resource-enter", + "context-factory", + "service-factory", + "callback", + "factory-close-hook", + "factory-resource-exit", + "lifecycle-exit", + ], + ) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_factory_failure_before_lazy_resolution_opaquely_redacts_argv(self) -> None: + import click + + failure = RuntimeError("factory stopped lazy resolution") + get_calls: list[str] = [] + callback_calls: list[None] = [] + log_files: list[Path] = [] + + @click.command(name="deferred-child") + @click.option("--access-code") + def deferred_child(access_code: str | None) -> None: + del access_code + callback_calls.append(None) + + class LazyGroup(click.Group): + def list_commands(self, ctx: Any) -> list[str]: + del ctx + raise AssertionError("factory failure must not enumerate commands") + + def get_command(self, ctx: Any, name: str) -> Any: + del ctx + get_calls.append(name) + return deferred_child if name == "deferred-child" else None + + def context_factory(context: base_cli.Context) -> object: + if context.log_file is not None: + log_files.append(context.log_file) + raise failure + + root = LazyGroup(name="lazy-factory-failure") + app = _CountingApp(name="lazy-factory-failure") + app.attach(root, context_factory=context_factory) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + [ + "deferred-child", + "--access-code", + "factory-pre-resolution-value", + "factory-positional-value", + ], + home=Path(tmpdir), + reraise_unexpected=True, + ) + self.assertEqual(len(log_files), 1) + log_text = log_files[0].read_text(encoding="utf-8") + + self.assertEqual(result.exit_code, 1) + self.assertIs(result.exception, failure) + self.assertEqual(get_calls, []) + self.assertEqual(callback_calls, []) + self.assertGreaterEqual(log_text.count("[REDACTED]"), 4) + self.assertNotIn("deferred-child", log_text) + self.assertNotIn("factory-pre-resolution-value", log_text) + self.assertNotIn("factory-positional-value", log_text) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_exact_app_and_click_root_names_must_match_before_mutation(self) -> None: + import click + + command = click.Command( + name="click-root", + callback=lambda: None, + params=[click.Option(["--vendor-mode"])], + ) + original_parameters = tuple(command.params) + mismatched = base_cli.App(name="app-root", log_to_file=False) + + with self.assertRaisesRegex(RuntimeError, "app-root.*click-root"): + mismatched.attach(command) + + self.assertEqual(tuple(command.params), original_parameters) + matching = base_cli.App(name="click-root", log_to_file=False) + self.assertIs(matching.attach(command), command) + + def test_distinct_nested_attachment_is_rejected_before_parent_callback(self) -> None: + import click + + parent_calls: list[None] = [] + child_calls: list[None] = [] + + @click.command(name="child") + def child() -> None: + child_calls.append(None) + + child_app = _CountingApp(name="child", log_to_file=False) + child_app.attach(child) + + @click.group(name="parent") + def parent() -> None: + parent_calls.append(None) + + parent.add_command(child) + parent_app = _CountingApp(name="parent", log_to_file=False) + parent_app.attach(parent) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + parent_app, + ["child"], + home=Path(tmpdir), + reraise_unexpected=True, + ) + + self.assertEqual(result.exit_code, 1) + self.assertIsInstance(result.exception, RuntimeError) + self.assertRegex(str(result.exception), "different base_cli.App|nested") + self.assertEqual(parent_calls, []) + self.assertEqual(child_calls, []) + self.assertEqual(parent_app.context_create_count, 1) + self.assertEqual(parent_app.context_cleanup_count, 1) + self.assertEqual(child_app.context_create_count, 0) + self.assertEqual(child_app.context_cleanup_count, 0) + + def test_native_app_command_cannot_be_attached_or_nested(self) -> None: + import click + + parent_calls: list[None] = [] + child_calls: list[None] = [] + child_app = _CountingApp(name="native-child", log_to_file=False) + + @child_app.command() + def native_child(context: base_cli.Context) -> None: + del context + child_calls.append(None) + + native_command = child_app.click_command + original_parameters = tuple(native_command.params) + attaching_app = base_cli.App(name="native-child", log_to_file=False) + + with self.assertRaisesRegex(RuntimeError, "native base_cli.App"): + attaching_app.attach(native_command) + + self.assertEqual(tuple(native_command.params), original_parameters) + self.assertIs(child_app.click_command, native_command) + + @click.group(name="native-parent") + def parent() -> None: + parent_calls.append(None) + + parent.add_command(native_command) + parent_app = _CountingApp(name="native-parent", log_to_file=False) + parent_app.attach(parent) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + parent_app, + ["native-child"], + home=Path(tmpdir), + reraise_unexpected=True, + ) + + self.assertEqual(result.exit_code, 1) + self.assertIsInstance(result.exception, RuntimeError) + self.assertRegex(str(result.exception), "native base_cli.App|second lifecycle") + self.assertEqual(parent_calls, []) + self.assertEqual(child_calls, []) + self.assertEqual(parent_app.context_create_count, 1) + self.assertEqual(parent_app.context_cleanup_count, 1) + self.assertEqual(child_app.context_create_count, 0) + self.assertEqual(child_app.context_cleanup_count, 0) + + def test_lazy_group_resolves_only_selected_path_and_redacts_dynamic_leaf(self) -> None: + import click + + calls: dict[str, Any] = { + "list": 0, + "get": [], + "commands_property": 0, + "imports": [], + "callbacks": [], + "log_files": [], + } + cached_commands: dict[str, Any] = {} + + def load_command(name: str) -> Any: + calls["imports"].append(name) + + @click.command(name=name, help=f"Run the {name} integration.") + @click.option("--access-code", required=True) + def selected(access_code: str) -> None: + context = base_cli.get_current_context() + calls["callbacks"].append((name, access_code, context.run_id)) + calls["log_files"].append(context.log_file) + + return selected + + class LazyGroup(click.Group): + @property + def commands(self) -> Any: + calls["commands_property"] += 1 + raise AssertionError("lazy commands mapping was accessed eagerly") + + @commands.setter + def commands(self, value: Any) -> None: + self._lazy_constructor_commands = value + + def list_commands(self, ctx: Any) -> list[str]: + del ctx + calls["list"] += 1 + return ["selected", "unused"] + + def get_command(self, ctx: Any, name: str) -> Any: + del ctx + calls["get"].append(name) + if name not in {"selected", "unused"}: + return None + if name not in cached_commands: + cached_commands[name] = load_command(name) + return cached_commands[name] + + lazy_root = LazyGroup(name="lazy-suite", help="Load commands on demand.") + app = _CountingApp(name="lazy-suite") + + self.assertIs( + app.attach(lazy_root, sensitive_parameters={"access_code"}), + lazy_root, + ) + self.assertIs(app.click_command, lazy_root) + self.assertEqual(calls["list"], 0) + self.assertEqual(calls["get"], []) + self.assertEqual(calls["commands_property"], 0) + self.assertEqual(calls["imports"], []) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + first = invoke( + app, + ["selected", "--access-code", "first-lazy-value"], + home=home, + ) + first_callback = cached_commands["selected"].callback + second = invoke( + app, + ["selected", "--access-code", "second-lazy-value"], + home=home, + ) + + log_texts = [ + path.read_text(encoding="utf-8") + for path in calls["log_files"] + if path is not None + ] + + self.assertEqual(first.exit_code, 0, first.output) + self.assertEqual(second.exit_code, 0, second.output) + self.assertEqual(calls["list"], 0) + self.assertEqual(calls["get"], ["selected", "selected"]) + self.assertEqual(calls["commands_property"], 0) + self.assertEqual(calls["imports"], ["selected"]) + self.assertEqual( + [(name, value) for name, value, _run_id in calls["callbacks"]], + [ + ("selected", "first-lazy-value"), + ("selected", "second-lazy-value"), + ], + ) + self.assertNotEqual(calls["callbacks"][0][2], calls["callbacks"][1][2]) + self.assertIs(cached_commands["selected"].callback, first_callback) + self.assertEqual(app.context_create_count, 2) + self.assertEqual(app.context_cleanup_count, 2) + self.assertEqual(len(log_texts), 2) + for log_text in log_texts: + self.assertIn("[REDACTED]", log_text) + self.assertNotIn("first-lazy-value", log_text) + self.assertNotIn("second-lazy-value", log_text) + + def test_context_and_service_factories_extend_base_context_without_replacing_click_obj(self) -> None: + import click + + vendor_object = {"vendor": "preserved"} + application_context = object() + services = object() + events: list[tuple[str, Any]] = [] + + @click.group(name="factories") + @click.pass_context + def root(click_context: Any) -> None: + context = base_cli.get_current_context() + click_context.obj = vendor_object + + def close_click_context() -> None: + self.assertIs(base_cli.get_current_context(), context) + events.append(("click-close", app.context_cleanup_count)) + + click_context.call_on_close(close_click_context) + events.append(( + "root", + ( + click_context.obj, + context.application_context, + context.services, + ), + )) + + @root.command(name="run") + @click.pass_obj + def run(vendor_state: Any) -> None: + context = base_cli.get_current_context() + events.append( + ( + "callback", + ( + vendor_state, + context.application_context, + context.services, + ), + ) + ) + + @root.result_callback() + def finish(result: Any) -> Any: + context = base_cli.get_current_context() + events.append(( + "result", + (context.application_context, context.services), + )) + return result + + def context_factory(context: base_cli.Context) -> object: + self.assertIs(base_cli.get_current_context(), context) + events.append(("context-factory", context)) + + def cleanup_base_context() -> None: + self.assertIs(base_cli.get_current_context(), context) + events.append(("base-cleanup", app.context_cleanup_count)) + + context.on_cleanup(cleanup_base_context) + return application_context + + def service_factory(context: base_cli.Context) -> object: + self.assertIs(base_cli.get_current_context(), context) + self.assertIs(context.application_context, application_context) + events.append(("service-factory", context)) + return services + + app = _CountingApp(name="factories", log_to_file=False) + app.attach( + root, + context_factory=context_factory, + service_factory=service_factory, + ) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + help_result = invoke(app, ["--help"], home=home) + result = invoke(app, ["run"], home=home) + + self.assertEqual(help_result.exit_code, 0, help_result.output) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + [name for name, _value in events], + [ + "context-factory", + "service-factory", + "root", + "callback", + "result", + "click-close", + "base-cleanup", + ], + ) + root_vendor_state, root_application_context, root_services = events[2][1] + self.assertIs(root_vendor_state, vendor_object) + self.assertIs(root_application_context, application_context) + self.assertIs(root_services, services) + vendor_state, actual_application_context, actual_services = events[3][1] + self.assertIs(vendor_state, vendor_object) + self.assertIs(actual_application_context, application_context) + self.assertIs(actual_services, services) + result_application_context, result_services = events[4][1] + self.assertIs(result_application_context, application_context) + self.assertIs(result_services, services) + self.assertEqual(events[5], ("click-close", 0)) + self.assertEqual(events[6], ("base-cleanup", 1)) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_chained_group_keeps_one_lifecycle_and_click_list_results(self) -> None: + import click + + events: list[tuple[str, Any]] = [] + run_ids: list[str] = [] + log_files: list[Path] = [] + + @click.group(name="pipeline", chain=True) + def pipeline() -> None: + context = base_cli.get_current_context() + run_ids.append(context.run_id) + if context.log_file is not None: + log_files.append(context.log_file) + events.append(("pipeline", None)) + + @pipeline.command(name="extract") + @click.option("--source-code", required=True) + def extract(source_code: str) -> str: + context = base_cli.get_current_context() + run_ids.append(context.run_id) + events.append(("extract", source_code)) + return "rows" + + @pipeline.command(name="load") + @click.option("--destination-code", required=True) + @click.argument("payload") + def load(destination_code: str, payload: str) -> dict[str, int]: + context = base_cli.get_current_context() + run_ids.append(context.run_id) + events.append(("load", (destination_code, payload))) + return {"loaded": 3} + + @pipeline.result_callback() + def finish(results: list[Any]) -> list[Any]: + context = base_cli.get_current_context() + run_ids.append(context.run_id) + events.append(("result", results)) + return results + + app = _CountingApp(name="pipeline") + app.attach( + pipeline, + sensitive_parameters={"source_code", "destination_code", "payload"}, + ) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + [ + "extract", + "--source-code", + "first-chain-value", + "load", + "--destination-code", + "second-chain-option", + "second-chain-argument", + ], + home=Path(tmpdir), + ) + self.assertEqual(len(log_files), 1) + log_text = log_files[0].read_text(encoding="utf-8") + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + events, + [ + ("pipeline", None), + ("extract", "first-chain-value"), + ("load", ("second-chain-option", "second-chain-argument")), + ("result", ["rows", {"loaded": 3}]), + ], + ) + self.assertGreaterEqual(log_text.count("[REDACTED]"), 3) + self.assertNotIn("first-chain-value", log_text) + self.assertNotIn("second-chain-option", log_text) + self.assertNotIn("second-chain-argument", log_text) + self.assertEqual(len(set(run_ids)), 1) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_later_chain_parse_failure_redacts_earlier_and_failing_values(self) -> None: + import click + + callback_calls: list[str] = [] + + @click.group(name="failing-chain", chain=True) + def pipeline() -> None: + callback_calls.append("pipeline") + + @pipeline.command(name="first") + @click.option("--source-code", required=True) + def first(source_code: str) -> None: + del source_code + callback_calls.append("first") + + @pipeline.command(name="second") + @click.option("--destination-code", required=True) + @click.option("--confirm", is_flag=True, required=True) + @click.argument("payload") + def second(destination_code: str, confirm: bool, payload: str) -> None: + del destination_code, confirm, payload + callback_calls.append("second") + + app = _CountingApp(name="failing-chain") + app.attach( + pipeline, + sensitive_parameters={"source_code", "destination_code", "payload"}, + ) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + [ + "first", + "--source-code", + "earlier-chain-value", + "second", + "--destination-code", + "failing-chain-option", + "failing-chain-argument", + ], + home=Path(tmpdir), + ) + self.assertEqual(len(app.created_contexts), 1) + log_file = app.created_contexts[0].log_file + self.assertIsNotNone(log_file) + log_text = log_file.read_text(encoding="utf-8") + + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("Missing option '--confirm'", _all_output(result)) + # Click invokes a chain group's root callback before constructing every + # child context. Neither selected member may run after the later parse + # failure. + self.assertEqual(callback_calls, ["pipeline"]) + self.assertGreaterEqual(log_text.count("[REDACTED]"), 3) + self.assertNotIn("earlier-chain-value", log_text) + self.assertNotIn("failing-chain-option", log_text) + self.assertNotIn("failing-chain-argument", log_text) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_preparse_click_resources_close_inside_lifecycle_and_record_failure( + self, + ) -> None: + import click + + history_exit_codes: list[int] = [] + failure = RuntimeError("vendor close failed") + events: list[str] = [] + test_case = self + + class OrderedApp(_CountingApp): + def _create_context( + self, + standard: dict[str, Any], + dry_run: bool = False, + ) -> base_cli.Context: + events.append("lifecycle-enter") + context = super()._create_context(standard, dry_run=dry_run) + original_cleanup = context.cleanup + + def ordered_cleanup() -> None: + events.append("lifecycle-exit") + original_cleanup() + + context.cleanup = ordered_cleanup # type: ignore[method-assign] + return context + + class VendorResource: + def __enter__(self) -> VendorResource: + events.append("vendor-enter") + return self + + def __exit__( + self, + exc_type: Any, + exc_value: Any, + traceback: Any, + ) -> None: + del exc_type, traceback + self_context = base_cli.get_current_context() + test_case.assertIsNotNone(self_context) + test_case.assertIs(exc_value, failure) + events.append("vendor-exit") + + class FactoryResource: + def __enter__(self) -> FactoryResource: + self_context = base_cli.get_current_context() + test_case.assertIsNotNone(self_context) + events.append("factory-resource-enter") + return self + + def __exit__( + self, + exc_type: Any, + exc_value: Any, + traceback: Any, + ) -> None: + del exc_type, traceback + self_context = base_cli.get_current_context() + test_case.assertIsNotNone(self_context) + test_case.assertIsNone(exc_value) + events.append("factory-resource-exit") + + def history_writer( + _context: base_cli.Context, + _argv: list[str], + _sensitive: set[str], + _started_at: Any, + exit_code: int, + ) -> None: + history_exit_codes.append(exit_code) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + + def context_factory(context: base_cli.Context) -> object: + self.assertIs(base_cli.get_current_context(), context) + click.get_current_context().with_resource(FactoryResource()) + events.append("context-factory") + return object() + + def register_close_failure( + click_context: Any, + _parameter: Any, + value: str, + ) -> str: + click_context.with_resource(VendorResource()) + + def fail_close() -> None: + self.assertIsNotNone(base_cli.get_current_context()) + events.append("close-hook") + raise failure + + click_context.call_on_close(fail_close) + return value + + @click.command(name="close-failure") + @click.option("--vendor", callback=register_close_failure) + def command(vendor: str | None) -> None: + del vendor + events.append("callback") + + app = OrderedApp(name="close-failure", profile=profile) + app.attach(command, context_factory=context_factory) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + ["--vendor", "value"], + home=Path(tmpdir), + reraise_unexpected=True, + ) + self.assertEqual(len(app.created_contexts), 1) + metadata_path = app.created_contexts[0].run_root / "run.json" + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + + self.assertEqual(result.exit_code, 1) + self.assertIs(result.exception, failure) + self.assertEqual(history_exit_codes, [1]) + self.assertEqual(metadata["status"], "error") + self.assertEqual(metadata["outcome"], "unexpected_error") + self.assertEqual(metadata["exit_code"], 1) + self.assertEqual( + events, + [ + "vendor-enter", + "lifecycle-enter", + "factory-resource-enter", + "context-factory", + "callback", + "factory-resource-exit", + "close-hook", + "vendor-exit", + "lifecycle-exit", + ], + ) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_close_hook_click_exit_controls_run_app_status_and_metadata(self) -> None: + import click + + history_exit_codes: list[int] = [] + + def history_writer( + _context: base_cli.Context, + _argv: list[str], + _sensitive: set[str], + _started_at: Any, + exit_code: int, + ) -> None: + history_exit_codes.append(exit_code) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + + @click.command(name="close-exit") + @click.pass_context + def command(click_context: Any) -> None: + click_context.call_on_close(lambda: click_context.exit(7)) + + app = _CountingApp(name="close-exit", profile=profile) + app.attach(command) + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke(app, [], home=Path(tmpdir)) + self.assertEqual(len(app.created_contexts), 1) + metadata_path = app.created_contexts[0].run_root / "run.json" + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + + self.assertEqual( + result.exit_code, + 7, + f"{result.output}\nhistory={history_exit_codes!r} metadata={metadata!r}", + ) + self.assertEqual(history_exit_codes, [7]) + self.assertEqual(metadata["status"], "error") + self.assertEqual(metadata["outcome"], "nonzero_return") + self.assertEqual(metadata["exit_code"], 7) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_direct_click_main_parity_preserves_results_and_redacts_argv(self) -> None: + import click + from click.testing import CliRunner + + log_files: list[Path] = [] + callback_values: list[str] = [] + + @click.command(name="direct-main") + @click.option("--access-code", required=True) + def command(access_code: str) -> dict[str, str]: + context = base_cli.get_current_context() + callback_values.append(access_code) + if context.log_file is not None: + log_files.append(context.log_file) + return {"value": access_code} + + app = _CountingApp(name="direct-main") + app.attach(command, sensitive_parameters={"access_code"}) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + environment = { + "HOME": str(home), + "USERPROFILE": str(home), + "XDG_CACHE_HOME": str(home / ".cache"), + "BASE_CLI_CACHE_DIR": str(home / ".cache"), + } + runner_result = CliRunner().invoke( + command, + ["--access-code", "runner-secret"], + env=environment, + ) + with mock.patch.dict(os.environ, environment): + direct_result = command.main( + args=["--access-code", "main-secret"], + prog_name="direct-main", + standalone_mode=False, + ) + iterator_result = command.main( + args=iter(["--access-code", "iterator-secret"]), + prog_name="direct-main", + standalone_mode=False, + ) + log_texts = [path.read_text(encoding="utf-8") for path in log_files] + + self.assertEqual(runner_result.exit_code, 0, runner_result.output) + self.assertEqual(direct_result, {"value": "main-secret"}) + self.assertEqual(iterator_result, {"value": "iterator-secret"}) + self.assertEqual( + callback_values, + ["runner-secret", "main-secret", "iterator-secret"], + ) + self.assertEqual(app.context_create_count, 3) + self.assertEqual(app.context_cleanup_count, 3) + self.assertEqual(len(log_texts), 3) + for log_text in log_texts: + self.assertIn("[REDACTED]", log_text) + self.assertNotIn("runner-secret", log_text) + self.assertNotIn("main-secret", log_text) + self.assertNotIn("iterator-secret", log_text) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_nested_attached_main_scopes_argv_and_preserves_native_status(self) -> None: + import click + + inner_calls: list[None] = [] + outer_secret = "OUTER-SUPER-SECRET" + + @click.command(name="inner-attached") + def inner_command() -> dict[str, bool]: + inner_calls.append(None) + return {"inner": True} + + inner_app = _CountingApp(name="inner-attached") + inner_app.attach(inner_command) + + outer_app = _CountingApp(name="outer-native") + + @outer_app.command() + @base_cli.option("--credential", sensitive=True, required=True) + def outer_native(context: base_cli.Context, credential: str) -> int: + self.assertEqual(context.cli_name, "outer-native") + self.assertEqual(credential, outer_secret) + result = inner_command.main( + args=[], + prog_name="inner-alias", + standalone_mode=False, + ) + self.assertEqual(result, {"inner": True}) + return 9 + + with tempfile.TemporaryDirectory() as tmpdir: + with mock.patch.dict( + os.environ, + { + "HOME": tmpdir, + "USERPROFILE": tmpdir, + "XDG_CACHE_HOME": str(Path(tmpdir) / ".cache"), + "BASE_CLI_CACHE_DIR": str(Path(tmpdir) / ".cache"), + }, + ): + status = base_cli.run_app( + outer_app, + ["--credential", outer_secret], + ) + log_paths = [ + outer_app.created_contexts[0].log_file, + inner_app.created_contexts[0].log_file, + ] + self.assertTrue(all(path is not None for path in log_paths)) + log_texts = [ + path.read_text(encoding="utf-8") + for path in log_paths + if path is not None + ] + + self.assertEqual(status, 9) + self.assertEqual(inner_calls, [None]) + self.assertEqual(len(log_texts), 2) + self.assertIn("[REDACTED]", log_texts[0]) + self.assertNotIn("--credential", log_texts[1]) + for log_text in log_texts: + self.assertNotIn(outer_secret, log_text) + self.assertEqual(outer_app.context_create_count, 1) + self.assertEqual(outer_app.context_cleanup_count, 1) + self.assertEqual(inner_app.context_create_count, 1) + self.assertEqual(inner_app.context_cleanup_count, 1) + + def test_run_app_materialization_failure_restores_invocation_state(self) -> None: + import importlib + + app_module = importlib.import_module("base_cli.app") + ambient_argv = ["ambient-cli", "ambient-secret"] + ambient_bypass = object() + argv_token = app_module._INVOCATION_ARGV.set(ambient_argv) + bypass_token = app_module._INVOCATION_MAIN_BYPASS.set(ambient_bypass) + unconfigured_app = base_cli.App( + name="materialization-failure", + log_to_file=False, + ) + + try: + with self.assertRaisesRegex(RuntimeError, "No command has been registered"): + base_cli.run_app( + unconfigured_app, + [], + reraise_unexpected=True, + ) + self.assertIs(app_module._INVOCATION_ARGV.get(), ambient_argv) + self.assertIs(app_module._INVOCATION_MAIN_BYPASS.get(), ambient_bypass) + finally: + app_module._INVOCATION_MAIN_BYPASS.reset(bypass_token) + app_module._INVOCATION_ARGV.reset(argv_token) + + def test_module_attach_with_existing_app_returns_the_same_command(self) -> None: + import click + + seen: list[str] = [] + + @click.command(name="module-existing") + def command() -> None: + seen.append(base_cli.get_current_context().cli_name) + + app = _CountingApp(name="module-existing", log_to_file=False) + + attached = base_cli.attach(command, app=app) + + self.assertIs(attached, command) + self.assertIs(app.click_command, command) + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke(app, [], home=Path(tmpdir)) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(seen, ["module-existing"]) + self.assertEqual(app.context_create_count, 1) + self.assertEqual(app.context_cleanup_count, 1) + + def test_module_attach_can_create_an_app_for_the_run_app_boundary(self) -> None: + import click + + seen: dict[str, Any] = {} + application_context = object() + services = object() + factory_calls: list[str] = [] + + def context_factory(context: base_cli.Context) -> object: + self.assertIs(base_cli.get_current_context(), context) + factory_calls.append("context") + return application_context + + def service_factory(context: base_cli.Context) -> object: + self.assertIs(context.application_context, application_context) + factory_calls.append("services") + return services + + @click.command(name="module-created") + def command() -> dict[str, str]: + context = base_cli.get_current_context() + seen["cli_name"] = context.cli_name + seen["context"] = context + seen["application_context"] = context.application_context + seen["services"] = context.services + context.on_cleanup(lambda: seen.update(cleaned=True)) + return {"vendor": "result"} + + attached = base_cli.attach( + command, + name="module-created", + log_to_file=False, + context_factory=context_factory, + service_factory=service_factory, + sensitive_parameters={"vendor_result"}, + ) + + self.assertIs(attached, command) + self.assertIs(base_cli.attach(command), command) + implicit_app = base_cli.get_command_app(command) + self.assertIs(base_cli.attach(command, app=implicit_app), command) + self.assertIs( + base_cli.attach( + command, + context_factory=context_factory, + service_factory=service_factory, + sensitive_parameters={"vendor_result"}, + ), + command, + ) + with self.assertRaisesRegex(RuntimeError, "already attached"): + base_cli.attach( + command, + context_factory=context_factory, + service_factory=service_factory, + sensitive_parameters={"different"}, + ) + with self.assertRaisesRegex(RuntimeError, "already attached"): + base_cli.attach( + command, + sensitive_parameters={"different-without-factories"}, + ) + self.assertIs( + base_cli.attach( + command, + context_factory=context_factory, + service_factory=service_factory, + ), + command, + ) + with self.assertRaisesRegex(RuntimeError, "already attached"): + base_cli.attach( + command, + context_factory=lambda context: context, + service_factory=service_factory, + ) + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + with mock.patch.dict( + os.environ, + { + "HOME": str(home), + "USERPROFILE": str(home), + "XDG_CACHE_HOME": str(home / ".cache"), + "BASE_CLI_CACHE_DIR": str(home / ".cache"), + }, + ): + status = base_cli.run_app(attached, []) + + self.assertEqual(status, 0) + self.assertEqual(seen["cli_name"], "module-created") + self.assertIs(seen["application_context"], application_context) + self.assertIs(seen["services"], services) + self.assertEqual(factory_calls, ["context", "services"]) + self.assertTrue(seen["cleaned"]) + with self.assertRaisesRegex(RuntimeError, "context is not active"): + base_cli.get_current_context() + + def test_module_attach_rejects_unnamed_click_commands_directly(self) -> None: + import click + + command = click.Command(name=None, callback=lambda: None) + + with self.assertRaisesRegex(TypeError, "named Click command"): + base_cli.attach(command, name="explicit-name") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_public_api.py b/tests/test_public_api.py index e8bc9ae..2f891c1 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -33,6 +33,7 @@ def test_facade_exports_supported_modules_functions_and_types(self) -> None: expected = { "CommandProtocolError", "ConfigurationError", + "attach", "command_filters", "command_matches", "command_protocol", @@ -77,6 +78,7 @@ def test_module_all_surfaces_are_explicit(self) -> None: def test_entry_points_have_docstrings(self) -> None: self.assertTrue(base_cli.App.__doc__) self.assertTrue(base_cli.Context.__doc__) + self.assertTrue(base_cli.attach.__doc__) self.assertTrue(base_cli.get_command_app.__doc__) self.assertTrue(base_cli.run_app.__doc__)