diff --git a/CHANGELOG.md b/CHANGELOG.md index c0a2611..d25dc17 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,10 @@ and versions are tracked in the repo-root `VERSION` file. ### Fixed +- Make `App.name` authoritative for single-command identity, reject duplicate + or post-materialization registrations deterministically, stabilize inferred + names across Click releases, and make module-level `@command()` functions + retrievable and directly runnable through `run_app()`. - Redact sensitive option values across every declared alias and Click value form, redact sensitive positional arguments, and protect conventional secret parameter names automatically before argv reaches logs or history writers. diff --git a/README.md b/README.md index 040eefd..650e8c9 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,10 @@ Use `App` when you want a named command: app = base_cli.App(name="workspace-tools", version="0.1.0") ``` +`App.name` is the canonical Click command and program name. It controls usage, +help, version output, runtime identity, and the default invocation label. Do not +pass a conflicting name to `@app.command(...)`; change `App(name=...)` instead. + Register the command function explicitly: ```python @@ -170,10 +174,19 @@ For small scripts, the module-level decorators are available: @base_cli.command() def main(ctx: base_cli.Context) -> None: ... + + +if __name__ == "__main__": + raise SystemExit(base_cli.run_app(main)) ``` -Prefer an explicit `App` when command names, versions, or consumer -policies should be visible at the top of the module. +The decorator returns the original function. `base_cli.get_command_app(main)` +retrieves its private owning `App` when an embedding layer needs the command +object. Every module-level registration gets an independent app; there is no +process-global command registry. Its default name is inferred from the function; +pass a public name such as `@base_cli.command("workspace-tools")` when needed. +Prefer an explicit `App` when versions or consumer policies should be visible at +the top of the module. Use `@app.subcommand()` when one CLI needs multiple verbs while keeping the standard context, logging, redaction, and cleanup lifecycle for each invocation: @@ -208,6 +221,15 @@ compatibility. Use either `@app.command()` for a single-command CLI or `@app.subcommand()` for a command group; do not mix the two registration styles on one `App`. +Finish all command and subcommand registration before the first access to +`app.click_command`, direct app invocation, `run_app()`, or +`base_cli.testing.invoke()`. Successful materialization freezes registration; +late mutations and duplicate effective command names fail deterministically. +Inferred names are stable across supported Click releases: underscores become +hyphens and conventional `_command`, `_cmd`, `_group`, and `_grp` suffixes are +removed. Pass an explicit subcommand name when a different public spelling is +required. + ## Options And Arguments `base_cli.option` and `base_cli.argument` mirror Click's decorators: diff --git a/lib/python/base_cli/__init__.py b/lib/python/base_cli/__init__.py index 76e6dbb..cceab32 100644 --- a/lib/python/base_cli/__init__.py +++ b/lib/python/base_cli/__init__.py @@ -31,7 +31,15 @@ def _resolve_version() -> str: __version__ = _resolve_version() from . import command_filters, command_protocol, history, testing -from .app import App, argument, command, delegated_display_command, option, run_app +from .app import ( + App, + argument, + command, + delegated_display_command, + get_command_app, + option, + run_app, +) from .command_filters import CommandFilterNormalizer, command_matches, normalize_command_filter, normalize_command_filters from .command_protocol import ( BOOLEAN, @@ -86,6 +94,7 @@ def _resolve_version() -> str: "command", "configure_logger", "delegated_display_command", + "get_command_app", "get_current_context", "log_critical", "log_debug", diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index cf90bec..5a42bc6 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -10,7 +10,7 @@ from dataclasses import dataclass from datetime import datetime from pathlib import Path -from threading import Lock +from threading import RLock from typing import Any, Callable from ._lifecycle import ( @@ -43,6 +43,12 @@ _GROUP_STANDARD_OPTIONS_KEY = "base_cli_standard_options" DISPLAY_COMMAND_ENV = "BASE_CLI_DISPLAY_COMMAND" _INVOCATION_ARGV: ContextVar[list[str] | None] = ContextVar("base_cli_invocation_argv", default=None) +_COMMAND_APP_ATTRIBUTE = "__base_cli_command_app__" +_COMMAND_APP_LOCK = RLock() +_REGISTRATION_OPEN = "open" +_REGISTRATION_MATERIALIZING = "materializing" +_REGISTRATION_FROZEN = "frozen" +_COMMAND_NAME_SUFFIXES = frozenset({"command", "cmd", "group", "grp"}) @dataclass @@ -54,6 +60,14 @@ class _InvocationState: options_parsed: bool = False +@dataclass(frozen=True) +class _SubcommandRegistration: + func: Callable[..., Any] + args: tuple[Any, ...] + kwargs: dict[str, Any] + name: str + + _INVOCATION_STATE: ContextVar[_InvocationState | None] = ContextVar("base_cli_invocation_state", default=None) @@ -177,6 +191,64 @@ def _require_click(): return click +def _explicit_command_name( + command_args: tuple[Any, ...], + command_kwargs: dict[str, Any], +) -> str | None: + if command_args and "name" in command_kwargs: + raise TypeError("Command name cannot be provided both positionally and by keyword.") + name = command_args[0] if command_args else command_kwargs.get("name") + if name is None: + return None + if not isinstance(name, str): + raise TypeError("Command name must be a string or None.") + return name + + +def _inferred_command_name(func: Callable[..., Any]) -> str: + name = func.__name__.lower().replace("_", "-") + prefix, separator, suffix = name.rpartition("-") + if separator and suffix in _COMMAND_NAME_SUFFIXES: + return prefix + return name + + +def _resolved_command_name( + func: Callable[..., Any], + command_args: tuple[Any, ...], + command_kwargs: dict[str, Any], +) -> str: + return _explicit_command_name(command_args, command_kwargs) or _inferred_command_name(func) + + +def _click_command_decorator( + click: Any, + name: str, + command_args: tuple[Any, ...], + command_kwargs: dict[str, Any], +) -> Callable[[Callable[..., Any]], Any]: + # ``name`` is resolved by base-cli so naming and duplicate behavior do not + # drift across supported Click versions. Preserve the optional positional + # command class and all non-name attributes. + args_after_name = command_args[1:] if command_args else () + attrs = dict(command_kwargs) + attrs.pop("name", None) + return click.command(name, *args_after_name, **attrs) + + +def _require_materialized_command_name( + command: Any, + expected_name: str, + app_name: str, +) -> None: + actual_name = getattr(command, "name", None) + if actual_name != expected_name: + raise RuntimeError( + f"App '{app_name}' expected Click command name '{expected_name}', " + f"but the configured command class produced {actual_name!r}." + ) + + # pylint: disable=too-many-statements class App: """Define a Click-backed command with a shared runtime lifecycle.""" @@ -193,7 +265,9 @@ def __init__( ) -> None: if max_log_files is not None and max_log_files < 1: raise ValueError("max_log_files must be greater than 0 when set.") - self.name = normalize_cli_name(name or sys.argv[0]) + self._registration_lock = RLock() + self._registration_state = _REGISTRATION_OPEN + self._name = normalize_cli_name(name or sys.argv[0]) self.version = version self.help = help self.log_to_file = log_to_file @@ -204,60 +278,148 @@ def __init__( self.profile = profile or CliProfile.generic() self._click_command = None self._redaction_plan: RedactionPlan | None = None - self._click_command_lock = Lock() self._command_func: Callable[..., Any] | None = None self._command_args: tuple[Any, ...] = () self._command_kwargs: dict[str, Any] = {} - self._subcommands: list[tuple[Callable[..., Any], tuple[Any, ...], dict[str, Any]]] = [] + self._subcommands: list[_SubcommandRegistration] = [] + self._subcommand_names: set[str] = set() - def command(self, *command_args: Any, **command_kwargs: Any): - def decorator(func: Callable[..., Any]): - if self._subcommands: - raise RuntimeError( - f"App '{self.name}' already has registered subcommands. " - "Use @app.subcommand() for additional entry points." - ) - if self._command_func is not None: + @property + def name(self) -> str: + return self._name + + @name.setter + def name(self, value: str) -> None: + normalized = normalize_cli_name(value) + with self._registration_lock: + self._ensure_registration_open() + explicit_name = _explicit_command_name( + self._command_args, + self._command_kwargs, + ) + if ( + self._command_func is not None + and explicit_name is not None + and normalize_cli_name(explicit_name) != normalized + ): raise RuntimeError( - f"App '{self.name}' already has a registered command. " - "Use subcommands for multiple entry points." + f"App '{self.name}' cannot be renamed to '{normalized}' because " + f"its registered command explicitly uses '{explicit_name}'." ) - self._command_func = func - self._command_args = command_args - self._command_kwargs = command_kwargs + self._name = normalized + + def _ensure_registration_open(self) -> None: + if self._registration_state == _REGISTRATION_MATERIALIZING: + raise RuntimeError( + f"App '{self.name}' registration is unavailable while its Click command " + "is being materialized." + ) + if self._registration_state == _REGISTRATION_FROZEN: + raise RuntimeError( + f"App '{self.name}' registration is frozen because its Click command " + "has already been materialized." + ) + + def _validate_single_command_name( + self, + command_args: tuple[Any, ...], + command_kwargs: dict[str, Any], + ) -> None: + explicit_name = _explicit_command_name(command_args, command_kwargs) + if explicit_name is not None and normalize_cli_name(explicit_name) != self.name: + raise RuntimeError( + f"App '{self.name}' is the authoritative command name; " + f"the registered command cannot use '{explicit_name}'." + ) + + def command(self, *command_args: Any, **command_kwargs: Any): + with self._registration_lock: + self._ensure_registration_open() + self._validate_single_command_name(command_args, command_kwargs) + + def decorator(func: Callable[..., Any]): + with self._registration_lock: + self._ensure_registration_open() + self._validate_single_command_name(command_args, command_kwargs) + if self._subcommands: + raise RuntimeError( + f"App '{self.name}' already has registered subcommands. " + "Use @app.subcommand() for additional entry points." + ) + if self._command_func is not None: + raise RuntimeError( + f"App '{self.name}' already has a registered command. " + "Use subcommands for multiple entry points." + ) + self._command_func = func + self._command_args = tuple(command_args) + self._command_kwargs = dict(command_kwargs) return func return decorator def subcommand(self, *command_args: Any, **command_kwargs: Any): + with self._registration_lock: + self._ensure_registration_open() + _explicit_command_name(command_args, command_kwargs) + def decorator(func: Callable[..., Any]): - if self._command_func is not None: - raise RuntimeError( - f"App '{self.name}' already has a registered command. " - "Use either @app.command() or @app.subcommand(), not both." + with self._registration_lock: + self._ensure_registration_open() + if self._command_func is not None: + raise RuntimeError( + f"App '{self.name}' already has a registered command. " + "Use either @app.command() or @app.subcommand(), not both." + ) + name = _resolved_command_name(func, command_args, command_kwargs) + if name in self._subcommand_names: + raise RuntimeError( + f"App '{self.name}' already has a registered subcommand named '{name}'." + ) + self._subcommands.append( + _SubcommandRegistration( + func=func, + args=tuple(command_args), + kwargs=dict(command_kwargs), + name=name, + ) ) - self._subcommands.append((func, command_args, command_kwargs)) + self._subcommand_names.add(name) return func return decorator 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 return self.click_command(*args, **kwargs) @property def click_command(self) -> Any: - command = self._click_command - if command is not None: - return command - - with self._click_command_lock: + with self._registration_lock: command = self._click_command - if command is None: + if command is not None: + return command + if self._registration_state == _REGISTRATION_MATERIALIZING: + raise RuntimeError( + f"App '{self.name}' Click command materialization is already in progress." + ) + + self._registration_state = _REGISTRATION_MATERIALIZING + try: command = self._build_click_command() redaction_plan = compile_redaction_plan(command) + except BaseException: + # A missing dependency, invalid custom Click class, or plan + # compilation failure must not strand an otherwise repairable + # application in a half-materialized state. + self._registration_state = _REGISTRATION_OPEN + raise + else: # Publish the command last so another thread can never invoke # its wrapper before the corresponding plan is available. self._redaction_plan = redaction_plan + self._registration_state = _REGISTRATION_FROZEN self._click_command = command return command @@ -271,13 +433,33 @@ def _build_click_command(self) -> Any: command_kwargs = dict(self._command_kwargs) if self.help is not None: command_kwargs.setdefault("help", self.help) - return click.command(*self._command_args, **command_kwargs)(wrapper) + command = _click_command_decorator( + click, + self.name, + self._command_args, + command_kwargs, + )(wrapper) + _require_materialized_command_name(command, self.name, self.name) + 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) - for func, command_args, command_kwargs in self._subcommands: - wrapper = self._build_command_wrapper(click, func, include_version=False) - group.add_command(click.command(*command_args, **command_kwargs)(wrapper)) + for registration in self._subcommands: + wrapper = self._build_command_wrapper(click, registration.func, include_version=False) + command = _click_command_decorator( + click, + registration.name, + registration.args, + registration.kwargs, + )(wrapper) + _require_materialized_command_name(command, registration.name, self.name) + # 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. + group.add_command( + command, + name=registration.name, + ) return group def _build_command_wrapper( @@ -536,8 +718,30 @@ def _rollback_context_creation( pass -def run_app(app: App, argv: list[str] | None = None, *, reraise_unexpected: bool = False) -> int: - """Run an :class:`App` and return its normalized process exit code.""" +def get_command_app(command_func: Callable[..., Any]) -> App: + """Return the isolated :class:`App` owned by ``@base_cli.command``.""" + + with _COMMAND_APP_LOCK: + owner = getattr(command_func, _COMMAND_APP_ATTRIBUTE, None) + if isinstance(owner, App): + with owner._registration_lock: # pylint: disable=protected-access + 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()." + ) + + +def run_app( + app: App | Callable[..., Any], + argv: list[str] | None = None, + *, + reraise_unexpected: bool = False, +) -> int: + """Run an :class:`App` or registered command and return its process exit code.""" + + if not isinstance(app, App): + app = get_command_app(app) try: click = _require_click() @@ -556,10 +760,11 @@ def run_app(app: App, argv: list[str] | None = None, *, reraise_unexpected: bool invocation_argv = _effective_invocation_argv(app, args, explicit_argv, display_command) invocation_token = _INVOCATION_ARGV.set(invocation_argv) try: - if display_command: - result = app.click_command.main(args=args, prog_name=display_command, standalone_mode=False) - else: - result = app.click_command.main(args=args, standalone_mode=False) + result = app.click_command.main( + args=args, + prog_name=display_command or app.name, + standalone_mode=False, + ) finally: _reset_context_var(_INVOCATION_ARGV, invocation_token) except click.Abort as exc: @@ -666,7 +871,26 @@ def delegated_display_command(default: str | None = None) -> str | None: def command(*args: Any, **kwargs: Any): - return App().command(*args, **kwargs) + explicit_name = _explicit_command_name(args, kwargs) + + def decorator(func: Callable[..., Any]): + with _COMMAND_APP_LOCK: + if getattr(func, _COMMAND_APP_ATTRIBUTE, None) is not None: + raise RuntimeError( + f"Function '{func.__name__}' is already registered with " + "@base_cli.command()." + ) + owner = App(name=explicit_name or _inferred_command_name(func)) + registered = owner.command(*args, **kwargs)(func) + try: + setattr(func, _COMMAND_APP_ATTRIBUTE, owner) + except (AttributeError, TypeError) as exc: + raise TypeError( + "@base_cli.command() requires a function that can retain its owning App." + ) from exc + return registered + + return decorator def option(*param_decls: str, sensitive: bool = False, dry_run: bool = False, **attrs: Any): diff --git a/tests/test_app_registration.py b/tests/test_app_registration.py new file mode 100644 index 0000000..520ec08 --- /dev/null +++ b/tests/test_app_registration.py @@ -0,0 +1,645 @@ +from __future__ import annotations + +import importlib.util +import io +import os +import tempfile +import threading +import unittest +from contextlib import redirect_stderr +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 _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 + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class AppRegistrationTests(unittest.TestCase): + def test_single_command_uses_app_identity_for_click_help_version_and_usage(self) -> None: + app = base_cli.App( + name="professional-tool", + version="1.2.3", + help="Operate the professional tool.", + log_to_file=False, + ) + + @app.command() + def implementation(ctx: base_cli.Context) -> None: + del ctx + + click_command = app.click_command + self.assertEqual(click_command.name, "professional-tool") + self.assertEqual(click_command.help, "Operate the professional tool.") + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + help_result = invoke(app, ["--help"], home=home) + version_result = invoke(app, ["--version"], home=home) + usage_result = invoke(app, ["--unknown"], home=home) + + self.assertEqual(help_result.exit_code, 0, help_result.output) + self.assertIn("Usage: professional-tool [OPTIONS]", help_result.output) + self.assertIn("Operate the professional tool.", help_result.output) + self.assertNotIn("Usage: implementation", help_result.output) + + self.assertEqual(version_result.exit_code, 0, version_result.output) + self.assertIn("professional-tool, version 1.2.3", version_result.output) + + self.assertEqual(usage_result.exit_code, 2, usage_result.output) + self.assertIn("Usage: professional-tool [OPTIONS]", _all_output(usage_result)) + self.assertNotIn("Usage: implementation", _all_output(usage_result)) + + def test_single_command_rejects_a_name_conflicting_with_app_identity(self) -> None: + app = base_cli.App(name="authoritative-name") + + with self.assertRaisesRegex((RuntimeError, ValueError), "conflicting-name|authoritative-name"): + @app.command("conflicting-name") + def implementation(ctx: base_cli.Context) -> None: + del ctx + + def test_single_command_accepts_an_explicit_name_matching_app_identity(self) -> None: + app = base_cli.App(name="matching-name") + + @app.command("matching-name") + def implementation(ctx: base_cli.Context) -> None: + del ctx + + self.assertEqual(app.click_command.name, "matching-name") + + def test_direct_app_invocation_defaults_to_app_identity(self) -> None: + app = base_cli.App(name="direct-identity") + + @app.command() + def implementation(ctx: base_cli.Context) -> None: + del ctx + + click_command = app.click_command + with mock.patch.object(click_command, "main", return_value=0) as click_main: + result = app(["--help"]) + + self.assertEqual(result, 0) + click_main.assert_called_once_with(["--help"], prog_name="direct-identity") + + def test_direct_app_invocation_honors_delegated_display_identity(self) -> None: + profile = replace( + base_cli.CliProfile.generic(), + display_command=lambda: "launcher delegated", + ) + app = base_cli.App(name="internal-identity", profile=profile) + + @app.command() + def implementation(ctx: base_cli.Context) -> None: + del ctx + + click_command = app.click_command + with mock.patch.object(click_command, "main", return_value=0) as click_main: + result = app([]) + + self.assertEqual(result, 0) + click_main.assert_called_once_with([], prog_name="launcher delegated") + + def test_custom_single_command_class_cannot_override_app_identity(self) -> None: + import click + + class RenamingCommand(click.Command): + def __init__(self, *args: object, **kwargs: object) -> None: + super().__init__(*args, **kwargs) + self.name = "custom-override" + + app = base_cli.App(name="canonical-single") + + @app.command(cls=RenamingCommand) + def implementation(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(RuntimeError, "canonical-single|custom-override"): + _ = app.click_command + + def test_app_name_can_change_before_materialization_and_becomes_canonical(self) -> None: + app = base_cli.App(name="original-name", log_to_file=False) + + @app.command() + def implementation(ctx: base_cli.Context) -> None: + del ctx + + app.name = "renamed tool" + + self.assertEqual(app.name, "renamed-tool") + self.assertEqual(app.click_command.name, "renamed-tool") + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke(app, ["--help"], home=Path(tmpdir)) + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("Usage: renamed-tool [OPTIONS]", result.output) + self.assertNotIn("Usage: original-name", result.output) + + def test_app_name_rejects_a_rename_conflicting_with_registered_explicit_name(self) -> None: + app = base_cli.App(name="original-name") + + @app.command("original-name") + def implementation(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(RuntimeError, "original-name|renamed-name"): + app.name = "renamed-name" + + self.assertEqual(app.name, "original-name") + self.assertEqual(app.click_command.name, "original-name") + + def test_app_name_cannot_change_during_or_after_materialization(self) -> None: + mutation_errors: list[BaseException] = [] + + class RenamingApp(base_cli.App): + def _build_click_command(self) -> object: + try: + self.name = "during-build" + except BaseException as exc: # pylint: disable=broad-exception-caught + mutation_errors.append(exc) + return super()._build_click_command() + + app = RenamingApp(name="stable-name") + + @app.command() + def implementation(ctx: base_cli.Context) -> None: + del ctx + + click_command = app.click_command + + self.assertEqual(len(mutation_errors), 1) + self.assertIsInstance(mutation_errors[0], RuntimeError) + self.assertRegex(str(mutation_errors[0]), "materializ|frozen") + self.assertEqual(app.name, "stable-name") + self.assertEqual(click_command.name, "stable-name") + + with self.assertRaisesRegex(RuntimeError, "materializ|frozen"): + app.name = "after-build" + self.assertEqual(app.name, "stable-name") + self.assertIs(app.click_command, click_command) + + def test_group_uses_app_identity_for_click_help_and_version(self) -> None: + app = base_cli.App( + name="workspace-suite", + version="2.4.0", + help="Manage the workspace suite.", + log_to_file=False, + ) + + @app.subcommand() + def status(ctx: base_cli.Context) -> None: + del ctx + + click_group = app.click_command + self.assertEqual(click_group.name, "workspace-suite") + self.assertEqual(click_group.help, "Manage the workspace suite.") + self.assertEqual(set(click_group.commands), {"status"}) + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + help_result = invoke(app, ["--help"], home=home) + version_result = invoke(app, ["--version"], home=home) + + self.assertEqual(help_result.exit_code, 0, help_result.output) + self.assertIn( + "Usage: workspace-suite [OPTIONS] COMMAND [ARGS]...", + help_result.output, + ) + self.assertIn("Manage the workspace suite.", help_result.output) + self.assertIn("status", help_result.output) + self.assertEqual(version_result.exit_code, 0, version_result.output) + self.assertIn("workspace-suite, version 2.4.0", version_result.output) + + def test_inferred_subcommand_suffixes_are_click_version_independent(self) -> None: + app = base_cli.App(name="stable-names") + + @app.subcommand() + def sync_command(ctx: base_cli.Context) -> None: + del ctx + + @app.subcommand() + def inspect_cmd(ctx: base_cli.Context) -> None: + del ctx + + @app.subcommand() + def report_group(ctx: base_cli.Context) -> None: + del ctx + + @app.subcommand() + def clean_grp(ctx: base_cli.Context) -> None: + del ctx + + self.assertEqual( + set(app.click_command.commands), + {"sync", "inspect", "report", "clean"}, + ) + + def test_rejects_duplicate_explicit_subcommand_names(self) -> None: + app = base_cli.App(name="duplicate-explicit") + + @app.subcommand("deploy") + def deploy_primary(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(RuntimeError, "deploy"): + @app.subcommand(name="deploy") + def deploy_secondary(ctx: base_cli.Context) -> None: + del ctx + + def test_rejects_duplicate_inferred_subcommand_names(self) -> None: + app = base_cli.App(name="duplicate-inferred") + + def primary(ctx: base_cli.Context) -> None: + del ctx + + primary.__name__ = "status" + app.subcommand()(primary) + + def secondary(ctx: base_cli.Context) -> None: + del ctx + + secondary.__name__ = "status" + with self.assertRaisesRegex(RuntimeError, "status"): + app.subcommand()(secondary) + + def test_rejects_explicit_collision_with_inferred_suffix_name(self) -> None: + app = base_cli.App(name="duplicate-mixed") + + @app.subcommand("sync") + def explicit_sync(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(RuntimeError, "sync"): + @app.subcommand() + def sync_command(ctx: base_cli.Context) -> None: + del ctx + + def test_custom_subcommand_class_cannot_override_registered_name(self) -> None: + import click + + class RenamingCommand(click.Command): + def __init__(self, *args: object, **kwargs: object) -> None: + super().__init__(*args, **kwargs) + self.name = "custom-override" + + app = base_cli.App(name="canonical-group") + + @app.subcommand(cls=RenamingCommand) + def status(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(RuntimeError, "status|custom-override"): + _ = app.click_command + + def test_registration_factories_reject_calls_after_materialization(self) -> None: + single = base_cli.App(name="frozen-single") + + @single.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + self.assertIsNotNone(single.click_command) + with self.assertRaisesRegex(RuntimeError, "(materialized|frozen)"): + single.command() + + group = base_cli.App(name="frozen-group") + + @group.subcommand() + def status(ctx: base_cli.Context) -> None: + del ctx + + self.assertIsNotNone(group.click_command) + with self.assertRaisesRegex(RuntimeError, "(materialized|frozen)"): + group.subcommand() + + def test_deferred_registration_decorators_reject_late_application(self) -> None: + single = base_cli.App(name="deferred-single") + deferred_command = single.command() + + @single.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + self.assertIsNotNone(single.click_command) + + def late_main(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(RuntimeError, "(materialized|frozen)"): + deferred_command(late_main) + + group = base_cli.App(name="deferred-group") + deferred_subcommand = group.subcommand() + + @group.subcommand() + def status(ctx: base_cli.Context) -> None: + del ctx + + self.assertIsNotNone(group.click_command) + + def late_status(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(RuntimeError, "(materialized|frozen)"): + deferred_subcommand(late_status) + + def test_failed_empty_materialization_does_not_freeze_registration(self) -> None: + app = base_cli.App(name="recover-empty") + + with self.assertRaisesRegex(RuntimeError, "No command has been registered"): + _ = app.click_command + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + self.assertEqual(app.click_command.name, "recover-empty") + + def test_failed_plan_compilation_restores_registration(self) -> None: + app = base_cli.App(name="recover-transaction") + + @app.subcommand() + def status(ctx: base_cli.Context) -> None: + del ctx + + with mock.patch( + "base_cli.app.compile_redaction_plan", + side_effect=RuntimeError("plan compilation failed"), + ): + with self.assertRaisesRegex(RuntimeError, "plan compilation failed"): + _ = app.click_command + + @app.subcommand() + def inspect(ctx: base_cli.Context) -> None: + del ctx + + self.assertEqual(set(app.click_command.commands), {"status", "inspect"}) + + def test_materialization_and_late_registration_are_serialized(self) -> None: + app = base_cli.App(name="registration-race") + + @app.subcommand() + def status(ctx: base_cli.Context) -> None: + del ctx + + real_build = app._build_click_command # pylint: disable=protected-access + build_entered = threading.Event() + release_build = threading.Event() + materialization_errors: list[BaseException] = [] + registration_errors: list[BaseException] = [] + registration_started = threading.Event() + registration_finished = threading.Event() + + def blocked_build() -> object: + build_entered.set() + if not release_build.wait(timeout=5): + raise AssertionError("test did not release command materialization") + return real_build() + + def materialize() -> None: + try: + _ = app.click_command + except BaseException as exc: # pylint: disable=broad-exception-caught + materialization_errors.append(exc) + + def register_late() -> None: + registration_started.set() + + def inspect(ctx: base_cli.Context) -> None: + del ctx + + try: + app.subcommand()(inspect) + except BaseException as exc: # pylint: disable=broad-exception-caught + registration_errors.append(exc) + finally: + registration_finished.set() + + with mock.patch.object(app, "_build_click_command", side_effect=blocked_build): + materialize_thread = threading.Thread(target=materialize, daemon=True) + materialize_thread.start() + self.assertTrue(build_entered.wait(timeout=2), "materialization did not start") + + registration_thread = threading.Thread(target=register_late, daemon=True) + registration_thread.start() + self.assertTrue(registration_started.wait(timeout=2), "registration did not start") + registration_finished.wait(timeout=0.5) + release_build.set() + materialize_thread.join(timeout=2) + registration_thread.join(timeout=2) + + self.assertFalse(materialize_thread.is_alive(), "materialization deadlocked") + self.assertFalse(registration_thread.is_alive(), "registration deadlocked") + self.assertEqual(materialization_errors, []) + self.assertEqual(len(registration_errors), 1) + self.assertIsInstance(registration_errors[0], RuntimeError) + self.assertRegex(str(registration_errors[0]), "materialized|frozen") + self.assertEqual(set(app.click_command.commands), {"status"}) + + def test_reentrant_materialization_fails_instead_of_deadlocking(self) -> None: + class ReentrantApp(base_cli.App): + def _build_click_command(self) -> object: + return self.click_command + + app = ReentrantApp(name="reentrant-materialization") + errors: list[BaseException] = [] + + def materialize() -> None: + try: + _ = app.click_command + except BaseException as exc: # pylint: disable=broad-exception-caught + errors.append(exc) + + thread = threading.Thread(target=materialize, daemon=True) + thread.start() + thread.join(timeout=1) + + self.assertFalse(thread.is_alive(), "reentrant materialization deadlocked") + self.assertEqual(len(errors), 1) + self.assertIsInstance(errors[0], RuntimeError) + self.assertRegex(str(errors[0]), "(reentrant|materializ)") + + def test_reentrant_registration_during_materialization_is_rejected(self) -> None: + import click + + app = base_cli.App(name="reentrant-registration") + errors: list[BaseException] = [] + + def late(ctx: base_cli.Context) -> None: + del ctx + + class RegisteringCommand(click.Command): + def __init__(self, *args: object, **kwargs: object) -> None: + try: + app.subcommand()(late) + except BaseException as exc: # pylint: disable=broad-exception-caught + errors.append(exc) + super().__init__(*args, **kwargs) + + @app.subcommand(cls=RegisteringCommand) + def status(ctx: base_cli.Context) -> None: + del ctx + + click_group = app.click_command + + self.assertEqual(len(errors), 1) + self.assertIsInstance(errors[0], RuntimeError) + self.assertRegex(str(errors[0]), "materializ|frozen") + self.assertEqual(set(click_group.commands), {"status"}) + + def test_module_command_returns_original_callable_and_exposes_stable_app(self) -> None: + def script(ctx: base_cli.Context) -> None: + del ctx + + decorated = base_cli.command()(script) + + self.assertIs(decorated, script) + command_app = base_cli.get_command_app(script) + self.assertIsInstance(command_app, base_cli.App) + self.assertIs(base_cli.get_command_app(decorated), command_app) + self.assertIs(base_cli.get_command_app(script), command_app) + + def test_module_command_rejects_stacked_duplicate_registration(self) -> None: + def script(ctx: base_cli.Context) -> None: + del ctx + + decorated = base_cli.command()(script) + original_app = base_cli.get_command_app(decorated) + + with self.assertRaisesRegex(RuntimeError, "already|@base_cli.command"): + base_cli.command()(decorated) + + self.assertIs(base_cli.get_command_app(decorated), original_app) + + def test_module_command_explicit_names_seed_the_owner_app(self) -> None: + def positional(ctx: base_cli.Context) -> None: + del ctx + + def keyword(ctx: base_cli.Context) -> None: + del ctx + + positional_decorated = base_cli.command("positional-tool")(positional) + keyword_decorated = base_cli.command(name="keyword-tool")(keyword) + + self.assertIs(positional_decorated, positional) + self.assertIs(keyword_decorated, keyword) + self.assertEqual(base_cli.get_command_app(positional).name, "positional-tool") + self.assertEqual(base_cli.get_command_app(keyword).name, "keyword-tool") + self.assertEqual( + base_cli.get_command_app(positional).click_command.name, + "positional-tool", + ) + self.assertEqual( + base_cli.get_command_app(keyword).click_command.name, + "keyword-tool", + ) + + def test_module_command_inference_is_click_version_independent(self) -> None: + @base_cli.command() + def sync_command(ctx: base_cli.Context) -> None: + del ctx + + command_app = base_cli.get_command_app(sync_command) + self.assertEqual(command_app.name, "sync") + self.assertEqual(command_app.click_command.name, "sync") + + def test_module_command_runs_through_run_app_with_outer_parameter_decorator(self) -> None: + seen: dict[str, str] = {} + + @base_cli.option("--name", required=True) + @base_cli.command() + def greet(ctx: base_cli.Context, name: str) -> None: + del ctx + seen["name"] = name + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + stderr = io.StringIO() + with mock.patch.dict( + os.environ, + { + "HOME": str(home), + "USERPROFILE": str(home), + "LOCALAPPDATA": str(home / "AppData" / "Local"), + "XDG_CACHE_HOME": str(home / ".cache"), + "BASE_CLI_CACHE_DIR": str(home / ".cache"), + }, + ), redirect_stderr(stderr): + status = base_cli.run_app(greet, ["--name", "Ada"]) + + self.assertEqual(status, 0, stderr.getvalue()) + self.assertEqual(seen, {"name": "Ada"}) + + def test_module_commands_own_independent_apps(self) -> None: + @base_cli.command() + def first(ctx: base_cli.Context) -> None: + del ctx + + @base_cli.command() + def second(ctx: base_cli.Context) -> None: + del ctx + + first_app = base_cli.get_command_app(first) + second_app = base_cli.get_command_app(second) + + self.assertIsNot(first_app, second_app) + self.assertIs(base_cli.get_command_app(first), first_app) + self.assertIs(base_cli.get_command_app(second), second_app) + self.assertEqual(first_app.click_command.name, "first") + self.assertEqual(second_app.click_command.name, "second") + + def test_ordinary_callable_is_rejected_by_command_app_resolution_and_run_app(self) -> None: + def ordinary(ctx: base_cli.Context) -> None: + del ctx + + with self.assertRaisesRegex(TypeError, "@base_cli.command"): + base_cli.get_command_app(ordinary) + with self.assertRaisesRegex(TypeError, "@base_cli.command"): + base_cli.run_app(ordinary, []) + + def test_delegated_display_identity_overrides_internal_app_name(self) -> None: + profile = replace( + base_cli.CliProfile.generic(), + display_command=lambda: "launcher delegated", + ) + app = base_cli.App( + name="internal-implementation", + version="3.1.4", + help="Delegated command help.", + profile=profile, + log_to_file=False, + ) + + @app.command() + def implementation(ctx: base_cli.Context) -> None: + del ctx + + with tempfile.TemporaryDirectory() as tmpdir: + home = Path(tmpdir) + help_result = invoke(app, ["--help"], home=home) + version_result = invoke(app, ["--version"], home=home) + usage_result = invoke(app, ["--unknown"], home=home) + + self.assertEqual(help_result.exit_code, 0, help_result.output) + self.assertIn("Usage: launcher delegated [OPTIONS]", help_result.output) + self.assertNotIn("Usage: internal-implementation", help_result.output) + self.assertEqual(version_result.exit_code, 0, version_result.output) + self.assertIn("launcher delegated, version 3.1.4", version_result.output) + self.assertEqual(usage_result.exit_code, 2, usage_result.output) + self.assertIn("Usage: launcher delegated [OPTIONS]", _all_output(usage_result)) + self.assertNotIn("Usage: internal-implementation", _all_output(usage_result)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_public_api.py b/tests/test_public_api.py index a00e1f7..e8bc9ae 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -38,6 +38,7 @@ def test_facade_exports_supported_modules_functions_and_types(self) -> None: "command_protocol", "dumps_record", "dumps_records", + "get_command_app", "history", "loads_records", "normalize_command_filter", @@ -76,6 +77,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.get_command_app.__doc__) self.assertTrue(base_cli.run_app.__doc__)