Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 35 additions & 1 deletion lib/python/base_cli/_lifecycle_install.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from __future__ import annotations

import os
from collections.abc import Callable
from collections.abc import Callable, Mapping
from pathlib import Path
from typing import Any

Expand Down Expand Up @@ -443,6 +443,30 @@ def _prefer_lifecycle_value(
return current


def _nested_default_map_value(click_context: Any, key: str) -> Any:
"""Read a command's nested default-map value before child invocation."""

default_map = getattr(click_context, "default_map", None)
if not isinstance(default_map, Mapping):
return None
context_values = getattr(click_context, "__dict__", {})
if isinstance(context_values, Mapping):
protected = context_values.get("_protected_args", context_values.get("protected_args", ()))
else:
protected = ()
remaining = [*protected, *getattr(click_context, "args", ())]
while remaining:
command_name = remaining[0]
if not isinstance(command_name, str) or command_name.startswith(("-", "+", "/")):
break
nested = default_map.get(command_name)
if not isinstance(nested, Mapping):
break
default_map = nested
remaining = remaining[1:]
return default_map.get(key)


def _normalize_lifecycle_values(
click: Any,
raw: dict[str, _RawLifecycleValue],
Expand Down Expand Up @@ -521,6 +545,7 @@ def _resolve_lifecycle_values(
depth = _context_depth(click_context)

for key, binding in bindings.items():
candidate: _RawLifecycleValue | None
if binding.adopted:
candidate = _RawLifecycleValue(
value=getattr(click_context, "params", {}).get(binding.parameter_name),
Expand All @@ -529,6 +554,15 @@ def _resolve_lifecycle_values(
)
else:
candidate = context_captures.get(key)
mapped_value = _nested_default_map_value(click_context, binding.parameter_name)
if isinstance(mapped_value, bool):
default_map_source = getattr(getattr(click, "core", click), "ParameterSource", None)
source = getattr(default_map_source, "DEFAULT_MAP", None)
if source is not None:
candidate = _prefer_lifecycle_value(
candidate,
_RawLifecycleValue(value=mapped_value, source=source, depth=depth + 1),
)
selected = _prefer_lifecycle_value(raw.get(key), candidate)
if selected is not None:
raw[key] = selected
Expand Down
102 changes: 102 additions & 0 deletions lib/python/base_cli/_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
_INVOCATION_ARGV,
_INVOCATION_MAIN_BYPASS,
_INVOCATION_STATE,
_LIFECYCLE_CAPTURE_META_KEY,
DISPLAY_COMMAND_ENV,
App,
_InvocationState,
Expand Down Expand Up @@ -77,6 +78,8 @@ def run_app(
args,
app.lifecycle_options,
default_map=_command_default_map(command),
command=command,
prog_name=display_command or app.name,
)
invocation_token = _INVOCATION_ARGV.set(invocation_argv)
try:
Expand Down Expand Up @@ -187,6 +190,8 @@ def _json_requested(
lifecycle_options: LifecycleOptions,
*,
default_map: Mapping[str, Any] | None = None,
command: Any | None = None,
prog_name: str | None = None,
) -> bool:
option = lifecycle_options.json
if option is None:
Expand All @@ -206,6 +211,16 @@ def _json_requested(
if explicit_value is not None:
return explicit_value

# Once the command object is available, let Click resolve the option. Its
# parser knows about auto_envvar_prefix, nested default maps, callable
# defaults, and the complete boolean environment grammar (including `t`
# and `y`). This is used only for the pre-invocation capture decision; the
# real command is still parsed and invoked exactly once below.
if command is not None:
click_value = _click_lifecycle_value(command, args, option, prog_name)
if click_value is not None:
return click_value

if option.envvar is not None:
envvars = (option.envvar,) if isinstance(option.envvar, str) else option.envvar
if any(os.environ.get(name, "").lower() in {"1", "true", "yes", "on"} for name in envvars):
Expand All @@ -218,6 +233,93 @@ def _json_requested(
return option.default is True


def _click_lifecycle_value(
command: Any,
args: list[str],
option: LifecycleOption,
prog_name: str | None,
) -> bool | None:
"""Resolve a lifecycle flag with the owning Click command parser."""

contexts: list[Any] = []
current_command = command
current_args = list(args)
current_context: Any | None = None
try:
current_context = command.make_context(
prog_name,
current_args,
resilient_parsing=True,
)
contexts.append(current_context)
while current_args:
resolve_command = getattr(current_command, "resolve_command", None)
if not callable(resolve_command):
break
command_name, next_command, remaining = resolve_command(
current_context,
_remaining_context_args(current_context),
)
if command_name is None or next_command is None:
break
next_context = next_command.make_context(
command_name,
remaining,
parent=current_context,
resilient_parsing=True,
)
contexts.append(next_context)
current_command = next_command
current_context = next_context
current_args = _remaining_context_args(current_context)

destination = option.name or _option_destination(option)
for context in reversed(contexts):
params = getattr(context, "params", {})
value = params.get(destination) if isinstance(params, Mapping) else None
if isinstance(value, bool):
return value
context_default_map = getattr(context, "default_map", None)
if isinstance(context_default_map, Mapping):
mapped_value = context_default_map.get(destination)
if isinstance(mapped_value, bool):
return mapped_value
meta = getattr(context, "meta", {})
captures = meta.get(_LIFECYCLE_CAPTURE_META_KEY) if isinstance(meta, Mapping) else None
if isinstance(captures, Mapping):
for captured in captures.values():
if not isinstance(captured, Mapping):
continue
raw = captured.get("json")
raw_value = getattr(raw, "value", None)
if isinstance(raw_value, bool):
return raw_value
except (Exception, SystemExit):
# Invalid command lines still need the lightweight explicit-token
# detector above so Click can render its normal machine error. A
# resilient parse may not be able to resolve a leaf command; in that
# case retain the existing fallback behavior.
return None
finally:
for context in reversed(contexts):
close = getattr(context, "close", None)
if callable(close):
close()
return None


def _remaining_context_args(context: Any) -> list[str]:
"""Return unparsed group/command arguments without Click deprecation warnings."""

values = getattr(context, "__dict__", {})
if isinstance(values, Mapping):
protected = values.get("_protected_args", values.get("protected_args", ()))
else:
protected = ()
args = getattr(context, "args", ())
return [*protected, *args]


def _option_destination(option: LifecycleOption) -> str:
for declaration in option.param_decls:
if declaration.startswith("--"):
Expand Down
93 changes: 93 additions & 0 deletions tests/test_json_contracts.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,99 @@ def main(ctx: base_cli.Context) -> None:
self.assertEqual(envelope["type"], "success")
self.assertEqual(envelope["details"]["stdout"], "hello from env\n")

def test_json_click_parser_handles_boolean_envvar_spellings(self) -> None:
app = base_cli.App(
name="json-envvar-spellings",
log_to_file=False,
lifecycle_options=base_cli.LifecycleOptions(
json=base_cli.LifecycleOption("--json", envvar="BASE_JSON_MODE"),
),
)

@app.command()
def main(ctx: base_cli.Context) -> None:
del ctx
print("hello from env")

for value in ("t", "y"):
with self.subTest(value=value), tempfile.TemporaryDirectory() as home:
with mock.patch.dict(os.environ, {"BASE_JSON_MODE": value}, clear=False):
result = base_cli.testing.invoke(app, [], home=Path(home))
self.assertEqual(result.exit_code, 0, result.output)
envelope = json.loads(result.stdout)
self.assertEqual(envelope["details"]["stdout"], "hello from env\n")

def test_json_click_parser_handles_auto_envvar_prefix(self) -> None:
app = base_cli.App(
name="json-auto-envvar",
log_to_file=False,
lifecycle_options=base_cli.LifecycleOptions(
json=base_cli.LifecycleOption("--json"),
),
)

@app.command(context_settings={"auto_envvar_prefix": "TOOL"})
def main(ctx: base_cli.Context) -> None:
del ctx
print("hello from auto env")

with tempfile.TemporaryDirectory() as home:
with mock.patch.dict(os.environ, {"TOOL_JSON": "1"}, clear=False):
result = base_cli.testing.invoke(app, [], home=Path(home))
self.assertEqual(result.exit_code, 0, result.output)
envelope = json.loads(result.stdout)
self.assertEqual(envelope["details"]["stdout"], "hello from auto env\n")

def test_json_click_parser_handles_callable_defaults(self) -> None:
app = base_cli.App(
name="json-callable-default",
log_to_file=False,
lifecycle_options=base_cli.LifecycleOptions(
json=base_cli.LifecycleOption("--json", default=lambda: True),
),
)

@app.command()
def main(ctx: base_cli.Context) -> None:
del ctx
print("hello from default")

with tempfile.TemporaryDirectory() as home:
result = base_cli.testing.invoke(app, [], home=Path(home))
self.assertEqual(result.exit_code, 0, result.output)
envelope = json.loads(result.stdout)
self.assertEqual(envelope["details"]["stdout"], "hello from default\n")

def test_json_click_parser_handles_nested_default_map(self) -> None:
import click

app = base_cli.App(
name="json-nested-default-map",
log_to_file=False,
lifecycle_options=base_cli.LifecycleOptions(
json=base_cli.LifecycleOption("--json"),
),
)

@click.group(
name="json-nested-default-map",
context_settings={"default_map": {"status": {"json": True}}},
)
def command() -> None:
pass

@command.command()
def status() -> None:
print("hello from leaf")

app.attach(command)

with tempfile.TemporaryDirectory() as home:
result = base_cli.testing.invoke(command, ["status"], home=Path(home))
self.assertEqual(result.exit_code, 0, result.output)
envelope = json.loads(result.stdout)
self.assertEqual(envelope["details"]["stdout"], "hello from leaf\n")

def test_json_mode_captures_combined_short_flags(self) -> None:
app = base_cli.App(
name="json-combined-short",
Expand Down
Loading