diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e2cbe6..ab36bd7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,9 @@ and versions are tracked in the repo-root `VERSION` file. - Treat plain profile-callback exceptions as private internal failures. Profiles that used `ValueError` for expected configuration problems should raise `ConfigurationError` instead. +- Route `base_cli.testing.invoke()` through the production `run_app()` boundary + and add a keyword-only `reraise_unexpected` opt-in for tests that need the + original exception. ### Fixed @@ -34,6 +37,8 @@ and versions are tracked in the repo-root `VERSION` file. command outcome or skip cleanup, context reset, and logger shutdown. - Allow finished history records to omit `log_path` when file logging is disabled. +- Restore Click-native `--option=value` parsing, including redaction of + sensitive equals-form values. ## [0.3.0] - 2026-08-01 diff --git a/README.md b/README.md index 7e0aa2a..5b85df4 100644 --- a/README.md +++ b/README.md @@ -139,11 +139,10 @@ hello --keep-temp --name Ada hello --log-file /tmp/hello.log --name Ada ``` -Long options with values use space-separated syntax. `base_cli.run_app()` rejects -equals-form values such as `--name=Ada` before Click parses arguments. -These options belong to the application-level lifecycle. A consumer may expose -them through its own launcher or compose them with a higher-level command -wrapper. +Long options use Click's native syntax, so both `--name Ada` and `--name=Ada` +are accepted. These options belong to the application-level lifecycle. A +consumer may expose them through its own launcher or compose them with a +higher-level command wrapper. ## Command Registration @@ -230,9 +229,8 @@ def main(ctx: base_cli.Context, token: str) -> None: ... ``` -Both `--token secret` and an externally supplied `--token=secret` token are -redacted in debug logs. The lifecycle rejects equals-form option values before -Click parses them. +Both `--token secret` and `--token=secret` are accepted and redacted in debug +logs. Use `dry_run=True` when a nonstandard option should drive `ctx.dry_run` and the lifecycle's default durable-write suppression: @@ -300,8 +298,8 @@ available. The traceback is kept in the persistent log when enabled and is shown on stderr with an effective `--debug` setting. A failure before option parsing can provide a traceback only when `--debug` is an unambiguous leading flag; otherwise the message says that diagnostic context was unavailable. -Tests or embedding code that need the original exception can pass -`reraise_unexpected=True`. +Embedding code that needs the original exception can pass the keyword-only +`reraise_unexpected=True` argument to `run_app()`. | Command result or exception | `outcome` | Exit code | Default message | | --- | --- | ---: | --- | @@ -555,15 +553,34 @@ def test_command(tmp_path: Path) -> None: assert "hello Ada" in result.stdout ``` -The helper wraps Click's `CliRunner`, sets `HOME` plus the relevant -`USERPROFILE`, `LOCALAPPDATA`, and `XDG_CACHE_HOME` values when requested, and -supplies `cwd` to the invocation for the duration of the test. Calls that use -`cwd` are serialized and the caller's cwd is restored afterward, but this -remains process-global: do not use it concurrently with code that changes cwd -outside `invoke()` or from threads spawned by the invoked command. A -generic profile should receive project fixtures through its -`discover_project` callback. The helper does not create or interpret any -product-specific manifest fixture. +The helper wraps Click's `CliRunner` but routes the invocation through the same +`run_app()` boundary used by production entry points. Option parsing (including +native forms such as `--name=Ada`), effective and logged argv, exit-code and +error normalization, lifecycle behavior, and command-group dispatch therefore +follow the production path. + +By default, unexpected exceptions receive the production-safe rendering and +exit code. Pass the keyword-only `reraise_unexpected=True` argument when a test +needs the original unexpected exception in `result.exception`: + +```python +result = invoke(app, [], home=tmp_path, reraise_unexpected=True) +assert isinstance(result.exception, RuntimeError) +``` + +As with direct `CliRunner` use, a handled nonzero exit normally also gives +`Result.exception` a `SystemExit` carrying that exit code. That does not by +itself indicate an unexpected crash; assert `result.exit_code` and the rendered +stdout or stderr for expected usage or application failures. + +`invoke()` sets `HOME` plus the relevant `USERPROFILE`, `LOCALAPPDATA`, and +`XDG_CACHE_HOME` values when requested, and supplies `cwd` to the invocation for +the duration of the test. Calls that use `cwd` are serialized and the caller's +cwd is restored afterward, but this remains process-global: do not use it +concurrently with code that changes cwd outside `invoke()` or from threads +spawned by the invoked command. A generic profile should receive project +fixtures through its `discover_project` callback. The helper does not create or +interpret any product-specific manifest fixture. When `home` is supplied, `invoke()` provides an isolated default cache environment for tests. Pass `env={"BASE_CLI_CACHE_DIR": str(path)}` when a test diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index 6d88da0..dfda1df 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -574,7 +574,6 @@ def run_app(app: App, argv: list[str] | None = None, *, reraise_unexpected: bool state_token = _INVOCATION_STATE.set(state) try: try: - _reject_equals_option_values(click, args) display_command = app.profile.display_command() invocation_argv = _effective_invocation_argv(app, args, explicit_argv, display_command) invocation_token = _INVOCATION_ARGV.set(invocation_argv) @@ -768,21 +767,6 @@ def _validate_standard_options(click: Any, standard: dict[str, Any]) -> None: raise click.UsageError("--debug and --quiet cannot be used together.") -def _reject_equals_option_values(click: Any, argv: list[str]) -> None: - for token in argv: - if token == "--": - return - if token.startswith("--") and "=" in token and len(token) > 2: - option_name, value = token.split("=", 1) - if value: - raise click.UsageError( - f"Option '{option_name}' uses unsupported equals syntax. Use '{option_name} {value}' instead." - ) - raise click.UsageError( - f"Option '{option_name}' uses unsupported equals syntax. Pass its value as the next argument." - ) - - def _group_standard_options(click: Any) -> dict[str, Any]: context = click.get_current_context(silent=True) parent = context.parent if context is not None else None diff --git a/lib/python/base_cli/testing.py b/lib/python/base_cli/testing.py index 2740c14..c60f6d4 100644 --- a/lib/python/base_cli/testing.py +++ b/lib/python/base_cli/testing.py @@ -2,6 +2,7 @@ import inspect import os +import sys from pathlib import Path from threading import RLock from typing import TYPE_CHECKING, Any @@ -22,13 +23,19 @@ def invoke( home: Path | None = None, cwd: Path | str | None = None, env: dict[str, str] | None = None, + *, + reraise_unexpected: bool = False, ) -> Result: + """Invoke an app through the production ``run_app`` boundary.""" + cwd_path = Path(cwd).expanduser().resolve() if cwd is not None else None + invocation_argv = list(args or []) try: - from click.testing import CliRunner + from click import testing as click_testing except ImportError as exc: raise RuntimeError("Click is required for base_cli.testing. Install it with 'pip install click'.") from exc + CliRunner = click_testing.CliRunner invoke_env = dict(env or {}) if home is not None: @@ -41,15 +48,52 @@ def invoke( if "mix_stderr" in inspect.signature(CliRunner).parameters: runner_kwargs["mix_stderr"] = False runner = CliRunner(**runner_kwargs) + command = _RunAppInvocation( + app, + invocation_argv, + reraise_unexpected=reraise_unexpected, + ) if cwd_path is None: with use_working_dir(None): - return runner.invoke(app.click_command, args or [], env=invoke_env) + return runner.invoke(command, [], env=invoke_env) with _INVOKE_CWD_LOCK: with use_working_dir(cwd_path): original_cwd = Path.cwd() os.chdir(cwd_path) try: - return runner.invoke(app.click_command, args or [], env=invoke_env) + return runner.invoke(command, [], env=invoke_env) finally: os.chdir(original_cwd) + + +class _RunAppInvocation: + """Minimal command interface used only by ``CliRunner.invoke``.""" + + name = "base-cli-testing-invoke" + + def __init__( + self, + app: Any, + argv: list[str], + *, + reraise_unexpected: bool, + ) -> None: + self._app = app + self._argv = tuple(argv) + self._reraise_unexpected = reraise_unexpected + + def main(self, *_args: Any, **_kwargs: Any) -> None: + from .app import run_app + + status = run_app( + self._app, + list(self._argv), + reraise_unexpected=self._reraise_unexpected, + ) + for stream in (sys.stdout, sys.stderr): + try: + stream.flush() + except Exception: # pylint: disable=broad-exception-caught + pass + raise SystemExit(status) diff --git a/tests/test_app_lifecycle.py b/tests/test_app_lifecycle.py index d86e1e5..aa9df99 100644 --- a/tests/test_app_lifecycle.py +++ b/tests/test_app_lifecycle.py @@ -122,7 +122,8 @@ def main(ctx: base_cli.Context) -> None: result = invoke(app, [], home=Path(tmpdir)) self.assertEqual(result.exit_code, 1) - self.assertIs(result.exception, primary_failure) + self.assertIsInstance(result.exception, SystemExit) + self.assertIsNot(result.exception, primary_failure) self.assertTrue(seen["cleanup_called"]) self.assertFalse(Path(seen["temp_dir"]).exists()) self.assertEqual(seen["logger"].handlers, []) @@ -130,6 +131,28 @@ def main(ctx: base_cli.Context) -> None: with self.assertRaisesRegex(RuntimeError, "context is not active"): base_cli.get_current_context() self.assertIn("History finalization failed: history unavailable", result.stderr) + self.assertIn("Error: Unexpected internal error.", result.stderr) + self.assertNotIn("command failed", result.stderr) + + def test_invoke_can_capture_original_unexpected_exception_for_debugging(self) -> None: + app = base_cli.App(name="lifecycle-reraise", log_to_file=False) + primary_failure = _CommandFailure("command failed") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + raise primary_failure + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke( + app, + [], + home=Path(tmpdir), + reraise_unexpected=True, + ) + + self.assertEqual(result.exit_code, 1) + self.assertIs(result.exception, primary_failure) def test_non_os_temp_cleanup_failure_still_closes_handlers(self) -> None: app = base_cli.App(name="cleanup-runtime-failure") diff --git a/tests/test_app_quiet.py b/tests/test_app_quiet.py index 0de1b1f..dd7b535 100644 --- a/tests/test_app_quiet.py +++ b/tests/test_app_quiet.py @@ -68,7 +68,7 @@ def main(ctx: base_cli.Context) -> None: result = invoke(app, ["--debug", "--quiet"], home=home) self.assertEqual(result.exit_code, 2, result.output) - self.assertIn("--debug and --quiet cannot be used together", result.output) + self.assertIn("--debug and --quiet cannot be used together", result.stderr) def test_quiet_before_subcommand_uses_warning_user_stream(self) -> None: app = base_cli.App(name="quiet-subcommand", log_to_file=False) diff --git a/tests/test_app_run.py b/tests/test_app_run.py index b92d485..4ba59fb 100644 --- a/tests/test_app_run.py +++ b/tests/test_app_run.py @@ -194,7 +194,7 @@ def main(ctx: base_cli.Context) -> dict[str, str]: self.assertIn("Commands must return None or an int exit code", stderr.getvalue()) @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") - def test_run_app_rejects_equals_form_long_option_values(self) -> None: + def test_run_app_accepts_click_native_equals_form_long_option_values(self) -> None: app = base_cli.App(name="space-options", log_to_file=False) seen = {} @@ -216,13 +216,9 @@ def main(ctx: base_cli.Context, name: str) -> None: ), redirect_stderr(stderr): status = base_cli.run_app(app, ["--name=demo"]) - self.assertEqual(status, 2) - self.assertEqual(seen, {}) - self.assertIn( - "Option '--name' uses unsupported equals syntax. Use '--name demo' instead.", - stderr.getvalue(), - ) - self.assertNotIn("Traceback", stderr.getvalue()) + self.assertEqual(status, 0) + self.assertEqual(seen, {"name": "demo"}) + self.assertEqual(stderr.getvalue(), "") @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") def test_run_app_uses_delegated_display_command_for_usage_errors(self) -> None: @@ -245,7 +241,8 @@ def main(ctx: base_cli.Context) -> None: self.assertEqual(status, 2) self.assertIn("Usage: tool demo", stderr.getvalue()) - self.assertIn("No such option '--bad-option'.", stderr.getvalue()) + self.assertIn("No such option", stderr.getvalue()) + self.assertIn("--bad-option", stderr.getvalue()) self.assertNotIn("internal-cli", stderr.getvalue()) diff --git a/tests/test_invocation_parity.py b/tests/test_invocation_parity.py new file mode 100644 index 0000000..16148b8 --- /dev/null +++ b/tests/test_invocation_parity.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +import ast +import importlib.util +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout +from dataclasses import dataclass, replace +from pathlib import Path +from unittest import mock + +import base_cli +from base_cli.testing import invoke + + +@dataclass(frozen=True) +class _Observation: + exit_code: int + stdout: str + stderr: str + logged_argv: list[str] + log_text: str + metadata: dict[str, object] + home: Path + + +def _profile() -> base_cli.CliProfile: + return replace( + base_cli.CliProfile.generic(), + display_command=lambda: "parity-tool", + ) + + +def _make_app(case: str) -> base_cli.App: + app = base_cli.App(name="parity-tool", profile=_profile()) + + if case == "success": + + @app.command() + @base_cli.option("--name", required=True) + def success(ctx: base_cli.Context, name: str) -> None: + del ctx + print(f"hello {name}") + + return app + + if case == "usage": + import click + + @app.command() + @base_cli.option("--target", required=True) + def usage(ctx: base_cli.Context, target: str) -> None: + del ctx, target + raise click.UsageError("choose a valid target") + + return app + + if case == "unexpected": + + @app.command() + def unexpected(ctx: base_cli.Context) -> None: + del ctx + raise RuntimeError("private invocation detail") + + return app + + if case == "group": + + @app.subcommand("show") + @base_cli.option("--name", required=True) + def show(ctx: base_cli.Context, name: str) -> None: + print(f"{ctx.environment}:{name}") + + return app + + if case == "sensitive": + + @app.command() + @base_cli.option("--token", sensitive=True, required=True) + def sensitive(ctx: base_cli.Context, token: str) -> None: + del ctx + print(f"token-length={len(token)}") + + return app + + raise AssertionError(f"unknown parity case: {case}") + + +def _isolated_environment(home: Path) -> dict[str, str]: + return { + "HOME": str(home), + "USERPROFILE": str(home), + "LOCALAPPDATA": str(home / "AppData" / "Local"), + "XDG_CACHE_HOME": str(home / ".cache"), + "BASE_CLI_CACHE_DIR": str(home / ".cache"), + } + + +def _load_run_artifacts(home: Path) -> tuple[dict[str, object], list[str], str]: + metadata_paths = sorted((home / ".cache").glob("**/run.json")) + if len(metadata_paths) != 1: + raise AssertionError(f"expected one run.json below {home}, found {metadata_paths}") + + metadata_path = metadata_paths[0] + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + if not isinstance(metadata, dict): + raise AssertionError(f"run metadata is not an object: {metadata!r}") + if metadata.get("status") == "running": + raise AssertionError(f"run metadata was not finalized: {metadata!r}") + for key in ("outcome", "exit_code", "ended_at", "duration_ms"): + if key not in metadata: + raise AssertionError(f"terminal run metadata is missing {key!r}: {metadata!r}") + + log_path = metadata_path.parent / "logs" / "primary.log" + log_text = log_path.read_text(encoding="utf-8") + invocation_lines = [line for line in log_text.splitlines() if "argv=" in line] + if len(invocation_lines) != 1: + raise AssertionError(f"expected one logged argv line in {log_path}: {invocation_lines!r}") + logged_argv = ast.literal_eval(invocation_lines[0].split("argv=", 1)[1]) + if not isinstance(logged_argv, list) or not all(isinstance(value, str) for value in logged_argv): + raise AssertionError(f"logged argv is not a string list: {logged_argv!r}") + return metadata, logged_argv, log_text + + +def _observe_production(app: base_cli.App, args: list[str], home: Path) -> _Observation: + stdout = io.StringIO() + stderr = io.StringIO() + with mock.patch.dict(os.environ, _isolated_environment(home)), redirect_stdout(stdout), redirect_stderr(stderr): + exit_code = base_cli.run_app(app, args) + metadata, logged_argv, log_text = _load_run_artifacts(home) + return _Observation( + exit_code, + stdout.getvalue(), + stderr.getvalue(), + logged_argv, + log_text, + metadata, + home, + ) + + +def _observe_testing(app: base_cli.App, args: list[str], home: Path) -> _Observation: + result = invoke(app, args, home=home) + metadata, logged_argv, log_text = _load_run_artifacts(home) + return _Observation( + result.exit_code, + result.stdout, + result.stderr, + logged_argv, + log_text, + metadata, + home, + ) + + +def _stable_metadata(metadata: dict[str, object]) -> dict[str, object]: + dynamic_keys = {"run_id", "started_at", "ended_at", "duration_ms"} + return {key: value for key, value in metadata.items() if key not in dynamic_keys} + + +def _stable_stderr(observation: _Observation) -> str: + value = observation.stderr.replace(str(observation.home), "") + run_id = observation.metadata.get("run_id") + if isinstance(run_id, str): + value = value.replace(run_id, "") + return value + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class InvocationParityTests(unittest.TestCase): + def test_production_and_testing_match_parser_usage_errors_without_bundles(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + production_home = root / "production" + testing_home = root / "testing" + production_home.mkdir() + testing_home.mkdir() + production_stdout = io.StringIO() + production_stderr = io.StringIO() + with ( + mock.patch.dict(os.environ, _isolated_environment(production_home)), + redirect_stdout(production_stdout), + redirect_stderr(production_stderr), + ): + production_status = base_cli.run_app(_make_app("success"), []) + + testing = invoke(_make_app("success"), [], home=testing_home) + + self.assertEqual(testing.exit_code, production_status) + self.assertEqual(testing.stdout, production_stdout.getvalue()) + self.assertEqual(testing.stderr, production_stderr.getvalue()) + self.assertEqual(list(production_home.glob("**/run.json")), []) + self.assertEqual(list(testing_home.glob("**/run.json")), []) + + def test_production_and_testing_match_representative_invocations(self) -> None: + cases = ( + ("success", ["--name=Ada"]), + ("usage", ["--target", "invalid"]), + ("unexpected", []), + ("group", ["--environment=prod", "show", "--name=grouped"]), + ) + + for case, args in cases: + with self.subTest(case=case), tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + production_home = root / "production" + testing_home = root / "testing" + production_home.mkdir() + testing_home.mkdir() + + production = _observe_production(_make_app(case), args, production_home) + testing = _observe_testing(_make_app(case), args, testing_home) + + self.assertEqual(testing.exit_code, production.exit_code) + self.assertEqual(testing.stdout, production.stdout) + self.assertEqual(_stable_stderr(testing), _stable_stderr(production)) + self.assertEqual(testing.logged_argv, production.logged_argv) + self.assertEqual(_stable_metadata(testing.metadata), _stable_metadata(production.metadata)) + + def test_equals_form_sensitive_value_is_accepted_and_redacted_in_both_paths(self) -> None: + secret = "super-secret-value" + args = [f"--token={secret}"] + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + production_home = root / "production" + testing_home = root / "testing" + production_home.mkdir() + testing_home.mkdir() + + production = _observe_production(_make_app("sensitive"), args, production_home) + testing = _observe_testing(_make_app("sensitive"), args, testing_home) + + self.assertEqual(production.exit_code, 0) + self.assertEqual(testing.exit_code, production.exit_code) + self.assertEqual(testing.stdout, production.stdout) + self.assertEqual(_stable_stderr(testing), _stable_stderr(production)) + self.assertEqual( + production.logged_argv, + ["parity-tool", "--token=[REDACTED]"], + ) + self.assertEqual(testing.logged_argv, production.logged_argv) + self.assertNotIn(secret, production.log_text) + self.assertNotIn(secret, testing.log_text) + self.assertIn("[REDACTED]", production.log_text) + self.assertIn("[REDACTED]", testing.log_text) + self.assertEqual(_stable_metadata(testing.metadata), _stable_metadata(production.metadata)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_testing.py b/tests/test_testing.py index eb9379a..cae51b9 100644 --- a/tests/test_testing.py +++ b/tests/test_testing.py @@ -9,8 +9,10 @@ import threading import unittest from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace from pathlib import Path from unittest import mock + import base_cli from base_cli.testing import invoke @@ -61,6 +63,31 @@ def test_invoke_declares_click_result_return_type(self) -> None: self.assertNotEqual(return_annotation, inspect.Signature.empty) self.assertIn("Result", str(return_annotation)) + def test_invoke_exposes_keyword_only_unexpected_exception_debugging(self) -> None: + parameter = inspect.signature(invoke).parameters["reraise_unexpected"] + + self.assertIs(parameter.kind, inspect.Parameter.KEYWORD_ONLY) + self.assertIs(parameter.default, False) + + def test_invoke_reraise_preserves_click_special_exception_identity(self) -> None: + import click + + original = click.exceptions.Exit(9) + profile = replace( + base_cli.CliProfile.generic(), + display_command=lambda: (_ for _ in ()).throw(original), + ) + app = base_cli.App(name="testing-reraise-click-exit", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + + result = invoke(app, [], reraise_unexpected=True) + + self.assertEqual(result.exit_code, 1) + self.assertIs(result.exception, original) + def test_invoke_writes_manifest_fixture_into_cwd(self) -> None: app = manifest_app(name="testing-manifest", log_to_file=False) seen: dict[str, Path | None] = {}