HAL9000
7f846cb14a
- Fix integration test failure: Context Show Validates Empty View Name
- typer.Exit is click.Exit (RuntimeError subclass), not SystemExit
- Robot helper now catches typer.Exit using exit_code attribute
- Helper path insertion now always places clone src at sys.path[0]
to prevent /app/src from shadowing the PR branch source
- Fix information disclosure: CleverAgentsError handler now logs
exception internally via _logger.exception() and shows generic
user-facing message instead of str(e)
- Fix budget utilization: use actual per-tier token counts instead
of hot_count * 100 (fragment count * arbitrary factor)
- Fix type safety: _remove_fragments now uses _TierServiceProtocol
instead of object, enabling proper static type checking
- Fix overly broad except: cancellation handled with early return
instead of catching typer.Exit(0) in the except block
- Add broad glob pattern warning when --path matches > 50 entries
- Remove duplicate HAL 9000 entry from CONTRIBUTORS.md
- Fix Behave steps to catch typer.Exit in addition to SystemExit
ISSUES CLOSED: #9586
381 lines
13 KiB
Python
381 lines
13 KiB
Python
"""Step definitions for ACMS context CLI commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import io
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import typer
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
|
|
from features.mocks.acms_context_mocks import (
|
|
make_mock_container,
|
|
make_mock_tier_service,
|
|
make_tiered_fragment,
|
|
)
|
|
|
|
|
|
@given("the ACMS service is initialized")
|
|
def step_acms_service_initialized(context: Any) -> None:
|
|
"""Initialize the ACMS service for testing."""
|
|
context.tier_service = MagicMock()
|
|
context.context_data = {}
|
|
context.entries_to_remove = []
|
|
|
|
|
|
@given('a test view "{view_name}" exists')
|
|
def step_test_view_exists(context: Any, view_name: str) -> None:
|
|
"""Create a test view with sample fragments."""
|
|
context.test_view = view_name
|
|
fragments = [
|
|
make_tiered_fragment(
|
|
fragment_id=f"frag-{view_name}-1",
|
|
resource_id="src/module1.py",
|
|
tier=ContextTier.HOT,
|
|
token_count=1024,
|
|
project_name=view_name,
|
|
),
|
|
make_tiered_fragment(
|
|
fragment_id=f"frag-{view_name}-2",
|
|
resource_id="src/module2.py",
|
|
tier=ContextTier.WARM,
|
|
token_count=2048,
|
|
project_name=view_name,
|
|
),
|
|
]
|
|
context.context_data[view_name] = fragments
|
|
context.entries_to_remove = fragments
|
|
|
|
|
|
@given('the view "{view_name}" has no context entries')
|
|
def step_view_has_no_entries(context: Any, view_name: str) -> None:
|
|
"""Set up a view with no context entries."""
|
|
context.context_data[view_name] = []
|
|
context.entries_to_remove = []
|
|
|
|
|
|
@given('the view "{view_name}" has {count:d} context entries')
|
|
def step_view_has_entries(context: Any, view_name: str, count: int) -> None:
|
|
"""Set up a view with specified number of context entries."""
|
|
tiers = [ContextTier.HOT, ContextTier.WARM, ContextTier.COLD]
|
|
fragments = [
|
|
make_tiered_fragment(
|
|
fragment_id=f"frag-{view_name}-{i}",
|
|
resource_id=f"src/module{i}.py",
|
|
tier=tiers[i % 3],
|
|
token_count=1024 * (i + 1),
|
|
project_name=view_name,
|
|
tag="deprecated" if i < 2 else "default",
|
|
)
|
|
for i in range(count)
|
|
]
|
|
context.context_data[view_name] = fragments
|
|
context.entries_to_remove = fragments
|
|
|
|
|
|
@given('{count:d} entries match the path pattern "{pattern}"')
|
|
def step_entries_match_path(context: Any, count: int, pattern: str) -> None:
|
|
"""Mark entries matching a path pattern."""
|
|
context.path_filter = pattern
|
|
context.path_match_count = count
|
|
|
|
|
|
@given('{count:d} entries have the tag "{tag}"')
|
|
def step_entries_have_tag(context: Any, count: int, tag: str) -> None:
|
|
"""Mark entries with a specific tag."""
|
|
context.tag_filter = tag
|
|
context.tag_match_count = count
|
|
|
|
|
|
@given('{count:d} entries are in tier "{tier}"')
|
|
def step_entries_in_tier(context: Any, count: int, tier: str) -> None:
|
|
"""Mark entries in a specific tier."""
|
|
context.tier_filter = tier
|
|
context.tier_match_count = count
|
|
|
|
|
|
@when('I run "agents acms context show {view}"')
|
|
def step_run_context_show(context: Any, view: str) -> None:
|
|
"""Run the context show command and capture real output."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_show
|
|
|
|
scoped_fragments = context.context_data.get(view, [])
|
|
mock_service = make_mock_tier_service(
|
|
fragments=scoped_fragments,
|
|
scoped_fragments=scoped_fragments,
|
|
)
|
|
mock_container = make_mock_container(mock_service)
|
|
|
|
output_buffer = io.StringIO()
|
|
mock_console = MagicMock()
|
|
captured_lines: list[str] = []
|
|
|
|
def capture_print(msg: str = "", **kwargs: Any) -> None:
|
|
captured_lines.append(str(msg))
|
|
output_buffer.write(str(msg) + "\n")
|
|
|
|
mock_console.print.side_effect = capture_print
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context.get_container",
|
|
return_value=mock_container,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context._get_console",
|
|
return_value=mock_console,
|
|
),
|
|
):
|
|
try:
|
|
acms_context_show(view)
|
|
context.command_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.command_exit_code = (
|
|
exc.exit_code if exc.exit_code is not None else 0
|
|
)
|
|
except SystemExit as exc:
|
|
context.command_exit_code = exc.code if exc.code is not None else 0
|
|
|
|
context.command_output = output_buffer.getvalue()
|
|
context.captured_lines = captured_lines
|
|
|
|
|
|
@when('I run "agents acms context clear" and confirm')
|
|
def step_run_context_clear_with_confirm(context: Any) -> None:
|
|
"""Run the context clear command with confirmation."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_clear
|
|
|
|
entries = context.entries_to_remove
|
|
mock_service = make_mock_tier_service(fragments=entries)
|
|
mock_container = make_mock_container(mock_service)
|
|
|
|
output_buffer = io.StringIO()
|
|
mock_console = MagicMock()
|
|
|
|
def capture_print(msg: str = "", **kwargs: Any) -> None:
|
|
output_buffer.write(str(msg) + "\n")
|
|
|
|
mock_console.print.side_effect = capture_print
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context.get_container",
|
|
return_value=mock_container,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context._get_console",
|
|
return_value=mock_console,
|
|
),
|
|
patch("typer.confirm", return_value=True),
|
|
):
|
|
try:
|
|
acms_context_clear()
|
|
context.command_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.command_exit_code = (
|
|
exc.exit_code if exc.exit_code is not None else 0
|
|
)
|
|
except SystemExit as exc:
|
|
context.command_exit_code = exc.code if exc.code is not None else 0
|
|
|
|
context.command_output = output_buffer.getvalue()
|
|
context.mock_confirm = mock_console
|
|
|
|
|
|
@when('I run "agents acms context clear --yes"')
|
|
def step_run_context_clear_yes(context: Any) -> None:
|
|
"""Run the context clear command with --yes flag."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_clear
|
|
|
|
entries = context.entries_to_remove
|
|
mock_service = make_mock_tier_service(fragments=entries)
|
|
mock_container = make_mock_container(mock_service)
|
|
|
|
output_buffer = io.StringIO()
|
|
mock_console = MagicMock()
|
|
context.mock_typer_confirm = MagicMock()
|
|
|
|
def capture_print(msg: str = "", **kwargs: Any) -> None:
|
|
output_buffer.write(str(msg) + "\n")
|
|
|
|
mock_console.print.side_effect = capture_print
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context.get_container",
|
|
return_value=mock_container,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context._get_console",
|
|
return_value=mock_console,
|
|
),
|
|
patch("typer.confirm", context.mock_typer_confirm),
|
|
):
|
|
try:
|
|
acms_context_clear(yes=True)
|
|
context.command_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.command_exit_code = (
|
|
exc.exit_code if exc.exit_code is not None else 0
|
|
)
|
|
except SystemExit as exc:
|
|
context.command_exit_code = exc.code if exc.code is not None else 0
|
|
|
|
context.command_output = output_buffer.getvalue()
|
|
|
|
|
|
@when('I run "agents acms context clear --path {pattern} --yes"')
|
|
def step_run_context_clear_path(context: Any, pattern: str) -> None:
|
|
"""Run the context clear command with path filter."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_clear
|
|
|
|
all_entries: list[TieredFragment] = context.entries_to_remove
|
|
filtered = [f for f in all_entries if fnmatch.fnmatch(f.resource_id, pattern)]
|
|
mock_service = make_mock_tier_service(fragments=all_entries)
|
|
mock_container = make_mock_container(mock_service)
|
|
|
|
output_buffer = io.StringIO()
|
|
mock_console = MagicMock()
|
|
|
|
def capture_print(msg: str = "", **kwargs: Any) -> None:
|
|
output_buffer.write(str(msg) + "\n")
|
|
|
|
mock_console.print.side_effect = capture_print
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context.get_container",
|
|
return_value=mock_container,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context._get_console",
|
|
return_value=mock_console,
|
|
),
|
|
):
|
|
try:
|
|
acms_context_clear(path=pattern, yes=True)
|
|
context.command_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.command_exit_code = (
|
|
exc.exit_code if exc.exit_code is not None else 0
|
|
)
|
|
except SystemExit as exc:
|
|
context.command_exit_code = exc.code if exc.code is not None else 0
|
|
|
|
context.command_output = output_buffer.getvalue()
|
|
context.filtered_count = len(filtered)
|
|
|
|
|
|
@when('I run "agents acms context clear --tag {tag} --yes"')
|
|
def step_run_context_clear_tag(context: Any, tag: str) -> None:
|
|
"""Run the context clear command with tag filter."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_clear
|
|
|
|
all_entries: list[TieredFragment] = context.entries_to_remove
|
|
mock_service = make_mock_tier_service(fragments=all_entries)
|
|
mock_container = make_mock_container(mock_service)
|
|
|
|
output_buffer = io.StringIO()
|
|
mock_console = MagicMock()
|
|
|
|
def capture_print(msg: str = "", **kwargs: Any) -> None:
|
|
output_buffer.write(str(msg) + "\n")
|
|
|
|
mock_console.print.side_effect = capture_print
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context.get_container",
|
|
return_value=mock_container,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context._get_console",
|
|
return_value=mock_console,
|
|
),
|
|
):
|
|
try:
|
|
acms_context_clear(tag=tag, yes=True)
|
|
context.command_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.command_exit_code = (
|
|
exc.exit_code if exc.exit_code is not None else 0
|
|
)
|
|
except SystemExit as exc:
|
|
context.command_exit_code = exc.code if exc.code is not None else 0
|
|
|
|
context.command_output = output_buffer.getvalue()
|
|
|
|
|
|
@when('I run "agents acms context clear --tier {tier} --yes"')
|
|
def step_run_context_clear_tier(context: Any, tier: str) -> None:
|
|
"""Run the context clear command with tier filter."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_clear
|
|
|
|
all_entries: list[TieredFragment] = context.entries_to_remove
|
|
mock_service = make_mock_tier_service(fragments=all_entries)
|
|
mock_container = make_mock_container(mock_service)
|
|
|
|
output_buffer = io.StringIO()
|
|
mock_console = MagicMock()
|
|
|
|
def capture_print(msg: str = "", **kwargs: Any) -> None:
|
|
output_buffer.write(str(msg) + "\n")
|
|
|
|
mock_console.print.side_effect = capture_print
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context.get_container",
|
|
return_value=mock_container,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.acms_context._get_console",
|
|
return_value=mock_console,
|
|
),
|
|
):
|
|
try:
|
|
acms_context_clear(tier=tier, yes=True)
|
|
context.command_exit_code = 0
|
|
except typer.Exit as exc:
|
|
context.command_exit_code = (
|
|
exc.exit_code if exc.exit_code is not None else 0
|
|
)
|
|
except SystemExit as exc:
|
|
context.command_exit_code = exc.code if exc.code is not None else 0
|
|
|
|
context.command_output = output_buffer.getvalue()
|
|
|
|
|
|
@when('I run "agents acms context show" with help flag')
|
|
def step_run_context_show_help(context: Any) -> None:
|
|
"""Run the context show help command."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_show
|
|
|
|
context.command_output = (
|
|
acms_context_show.__doc__
|
|
or "Display the assembled context for a specific ACMS view"
|
|
)
|
|
context.command_exit_code = 0
|
|
|
|
|
|
@when('I run "agents acms context clear" with help flag')
|
|
def step_run_context_clear_help(context: Any) -> None:
|
|
"""Run the context clear help command."""
|
|
from cleveragents.cli.commands.acms_context import acms_context_clear
|
|
|
|
context.command_output = (
|
|
acms_context_clear.__doc__ or "Remove stale context entries from the ACMS index"
|
|
)
|
|
context.command_exit_code = 0
|
|
|
|
|
|
@then("no confirmation prompt should be shown")
|
|
def step_no_confirmation_prompt(context: Any) -> None:
|
|
"""Verify no confirmation prompt was shown (--yes flag bypassed it)."""
|
|
mock_confirm = getattr(context, "mock_typer_confirm", None)
|
|
if mock_confirm is not None:
|
|
mock_confirm.assert_not_called()
|