"""Step definitions for project_context_cli_coverage_boost.feature. Targets uncovered lines in ``src/cleveragents/cli/commands/project_context.py``. """ from __future__ import annotations import json as _json from io import StringIO from typing import Any from unittest.mock import MagicMock, patch from behave import given, then, when # type: ignore[import-untyped] from sqlalchemy import create_engine, text from sqlalchemy.orm import Session, sessionmaker # ------------------------------------------------------------------ # Session wrapper - prevents the production ``close()`` from # destroying our shared in-memory session. # ------------------------------------------------------------------ class _SafeSession: """Thin wrapper so ``close()`` is a no-op.""" def __init__(self, real: Session) -> None: object.__setattr__(self, "_real", real) 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_session_factory(context: Any) -> tuple[Any, Any]: 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 = sessionmaker( bind=engine, expire_on_commit=False, autoflush=True, autocommit=False )() wrapper = _SafeSession(real) return engine, lambda: wrapper # ------------------------------------------------------------------ # Mock container # ------------------------------------------------------------------ def _mock_container(context: Any) -> MagicMock: from cleveragents.application.services.context_tiers import ContextTierService mc = MagicMock() mc.namespaced_project_repo.return_value = context.cb_project_repo mc.session_factory.return_value = context.cb_session_factory if not hasattr(context, "cb_tier_service"): context.cb_tier_service = ContextTierService() mc.context_tier_service.return_value = context.cb_tier_service return mc def _run(context: Any, func: Any, *args: Any, **kwargs: Any) -> None: import typer from rich.console import Console as RichConsole mc = _mock_container(context) buf = StringIO() test_console = RichConsole( file=buf, no_color=True, highlight=False, force_terminal=False, width=500, soft_wrap=True, ) with ( patch( "cleveragents.application.container.get_container", return_value=mc, ), patch( "cleveragents.cli.commands.project_context.console", test_console, ), patch( "cleveragents.cli.commands.project_context.err_console", test_console, ), ): try: func(*args, **kwargs) context.cb_exit_code = 0 except typer.Exit as exc: context.cb_exit_code = exc.exit_code except typer.Abort: context.cb_exit_code = 1 context.cb_output = buf.getvalue() # ------------------------------------------------------------------ # Background # ------------------------------------------------------------------ @given("a coverage-boost in-memory database is initialized") def step_cb_init_db(context: Any) -> None: from cleveragents.infrastructure.database.repositories import ( NamespacedProjectRepository, ) engine, sf = _make_session_factory(context) context.cb_engine = engine context.cb_session_factory = sf context.cb_project_repo = NamespacedProjectRepository(session_factory=sf) context.cb_output = "" context.cb_exit_code = 0 @given('a project "{name}" exists for coverage boost') def step_cb_create_project(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.cb_project_repo.create(proj) # ------------------------------------------------------------------ # Helper: direct DB operations for seeding # ------------------------------------------------------------------ def _seed_policy_blob(context: Any, ns: str, blob: dict) -> None: """Write a raw JSON blob into ns_projects.context_policy_json.""" session = context.cb_session_factory() session.execute( text( "UPDATE ns_projects SET context_policy_json = :blob " "WHERE namespaced_name = :ns" ), {"blob": _json.dumps(blob), "ns": ns}, ) session.flush() def _load_raw_blob(context: Any, ns: str) -> dict | None: session = context.cb_session_factory() row = session.execute( text("SELECT context_policy_json FROM ns_projects WHERE namespaced_name = :ns"), {"ns": ns}, ).fetchone() if row is None or row[0] is None: return None return _json.loads(row[0]) # ------------------------------------------------------------------ # Given steps # ------------------------------------------------------------------ @given('I seed ACMS config for "{project}" with hot_max_tokens {val:d}') def step_seed_acms_hot(context: Any, project: str, val: int) -> None: from cleveragents.cli.commands.project_context import _default_acms_config blob = {"acms_config": {**_default_acms_config(), "hot_max_tokens": val}} _seed_policy_blob(context, project, blob) @given('I seed ACMS config for "{project}" with temporal_scope "{scope}"') def step_seed_acms_temporal(context: Any, project: str, scope: str) -> None: from cleveragents.cli.commands.project_context import _default_acms_config blob = {"acms_config": {**_default_acms_config(), "temporal_scope": scope}} _seed_policy_blob(context, project, blob) @given('the tier service has fragments for "{project}" with strategy metadata') def step_seed_tier_fragments(context: Any, project: str) -> None: from cleveragents.application.services.context_tiers import ContextTierService from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment svc = ContextTierService() svc.store( TieredFragment( fragment_id="frag-alpha", content="Alpha content for testing tier retrieval", tier=ContextTier.HOT, resource_id="res:file-alpha", project_name=project, token_count=30, access_count=5, metadata={"strategy": "tier_a"}, ) ) svc.store( TieredFragment( fragment_id="frag-beta", content="Beta content with different strategy", tier=ContextTier.WARM, resource_id="res:file-beta", project_name=project, token_count=20, access_count=0, metadata={"strategy": "tier_b"}, ) ) context.cb_tier_service = svc context.cb_stored_fragment_count = 2 @given('the tier service has large fragments for "{project}" exceeding budget') def step_seed_large_fragments(context: Any, project: str) -> None: from cleveragents.application.services.context_tiers import ContextTierService from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment svc = ContextTierService() # First fragment: 60 tokens, long content, access_count > 0 svc.store( TieredFragment( fragment_id="big-1", content="A" * 200, # long content -> triggers truncation (line 339) tier=ContextTier.HOT, resource_id="res:big1", project_name=project, token_count=60, access_count=12, # > 0 triggers relevance calc (line 344) ) ) # Second fragment: 60 tokens -> total would be 120 > budget of 100 svc.store( TieredFragment( fragment_id="big-2", content="Short", # short content -> no truncation tier=ContextTier.HOT, resource_id="res:big2", project_name=project, token_count=60, access_count=0, ) ) context.cb_tier_service = svc context.cb_stored_fragment_count = 2 # ------------------------------------------------------------------ # When steps # ------------------------------------------------------------------ @when('I read the stored policy for "{project}"') def step_read_policy(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import ( _read_acms_config, _read_policy, ) context.cb_policy = _read_policy(context.cb_session_factory, project) context.cb_acms = _read_acms_config(context.cb_session_factory, project) @when('I write a policy for "{project}" without explicit ACMS config') def step_write_policy_no_acms(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import _write_policy from cleveragents.domain.models.core.context_policy import ProjectContextPolicy policy = ProjectContextPolicy() # acms_config=None triggers the preservation branch (lines 199-201) _write_policy(context.cb_session_factory, project, policy, acms_config=None) @when( 'I run coverage-boost context set on "{project}" with view "{view}" and include-resource "{res}"' ) def step_cb_set_include(context: Any, project: str, view: str, res: str) -> None: from cleveragents.cli.commands.project_context import context_set _run(context, context_set, project=project, view=view, include_resource=[res]) @when( 'I run coverage-boost context set on "{project}" with view "{view}" and clear flag' ) def step_cb_set_clear(context: Any, project: str, view: str) -> None: from cleveragents.cli.commands.project_context import context_set _run(context, context_set, project=project, view=view, clear=True) @when('I run coverage-boost context set on "{project}" with default_depth "{depth}"') def step_cb_set_depth_string(context: Any, project: str, depth: str) -> None: from cleveragents.cli.commands.project_context import context_set _run(context, context_set, project=project, view="default", default_depth=depth) @when( 'I run coverage-boost context set on "{project}" with execution_environment "{env}"' ) def step_cb_set_exec_env(context: Any, project: str, env: str) -> None: from cleveragents.cli.commands.project_context import context_set _run( context, context_set, project=project, view="default", execution_environment=env, ) @when( 'I run coverage-boost context set on "{project}" with view "{view}" include-resource "{res}" and format "{fmt}"' ) def step_cb_set_json_fmt( context: Any, project: str, view: str, res: str, fmt: str ) -> None: from cleveragents.cli.commands.project_context import context_set _run( context, context_set, project=project, view=view, include_resource=[res], output_format=fmt, ) @when('I run coverage-boost context set on "{project}" with temporal_scope "{scope}"') def step_cb_set_temporal(context: Any, project: str, scope: str) -> None: from cleveragents.cli.commands.project_context import context_set _run(context, context_set, project=project, view="default", temporal_scope=scope) @when('I run coverage-boost context set on "{project}" with all ACMS overrides') def step_cb_set_all_acms(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import context_set _run( context, context_set, project=project, view="default", hot_max_tokens=1234, warm_max_decisions=111, cold_max_decisions=222, query_limit=50, summarize=False, summary_max_tokens=999, strategy=["bfs", "dfs"], default_breadth=7, default_depth="4", skeleton_ratio=0.5, temporal_scope="recent", auto_refresh=False, ) @when('I run coverage-boost context show on "{project}" with view "{view}"') def step_cb_show_view(context: Any, project: str, view: str) -> None: from cleveragents.cli.commands.project_context import context_show _run(context, context_show, project=project, view=view) @when( 'I run coverage-boost context inspect on "{project}" with strategy filter "{strat}"' ) def step_cb_inspect_strategy(context: Any, project: str, strat: str) -> None: from cleveragents.cli.commands.project_context import context_inspect _run(context, context_inspect, project=project, strategy_filter=strat) @when('I run coverage-boost context inspect on "{project}" with focus "{uri}"') def step_cb_inspect_focus(context: Any, project: str, uri: str) -> None: from cleveragents.cli.commands.project_context import context_inspect _run(context, context_inspect, project=project, focus=[uri]) @when( 'I run coverage-boost context inspect on "{project}" with breadth {b:d} and depth "{d}"' ) def step_cb_inspect_breadth_depth(context: Any, project: str, b: int, d: str) -> None: from cleveragents.cli.commands.project_context import context_inspect _run(context, context_inspect, project=project, breadth=b, depth=d) @when('I run coverage-boost context inspect on "{project}" with view "{view}"') def step_cb_inspect_view(context: Any, project: str, view: str) -> None: from cleveragents.cli.commands.project_context import context_inspect _run(context, context_inspect, project=project, view=view) @when('I run coverage-boost context simulate on "{project}" with defaults') def step_cb_simulate(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import context_simulate _run(context, context_simulate, project=project) @when('I run coverage-boost context simulate on "{project}" with budget {b:d}') def step_cb_simulate_budget(context: Any, project: str, b: int) -> None: from cleveragents.cli.commands.project_context import context_simulate _run(context, context_simulate, project=project, budget=b) @when('I run coverage-boost context simulate on "{project}" in rich format') def step_cb_simulate_rich(context: Any, project: str) -> None: from cleveragents.cli.commands.project_context import context_simulate _run(context, context_simulate, project=project, output_format="rich") @when('I run coverage-boost context simulate on "{project}" with view "{view}"') def step_cb_simulate_view(context: Any, project: str, view: str) -> None: from cleveragents.cli.commands.project_context import context_simulate _run(context, context_simulate, project=project, view=view) @when('I run coverage-boost context simulate on "{project}" with focus "{uri}"') def step_cb_simulate_focus(context: Any, project: str, uri: str) -> None: from cleveragents.cli.commands.project_context import context_simulate _run(context, context_simulate, project=project, focus=[uri]) @when( 'I run coverage-boost context simulate on "{project}" with strategy hints "{hints}"' ) def step_cb_simulate_strategies(context: Any, project: str, hints: str) -> None: from cleveragents.cli.commands.project_context import context_simulate _run( context, context_simulate, project=project, strategy_hint=hints.split(","), ) # ------------------------------------------------------------------ # Then steps # ------------------------------------------------------------------ @then("the policy should be a fresh default policy") def step_assert_default_policy(context: Any) -> None: from cleveragents.domain.models.core.context_policy import ProjectContextPolicy default = ProjectContextPolicy() assert context.cb_policy.default_view == default.default_view assert context.cb_policy.strategize_view is None assert context.cb_policy.execute_view is None assert context.cb_policy.apply_view is None @then("the ACMS config should equal the defaults") def step_assert_default_acms(context: Any) -> None: from cleveragents.cli.commands.project_context import _default_acms_config expected = _default_acms_config() assert context.cb_acms == expected, f"{context.cb_acms} != {expected}" @then("the stored project-context ACMS config hot_max_tokens should be {val}") def step_assert_acms_hot(context: Any, val: str) -> None: blob = _load_raw_blob(context, "local/cov-app") assert blob is not None, "No policy blob stored" assert blob.get("acms_config", {}).get("hot_max_tokens") == int(val) @then("the coverage-boost command should succeed") def step_cb_ok(context: Any) -> None: assert context.cb_exit_code == 0, ( f"Expected exit 0, got {context.cb_exit_code}. Output: {context.cb_output}" ) @then("the coverage-boost command should have exit code {code:d}") def step_cb_exit(context: Any, code: int) -> None: assert context.cb_exit_code == code, ( f"Expected exit {code}, got {context.cb_exit_code}. Output: {context.cb_output}" ) @then("the stored coverage-boost default view include_resources should be empty") def step_cb_default_empty(context: Any) -> None: from cleveragents.cli.commands.project_context import _read_policy policy = _read_policy(context.cb_session_factory, "local/cov-app") assert policy.default_view.include_resources == [], ( f"Expected empty, got {policy.default_view.include_resources}" ) @then('the stored ACMS config default_depth should be "{val}"') def step_assert_acms_depth_str(context: Any, val: str) -> None: blob = _load_raw_blob(context, "local/cov-app") assert blob is not None assert blob.get("acms_config", {}).get("default_depth") == val @then('the stored execution_environment should be "{env}"') def step_assert_exec_env(context: Any, env: str) -> None: blob = _load_raw_blob(context, "local/cov-app") assert blob is not None, "No policy blob stored" assert blob.get("execution_environment") == env, ( f"Expected '{env}', got {blob.get('execution_environment')}" ) @then('the coverage-boost output should contain "{text}"') def step_cb_output_contains(context: Any, text: str) -> None: assert text in context.cb_output, f"'{text}' not in output: {context.cb_output!r}" @then("the coverage-boost output should be valid JSON") def step_cb_output_json(context: Any) -> None: try: _json.loads(context.cb_output.strip()) except _json.JSONDecodeError as exc: raise AssertionError( f"Output is not valid JSON: {exc}\nOutput: {context.cb_output!r}" ) from exc @then("the simulation should have fewer fragments than stored") def step_cb_fewer_fragments(context: Any) -> None: # The output is rich format; we already checked success. # Just verify that the budget constrained the output # (budget=100, first frag=60 tokens, second would overflow). # We additionally verify via the internal function. from cleveragents.cli.commands.project_context import ( _default_acms_config, _simulate_context_assembly, ) assembled = _simulate_context_assembly( project_name="local/cov-app", acms_config=_default_acms_config(), budget_tokens=100, ) assert len(assembled.fragments) < context.cb_stored_fragment_count, ( f"Expected fewer than {context.cb_stored_fragment_count}, " f"got {len(assembled.fragments)}" ) @then("the stored ACMS config should reflect all overrides") def step_assert_all_acms(context: Any) -> None: blob = _load_raw_blob(context, "local/cov-app") assert blob is not None acms = blob.get("acms_config", {}) assert acms["hot_max_tokens"] == 1234 assert acms["warm_max_decisions"] == 111 assert acms["cold_max_decisions"] == 222 assert acms["query_limit"] == 50 assert acms["summarize"] is False assert acms["summary_max_tokens"] == 999 assert acms["strategies"] == ["bfs", "dfs"] assert acms["default_breadth"] == 7 assert acms["default_depth"] == 4 # "4" is parseable as int assert acms["skeleton_ratio"] == 0.5 assert acms["temporal_scope"] == "recent" assert acms["auto_refresh"] is False