"""Step definitions for project_context_cli.feature. Tests the ``agents project context set/show/inspect/simulate`` CLI commands using an in-memory SQLite database. """ from __future__ import annotations import json from contextlib import suppress from io import StringIO from typing import Any from unittest.mock import MagicMock, patch # third-party import yaml from behave import given, then, when # type: ignore[import-untyped] from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker # ------------------------------------------------------------------ # Session wrapper (no-op close) # ------------------------------------------------------------------ class _UnclosableSession: """Wraps a SQLAlchemy Session; ``close()`` is a no-op.""" def __init__(self, real_session: Session) -> None: object.__setattr__(self, "_real", real_session) def close(self) -> None: """No-op.""" def __getattr__(self, name: str) -> Any: return getattr(object.__getattribute__(self, "_real"), name) def __setattr__(self, name: str, value: Any) -> None: setattr(object.__getattribute__(self, "_real"), name, value) def _make_ctx_session_factory( context: Any, ) -> tuple[Any, Any]: """In-memory SQLite with all tables.""" from cleveragents.infrastructure.database.models import Base engine = create_engine( "sqlite:///:memory:", echo=False, connect_args={"check_same_thread": False}, ) Base.metadata.create_all(engine) real_session = sessionmaker( bind=engine, expire_on_commit=False, autoflush=True, autocommit=False, )() wrapper = _UnclosableSession(real_session) def _factory() -> Any: return wrapper return engine, _factory # ------------------------------------------------------------------ # Background # ------------------------------------------------------------------ @given("a project context CLI in-memory database is initialized") def step_init_ctx_cli_db(context: Any) -> None: from cleveragents.infrastructure.database.repositories import ( NamespacedProjectRepository, ) engine, session_factory = _make_ctx_session_factory(context) context.ctx_engine = engine context.ctx_session_factory = session_factory context.ctx_project_repo = NamespacedProjectRepository( session_factory=session_factory ) context.ctx_output = "" context.ctx_exit_code = 0 context.ctx_error = None @given('a project "{name}" exists for context CLI') def step_create_project_for_ctx(context: Any, name: str) -> None: from cleveragents.domain.models.core.project import ( NamespacedProject, parse_namespaced_name, ) parsed = parse_namespaced_name(name) proj = NamespacedProject( name=parsed.name, namespace=parsed.namespace, ) context.ctx_project_repo.create(proj) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _mock_container(context: Any) -> MagicMock: """Build a mock container wired to the test DB.""" from cleveragents.application.services.context_tiers import ( ContextTierService, ) mock_container = MagicMock() mock_container.namespaced_project_repo.return_value = context.ctx_project_repo mock_container.session_factory.return_value = context.ctx_session_factory # Provide a real (in-memory) ContextTierService for inspect/simulate if not hasattr(context, "ctx_tier_service"): context.ctx_tier_service = ContextTierService() mock_container.context_tier_service.return_value = context.ctx_tier_service return mock_container def _run_with_container(context: Any, func: Any, *args: Any, **kwargs: Any) -> None: """Execute a CLI function with a mocked DI container.""" import typer from rich.console import Console as RichConsole mock_cont = _mock_container(context) buf = StringIO() # force_terminal=False and highlight=False prevent Rich from emitting # ANSI escape codes when FORCE_COLOR is set or a real colour terminal # is detected. test_console = RichConsole( file=buf, no_color=True, highlight=False, force_terminal=False, width=500, soft_wrap=True, ) def _test_format_output(data, format_type, *args, **kwargs): import sys as _sys from io import StringIO as _SIO from cleveragents.cli.formatting import format_output as _fo _b = _SIO() _old = _sys.stdout _sys.stdout = _b try: _r = _fo(data, format_type, *args, **kwargs) finally: _sys.stdout = _old return _r or _b.getvalue().rstrip("\n") with ( patch( "cleveragents.application.container.get_container", return_value=mock_cont, ), patch( "cleveragents.cli.commands.project_context.console", test_console, ), patch( "cleveragents.cli.commands.project_context.err_console", test_console, ), patch( "cleveragents.cli.commands.project_context.format_output", _test_format_output, ), ): try: func(*args, **kwargs) context.ctx_exit_code = 0 except typer.Exit as exc: context.ctx_exit_code = exc.exit_code except typer.Abort: context.ctx_exit_code = 1 except NotImplementedError as exc: context.ctx_error = exc context.ctx_exit_code = 99 context.ctx_output = buf.getvalue() # ------------------------------------------------------------------ # context show # ------------------------------------------------------------------ @when('I run context show on "{project}" without options') def step_ctx_show(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import ( context_show, ) kwargs: dict[str, Any] = {} fmt_override = getattr(context, "ctx_show_format", None) if fmt_override is not None: kwargs["output_format"] = fmt_override del context.ctx_show_format _run_with_container(context, context_show, project=project, **kwargs) @when('I run context show on "{project}" with view "{view}"') def step_ctx_show_view(context: Any, project: str, view: str) -> None: from cleveragents.cli.commands.project_context import ( context_show, ) kwargs: dict[str, Any] = {} fmt_override = getattr(context, "ctx_show_format", None) if fmt_override is not None: kwargs["output_format"] = fmt_override del context.ctx_show_format _run_with_container(context, context_show, project=project, view=view, **kwargs) @when('I run context show on "{project}" with format "{fmt}"') def step_ctx_show_format(context: Any, project: str, fmt: str) -> None: from cleveragents.cli.commands.project_context import ( context_show, ) _run_with_container( context, context_show, project=project, output_format=fmt, ) # ------------------------------------------------------------------ # context inspect / simulate stubs # ------------------------------------------------------------------ @when('I run context inspect on "{project}"') def step_ctx_inspect(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import ( context_inspect, ) _run_with_container(context, context_inspect, project=project) @when('I run context simulate on "{project}"') def step_ctx_simulate(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import ( context_simulate, ) _run_with_container(context, context_simulate, project=project) # ------------------------------------------------------------------ # context show output format override # ------------------------------------------------------------------ @given('the context show output format is "{fmt}"') def step_set_ctx_show_format(context: Any, fmt: str) -> None: context.ctx_show_format = fmt # ------------------------------------------------------------------ # Generic Then assertions shared across context commands # ------------------------------------------------------------------ @then("the context show command should succeed") def step_ctx_show_ok(context: Any) -> None: assert context.ctx_exit_code == 0, ( f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}" ) @then("the context show command should fail") def step_ctx_show_fail(context: Any) -> None: assert context.ctx_exit_code != 0, ( f"Expected non-zero exit, got {context.ctx_exit_code}" ) @then("the context inspect command should succeed") def step_ctx_inspect_ok(context: Any) -> None: assert context.ctx_exit_code == 0, ( f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}" ) @then("the context simulate command should succeed") def step_ctx_simulate_ok(context: Any) -> None: assert context.ctx_exit_code == 0, ( f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}" ) @then('a NotImplementedError should be raised with message containing "{msg}"') def step_not_implemented(context: Any, msg: str) -> None: assert context.ctx_error is not None, "No error was raised" assert isinstance(context.ctx_error, NotImplementedError), ( f"Expected NotImplementedError, got {type(context.ctx_error)}" ) assert msg in str(context.ctx_error), f"'{msg}' not in '{context.ctx_error}'" @then('the resolved view should include resource "{res}"') def step_resolved_view_include_res(context: Any, res: str) -> None: """Verify that a resource appears in at least one resolved view.""" from cleveragents.cli.commands.project_context import _read_policy policy = _read_policy(context.ctx_session_factory, "local/ctx-app") found = False for phase in ["default", "strategize", "execute", "apply"]: resolved = policy.resolve_view(phase) if res in resolved.include_resources: found = True break assert found, f"Resource '{res}' not found in any resolved view" @then("the context output should be valid JSON") def step_ctx_output_valid_json(context: Any) -> None: try: json.loads(context.ctx_output.strip()) except json.JSONDecodeError as exc: raise AssertionError( f"Output is not valid JSON: {exc}\nOutput: {context.ctx_output!r}" ) from exc # ------------------------------------------------------------------ # Issue #6323 spec compliance helpers # ------------------------------------------------------------------ @given('issue 6323 project context sample is configured for "{project}"') def step_issue6323_configure_policy(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import ( _format_size as _format_size_helper, ) from cleveragents.cli.commands.project_context import context_set max_file = 1_048_576 max_total = 50 * 1_048_576 _run_with_container( context, context_set, project=project, view="strategize", include_resource=["repo"], exclude_path=["**/node_modules/**"], hot_max_tokens=12_000, warm_max_decisions=50, cold_max_decisions=200, query_limit=20, summarize=True, summary_max_tokens=800, max_file_size=max_file, max_total_size=max_total, ) context.issue6323_expected = { "context_policy": { "project": project, "view": "strategize", "include_resources": ["repo"], "exclude_resources": [], "include_paths": [], "exclude_paths": ["**/node_modules/**"], }, "limits": { "hot_max_tokens": 12_000, "warm_max_decisions": 50, "cold_max_decisions": 200, "query_limit": 20, "max_file_size": _format_size_helper(max_file), "max_total_size": _format_size_helper(max_total), }, "summarization": { "enabled": True, "max_tokens": 800, }, "current_usage": { "hot_context_tokens": 0, "hot_context_limit": 12_000, "warm_entries": 0, "warm_limit": 50, "cold_entries": 0, "cold_limit": 200, "indexed_resources": 0, }, } @given('the context tier service has sample usage data for "{project}"') def step_issue6323_seed_usage(context: Any, project: str) -> None: from cleveragents.application.services.context_tiers import ContextTierService from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment svc = getattr(context, "ctx_tier_service", ContextTierService()) context.ctx_tier_service = svc svc._hot.clear() svc._warm.clear() svc._cold.clear() with suppress(Exception): svc._budget.max_tokens_hot = 12_000 # pragma: no cover - fallback guard svc.store( TieredFragment( fragment_id="hot-frag", content="hot context fragment", tier=ContextTier.HOT, project_name=project, resource_id="res:repo", token_count=8_420, ) ) for idx in range(12): svc.store( TieredFragment( fragment_id=f"warm-{idx}", content="warm fragment", tier=ContextTier.WARM, project_name=project, resource_id="res:repo", token_count=10, ) ) for idx in range(47): svc.store( TieredFragment( fragment_id=f"cold-{idx}", content="cold fragment", tier=ContextTier.COLD, project_name=project, resource_id="res:repo", token_count=5, ) ) context.issue6323_expected["current_usage"].update( { "hot_context_tokens": "***REDACTED***", "warm_entries": 12, "cold_entries": 47, "indexed_resources": 1, } ) def _issue6323_parse_json(context: Any) -> dict[str, Any]: raw = context.ctx_output.strip() return json.loads(raw) def _issue6323_parse_yaml(context: Any) -> dict[str, Any]: raw = context.ctx_output.strip() return yaml.safe_load(raw) def _issue6323_assert_payload( parsed: dict[str, Any], expected: dict[str, Any], project: str, view: str ) -> None: assert parsed["command"] == "project context show" assert parsed["status"] == "ok" assert parsed["exit_code"] == 0 assert "timing" in parsed and "duration_ms" in parsed["timing"] assert parsed["messages"] == ["Context policy loaded"] payload = parsed.get("data") assert payload is not None, "Expected 'data' field in output payload" context_policy = payload.get("context_policy") assert context_policy == expected["context_policy"], ( f"Unexpected context_policy payload. Expected {expected['context_policy']}, got {context_policy}" ) limits = payload.get("limits") assert limits == expected["limits"], ( f"Unexpected limits payload. Expected {expected['limits']}, got {limits}" ) summarization = payload.get("summarization") assert summarization == expected["summarization"], ( "Unexpected summarization payload. " f"Expected {expected['summarization']}, got {summarization}" ) usage = payload.get("current_usage") assert usage == expected["current_usage"], ( f"Unexpected current_usage payload. Expected {expected['current_usage']}, got {usage}" ) @then( 'the context show json output should match issue 6323 spec for "{project}" and view "{view}"' ) def step_issue6323_assert_json(context: Any, project: str, view: str) -> None: parsed = _issue6323_parse_json(context) _issue6323_assert_payload(parsed, context.issue6323_expected, project, view) @then( 'the context show yaml output should match issue 6323 spec for "{project}" and view "{view}"' ) def step_issue6323_assert_yaml(context: Any, project: str, view: str) -> None: parsed = _issue6323_parse_yaml(context) _issue6323_assert_payload(parsed, context.issue6323_expected, project, view) # -------------------------------------------------------------------------- # Issue 6323 coverage boost — _format_size edge cases # -------------------------------------------------------------------------- @when('I invoke the _format_size helper with "{input_val}"') def step_invoke_format_size_helper(context: Any, input_val: str) -> None: from cleveragents.cli.commands.project_context import _format_size val: int | None = None if input_val == "None" else int(input_val) context.format_size_result = _format_size(val) @then('the _format_size result should be "{expected}"') def step_assert_format_size_helper_result(context: Any, expected: str) -> None: assert context.format_size_result == expected, ( f"Expected {expected!r}, got {context.format_size_result!r}" ) # -------------------------------------------------------------------------- # Issue 6323 coverage boost — format_output dict message paths # -------------------------------------------------------------------------- @when('I invoke format_output json with dict message level "{level}" text "{text}"') def step_invoke_format_output_dict_msg(context: Any, level: str, text: str) -> None: import io import sys from cleveragents.cli.formatting import format_output buf = io.StringIO() old = sys.stdout sys.stdout = buf try: format_output( {"k": "v"}, "json", command="cov-test", status="ok", exit_code=0, messages=[{"level": level, "text": text}], ) finally: sys.stdout = old context.cov_json_output = buf.getvalue() @then('the json envelope contains dict message with level "{level}" text "{text}"') def step_assert_json_envelope_dict_msg(context: Any, level: str, text: str) -> None: parsed = json.loads(context.cov_json_output.strip()) msgs = parsed.get("messages", []) assert any( isinstance(m, dict) and m.get("level") == level and m.get("text") == text for m in msgs ), f"Expected dict message with level={level!r} text={text!r} in {msgs!r}" @when("I invoke format_output json with malformed dict message") def step_invoke_format_output_malformed_dict(context: Any) -> None: import io import sys from cleveragents.cli.formatting import format_output buf = io.StringIO() old = sys.stdout sys.stdout = buf context.cov_raised = None try: format_output( {"k": "v"}, "json", command="cov-test", status="ok", exit_code=0, messages=[{"bad_key": "missing level and text"}], ) except ValueError as exc: context.cov_raised = str(exc) finally: sys.stdout = old @then("a ValueError is raised for the malformed message") def step_assert_value_error_malformed_msg(context: Any) -> None: assert context.cov_raised is not None, "Expected ValueError but none was raised" @when('I invoke format_output plain with empty-level dict message text "{text}"') def step_invoke_format_output_plain_empty_level(context: Any, text: str) -> None: import io import sys from cleveragents.cli.formatting import format_output buf = io.StringIO() old = sys.stdout sys.stdout = buf try: format_output( {}, "plain", command="cov-test", status="ok", exit_code=0, messages=[{"level": "", "text": text}], ) finally: sys.stdout = old context.cov_plain_output = buf.getvalue() @then('the plain output contains "{text}" without bracket prefix') def step_assert_plain_no_bracket_prefix(context: Any, text: str) -> None: output = context.cov_plain_output assert text in output, f"Expected {text!r} in plain output {output!r}" for line in output.strip().split("\n"): if text in line: assert not line.startswith("["), f"Line should not start with '[': {line!r}" # -------------------------------------------------------------------------- # Issue 6323 coverage boost — rich output paths (execution env, depth gradient) # -------------------------------------------------------------------------- _COV_ACMS_BASE: dict[str, Any] = { "hot_max_tokens": 8192, "warm_max_decisions": 50, "cold_max_decisions": 200, "summarize": True, "temporal_scope": "current", "auto_refresh": True, "default_breadth": 3, "default_depth": 3, "skeleton_ratio": 0.15, "strategies": [], } @when( 'I run context show rich on "{project}" with execution environment' " and integer depth gradient" ) def step_ctx_show_rich_ee_int_gradient(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import context_show acms = dict(_COV_ACMS_BASE) acms["depth_gradient"] = {"1": 0.8, "2": 0.5} with ( patch( "cleveragents.cli.commands.project_context._load_policy_json", return_value={ "execution_environment": "host", "execution_env_priority": "high", }, ), patch( "cleveragents.cli.commands.project_context._read_acms_config", return_value=acms, ), ): _run_with_container( context, context_show, project=project, view=None, output_format="rich" ) @when('I run context show rich on "{project}" with non-integer depth gradient keys') def step_ctx_show_rich_str_gradient(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import context_show acms = dict(_COV_ACMS_BASE) acms["depth_gradient"] = {"hop_a": 0.8, "hop_b": 0.5} with ( patch( "cleveragents.cli.commands.project_context._load_policy_json", return_value=None, ), patch( "cleveragents.cli.commands.project_context._read_acms_config", return_value=acms, ), ): _run_with_container( context, context_show, project=project, view=None, output_format="rich" ) @then('the context show output contains "{text}"') def step_ctx_show_output_contains(context: Any, text: str) -> None: assert text in context.ctx_output, ( f"Expected {text!r} in context show output.\nGot:\n{context.ctx_output}" )