forked from HAL9000/cleveragents-core
brent.edwards
f2be43dcc2
Centralize CLI output through shared renderers module with get_console() and get_err_console() accessors. Introduce ColumnSpec-based table rendering for plan list, resource list, and actor list commands. Add _FORMAT_HELP constant and fmt= parameter threading for consistent format handling. Includes Behave BDD tests for format output validation, Robot Framework integration tests for CLI formatting consistency, and CHANGELOG entry. ISSUES CLOSED: #210
811 lines
29 KiB
Python
811 lines
29 KiB
Python
"""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 that prevents ``close()`` from destroying the shared session.
|
|
|
|
Instead of a pure no-op, ``close()`` issues a ``rollback()`` on the
|
|
underlying session to reset any dirty/rolled-back state accumulated
|
|
during the previous call. This prevents phantom failures when the
|
|
production error path (e.g. ``_save_policy_json`` rollback) leaves
|
|
the session in an invalidated state.
|
|
"""
|
|
|
|
def __init__(self, real: Session) -> None:
|
|
object.__setattr__(self, "_real", real)
|
|
|
|
def close(self) -> None:
|
|
"""Reset session state without closing the underlying connection."""
|
|
real: Session = object.__getattribute__(self, "_real")
|
|
real.rollback()
|
|
|
|
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,
|
|
)
|
|
|
|
def _test_format_output(data, format_type):
|
|
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)
|
|
finally:
|
|
_sys.stdout = _old
|
|
return _r or _b.getvalue().rstrip("\n")
|
|
|
|
with (
|
|
patch(
|
|
"cleveragents.application.container.get_container",
|
|
return_value=mc,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.renderers._console",
|
|
test_console,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.renderers._err_console",
|
|
test_console,
|
|
),
|
|
patch(
|
|
"cleveragents.cli.commands.project_context.format_output",
|
|
_test_format_output,
|
|
),
|
|
):
|
|
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.commit()
|
|
|
|
|
|
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)}"
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# flush() → commit() regression test: real separate sessions
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I save a policy for "{project}" via a real session factory and read it back via a separate session'
|
|
)
|
|
def step_save_and_read_real_sessions(context: Any, project: str) -> None:
|
|
"""Exercise _save_policy_json + _load_policy_json via real sessions.
|
|
|
|
Uses a file-based SQLite database so that separate sessions truly
|
|
see each other's committed data. The _SafeSession wrapper is NOT
|
|
used here — if production code uses flush() instead of commit(),
|
|
the second session will not see the written data.
|
|
"""
|
|
import os
|
|
import tempfile
|
|
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker as real_sessionmaker
|
|
|
|
from cleveragents.cli.commands.project_context import (
|
|
_load_policy_json,
|
|
_save_policy_json,
|
|
)
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
# Create a file-based SQLite DB so separate sessions are independent.
|
|
fd, db_path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
db_url = f"sqlite:///{db_path}"
|
|
|
|
# M4: Use separate engines so that flush() vs commit() visibility
|
|
# is truly tested — a shared engine/pool could serve the same
|
|
# underlying connection, masking the difference.
|
|
engine_seed = create_engine(db_url, echo=False)
|
|
engine_write = create_engine(db_url, echo=False)
|
|
engine_read = create_engine(db_url, echo=False)
|
|
try:
|
|
Base.metadata.create_all(engine_seed)
|
|
factory_write = real_sessionmaker(bind=engine_write, expire_on_commit=False)
|
|
factory_read = real_sessionmaker(bind=engine_read, expire_on_commit=False)
|
|
|
|
# Seed the project row so UPDATE has a target.
|
|
from datetime import UTC, datetime
|
|
|
|
now = datetime.now(tz=UTC).isoformat()
|
|
session = real_sessionmaker(bind=engine_seed, expire_on_commit=False)()
|
|
try:
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO ns_projects (namespaced_name, namespace, tags_json, created_at, updated_at) "
|
|
"VALUES (:ns, :nsp, :tags, :ca, :ua)"
|
|
),
|
|
{"ns": project, "nsp": "local", "tags": "[]", "ca": now, "ua": now},
|
|
)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
# Write policy via production code (uses commit) with write-only factory.
|
|
policy_dict = {
|
|
"regression_marker": "commit_test",
|
|
"acms_config": {"hot_max_tokens": 42},
|
|
}
|
|
_save_policy_json(factory_write, project, policy_dict)
|
|
|
|
# Read back via a SEPARATE engine/factory to prove commit visibility.
|
|
loaded = _load_policy_json(factory_read, project)
|
|
|
|
context.cb_regression_loaded = loaded
|
|
finally:
|
|
engine_seed.dispose()
|
|
engine_write.dispose()
|
|
engine_read.dispose()
|
|
# Clean up the DB file and any WAL/SHM journal files left by
|
|
# SQLite WAL mode to avoid leaking temp files.
|
|
for suffix in ("", "-wal", "-shm"):
|
|
path = db_path + suffix
|
|
if os.path.exists(path):
|
|
os.unlink(path)
|
|
|
|
|
|
@then("the policy read from the separate session should contain the saved data")
|
|
def step_assert_regression_data(context: Any) -> None:
|
|
loaded = context.cb_regression_loaded
|
|
assert loaded is not None, "Policy was None — data did not persist across sessions"
|
|
assert loaded.get("regression_marker") == "commit_test", (
|
|
f"Expected 'commit_test', got {loaded.get('regression_marker')}"
|
|
)
|
|
assert loaded.get("acms_config", {}).get("hot_max_tokens") == 42, (
|
|
f"Expected hot_max_tokens=42, got {loaded.get('acms_config', {}).get('hot_max_tokens')}"
|
|
)
|
|
|
|
|
|
@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
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Rollback path test for _save_policy_json (P3-14)
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when("I attempt to save a policy with a commit that will fail")
|
|
def step_save_policy_commit_fail(context: Any) -> None:
|
|
"""Force a commit failure and verify the rollback path.
|
|
|
|
Uses the existing ``context.cb_session_factory`` (set up by the
|
|
Background step which already seeded the ``local/cov-app`` project
|
|
row) to ensure the UPDATE actually modifies data. Monkey-patches
|
|
``commit()`` on the underlying real session to simulate a commit
|
|
failure.
|
|
"""
|
|
from cleveragents.cli.commands.project_context import _save_policy_json
|
|
|
|
factory = context.cb_session_factory
|
|
|
|
# Get the _SafeSession wrapper and the real session underneath.
|
|
safe_session = factory()
|
|
real: Session = object.__getattribute__(safe_session, "_real")
|
|
original_commit = real.commit
|
|
|
|
def failing_commit() -> None:
|
|
raise RuntimeError("Simulated commit failure")
|
|
|
|
real.commit = failing_commit # type: ignore[assignment]
|
|
|
|
context.cb_rollback_exception = None
|
|
try:
|
|
_save_policy_json(factory, "local/cov-app", {"marker": "rollback_test"})
|
|
except RuntimeError as exc:
|
|
context.cb_rollback_exception = exc
|
|
finally:
|
|
# Restore original commit so subsequent scenarios can use it.
|
|
real.commit = original_commit # type: ignore[assignment]
|
|
|
|
# Store the session for dirty-state check.
|
|
context.cb_rollback_session = real
|
|
|
|
|
|
@then("the original commit exception should be re-raised")
|
|
def step_assert_commit_exception_reraised(context: Any) -> None:
|
|
exc = context.cb_rollback_exception
|
|
assert exc is not None, "_save_policy_json did not re-raise the commit exception"
|
|
assert "Simulated commit failure" in str(exc), (
|
|
f"Expected 'Simulated commit failure', got: {exc}"
|
|
)
|
|
|
|
|
|
@then("the session should not be in a dirty state after the failure")
|
|
def step_assert_session_clean_after_rollback(context: Any) -> None:
|
|
session: Session = context.cb_rollback_session
|
|
# After rollback, the session should have no pending changes.
|
|
assert not session.dirty, "Session has dirty objects after rollback"
|
|
assert not session.new, "Session has new objects after rollback"
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Nonexistent project test for _save_policy_json (P3-15)
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
@when('I save a policy for a nonexistent project "{project}"')
|
|
def step_save_policy_nonexistent(context: Any, project: str) -> None:
|
|
"""Call _save_policy_json for a project that has no ns_projects row."""
|
|
from cleveragents.cli.commands.project_context import _save_policy_json
|
|
|
|
context.cb_nonexistent_exception = None
|
|
try:
|
|
_save_policy_json(
|
|
context.cb_session_factory,
|
|
project,
|
|
{"marker": "nonexistent_test"},
|
|
)
|
|
except Exception as exc:
|
|
context.cb_nonexistent_exception = exc
|
|
|
|
|
|
@then("the save should complete without error")
|
|
def step_assert_save_no_error(context: Any) -> None:
|
|
exc = context.cb_nonexistent_exception
|
|
# _save_policy_json does not currently check rowcount, so it
|
|
# completes without error even when 0 rows are affected.
|
|
assert exc is None, f"_save_policy_json raised an unexpected error: {exc}"
|
|
|
|
|
|
@then('the policy should not be retrievable for "{project}"')
|
|
def step_assert_policy_not_retrievable(context: Any, project: str) -> None:
|
|
from cleveragents.cli.commands.project_context import _load_policy_json
|
|
|
|
loaded = _load_policy_json(context.cb_session_factory, project)
|
|
assert loaded is None, f"Expected None for nonexistent project, got: {loaded}"
|