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
15 changes: 11 additions & 4 deletions lib/python/base_cli/_lifecycle_install.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,10 @@ def _capture_lifecycle_option(
) -> Any:
source = click_context.get_parameter_source(parameter.name)
captures = click_context.meta.setdefault(_LIFECYCLE_CAPTURE_META_KEY, {})
context_values = captures.setdefault(id(click_context), {})
# Keep the context object itself as the key. ``id(context)`` values can be
# reused after Click releases a context, which could associate a later
# invocation with stale lifecycle values.
context_values = captures.setdefault(click_context, {})
Comment thread
codeforester marked this conversation as resolved.
Comment thread
codeforester marked this conversation as resolved.
context_values[key] = _RawLifecycleValue(
value=value,
source=source,
Expand Down Expand Up @@ -537,11 +540,15 @@ def _resolve_lifecycle_values(
_LIFECYCLE_RESOLUTION_META_KEY,
{},
)
# These maps live in Click's invocation-shared metadata and retain one
# context-keyed entry per context until the root invocation closes. That
# bounded, per-invocation retention is deliberate: it prevents id reuse
# without retaining state across invocations.
parent = getattr(click_context, "parent", None)
parent_resolution = resolution_map.get(id(parent)) if parent is not None else None
parent_resolution = resolution_map.get(parent) if parent is not None else None
raw = dict(parent_resolution.raw) if isinstance(parent_resolution, _LifecycleResolution) else {}
captures = click_context.meta.get(_LIFECYCLE_CAPTURE_META_KEY, {})
context_captures = captures.get(id(click_context), {})
context_captures = captures.get(click_context, {})
depth = _context_depth(click_context)

for key, binding in bindings.items():
Expand Down Expand Up @@ -576,7 +583,7 @@ def _resolve_lifecycle_values(
values=_normalize_lifecycle_values(click, raw),
raw=raw,
)
resolution_map[id(click_context)] = resolution
resolution_map[click_context] = resolution
click_context.meta[LIFECYCLE_META_KEY] = resolution.values
return resolution

Expand Down
100 changes: 100 additions & 0 deletions tests/test_click_tree_attachment.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from unittest import mock

import base_cli
from base_cli import _lifecycle_install
from base_cli._runtime import RuntimeDirectoryError
from base_cli.testing import invoke

Expand Down Expand Up @@ -63,6 +64,105 @@ def count_cleanup() -> None:

@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed")
class ClickTreeAttachmentTests(unittest.TestCase):
def test_lifecycle_metadata_is_keyed_by_context_identity(self) -> None:
class FakeContext:
def __init__(self) -> None:
self.meta: dict[object, Any] = {}

def get_parameter_source(self, name: str) -> None:
del name
return None

first = FakeContext()
second = FakeContext()
parameter = type("Parameter", (), {"name": "environment"})()

_lifecycle_install._capture_lifecycle_option(first, parameter, "first", key="environment")
_lifecycle_install._capture_lifecycle_option(second, parameter, "second", key="environment")

captures = first.meta[next(key for key in first.meta if key is not None)]
self.assertIn(first, captures)
self.assertNotIn(id(first), captures)
self.assertEqual(captures[first]["environment"].value, "first")
second_captures = second.meta[next(key for key in second.meta if key is not None)]
self.assertIn(second, second_captures)
self.assertEqual(second_captures[second]["environment"].value, "second")

def test_chained_click_contexts_keep_lifecycle_values_through_teardown(self) -> None:
import click

observed: list[tuple[str, str | None, int]] = []
closed: list[str] = []

def capture_environment(click_context: Any, parameter: Any, value: Any) -> Any:
return _lifecycle_install._capture_lifecycle_option(
click_context,
parameter,
value,
key="environment",
)

@click.group(name="pipeline", chain=True)
def pipeline() -> None:
pass

@pipeline.command(name="first")
@click.option("--environment", callback=capture_environment)
def first(environment: str | None) -> None:
del environment
click_context = click.get_current_context()
captures = click_context.meta[_lifecycle_install._LIFECYCLE_CAPTURE_META_KEY]
observed.append(
(
"first",
captures[click_context]["environment"].value,
len(captures),
)
)
click_context.call_on_close(lambda: closed.append("first"))

@pipeline.command(name="second")
@click.option("--environment", callback=capture_environment)
def second(environment: str | None) -> None:
del environment
click_context = click.get_current_context()
captures = click_context.meta[_lifecycle_install._LIFECYCLE_CAPTURE_META_KEY]
observed.append(
(
"second",
captures[click_context]["environment"].value,
len(captures),
)
)
self.assertEqual(closed, ["first"])
click_context.call_on_close(lambda: closed.append("second"))

app = base_cli.App(name="pipeline", log_to_file=False)
app.attach(pipeline)

with tempfile.TemporaryDirectory() as tmpdir:
result = invoke(
app,
[
"first",
"--environment",
"first-env",
"second",
"--environment",
"second-env",
],
home=Path(tmpdir),
)

self.assertEqual(result.exit_code, 0, result.output)
self.assertEqual(
[(name, value) for name, value, _capture_count in observed],
[("first", "first-env"), ("second", "second-env")],
)
self.assertGreaterEqual(observed[0][2], 2)
self.assertEqual(observed[0][2], observed[1][2])
self.assertEqual(closed, ["first", "second"])

def test_prebuilt_single_command_preserves_click_contract_and_lifecycle(self) -> None:
import click

Expand Down
Loading