forked from HAL9000/cleveragents-core
freemo
0ca1303927
Introduce a lightweight checkpoint/rollback system for sandbox state during plan execute and apply flows. CheckpointManager snapshots the sandbox working directory before each phase and can restore it on failure, giving the execution engine a reliable undo mechanism. Key changes: - SandboxCheckpoint model, Checkpointable protocol, and CheckpointManager in infrastructure/sandbox/checkpoint.py - PlanExecutor gains optional checkpoint_manager with pre/post execute hooks and automatic rollback on failure - PlanApplyService gains optional checkpoint_manager with pre-apply checkpoint and rollback helper - 12 BDD scenarios (features/sandbox_checkpoints.feature) - 5 Robot Framework smoke tests (robot/sandbox_checkpoint_smoke.robot) - ASV benchmarks for creation, rollback, and listing operations - Reference documentation in docs/reference/sandbox.md ISSUES CLOSED: #183
1073 lines
40 KiB
Python
1073 lines
40 KiB
Python
"""Step definitions for changeset repository coverage feature tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Callable
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.exc import OperationalError
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from cleveragents.core.exceptions import DatabaseError
|
|
from cleveragents.domain.models.core.change import (
|
|
ChangeEntry,
|
|
ChangeOperation,
|
|
ToolInvocation,
|
|
)
|
|
from cleveragents.infrastructure.database.changeset_repository import (
|
|
ChangeSetEntryRepository,
|
|
SqliteChangeSetStore,
|
|
ToolInvocationRepository,
|
|
)
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
ChangeSetEntryModel,
|
|
ToolInvocationModel,
|
|
)
|
|
|
|
# ------ Helpers ------
|
|
|
|
|
|
def _make_session_factory() -> tuple[Callable[[], Session], Session, Engine]:
|
|
"""Create an in-memory SQLite engine + shared session factory with schema.
|
|
|
|
Uses a single shared session so that flush() data is visible across
|
|
all repository calls within the same scenario. Without this,
|
|
short-lived sessions returned by sessionmaker() may be garbage-
|
|
collected, causing the StaticPool's reset_on_return to discard
|
|
uncommitted INSERTs.
|
|
"""
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
echo=False,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
session: Session = sessionmaker(bind=engine)()
|
|
return lambda: session, session, engine
|
|
|
|
|
|
def _make_change_entry(
|
|
plan_id: str = "plan-cov-1",
|
|
operation: ChangeOperation = ChangeOperation.CREATE,
|
|
path: str = "src/cov.py",
|
|
) -> ChangeEntry:
|
|
"""Create a ChangeEntry for testing."""
|
|
before_hash = "aaa111" if operation != ChangeOperation.CREATE else None
|
|
after_hash = "bbb222" if operation != ChangeOperation.DELETE else None
|
|
return ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="res-cov-001",
|
|
tool_name="file-write",
|
|
operation=operation,
|
|
path=path,
|
|
before_hash=before_hash,
|
|
after_hash=after_hash,
|
|
)
|
|
|
|
|
|
class _BrokenSession:
|
|
"""Stub session that raises OperationalError on mutating ops."""
|
|
|
|
def add(self, _obj: Any) -> None:
|
|
"""Raise OperationalError."""
|
|
raise OperationalError("simulated", {}, None)
|
|
|
|
def flush(self) -> None:
|
|
"""Raise OperationalError."""
|
|
raise OperationalError("simulated", {}, None)
|
|
|
|
def rollback(self) -> None:
|
|
"""No-op rollback."""
|
|
|
|
|
|
class _BrokenQuerySession:
|
|
"""Stub session that raises OperationalError on query."""
|
|
|
|
def query(self, _model: Any) -> _BrokenQuerySession:
|
|
"""Return self to chain."""
|
|
raise OperationalError("simulated", {}, None)
|
|
|
|
|
|
class _BrokenDeleteSession:
|
|
"""Stub session that raises OperationalError on query-based delete."""
|
|
|
|
class _FakeQuery:
|
|
"""Fake query that raises on delete."""
|
|
|
|
def filter_by(self, **_kwargs: Any) -> _BrokenDeleteSession._FakeQuery:
|
|
"""Return self."""
|
|
return self
|
|
|
|
def delete(self) -> None:
|
|
"""Raise OperationalError."""
|
|
raise OperationalError("simulated", {}, None)
|
|
|
|
def query(self, _model: Any) -> _BrokenDeleteSession._FakeQuery:
|
|
"""Return a fake query."""
|
|
return self._FakeQuery()
|
|
|
|
def rollback(self) -> None:
|
|
"""No-op rollback."""
|
|
|
|
|
|
# ------ ChangeSetEntryRepository validation steps ------
|
|
|
|
|
|
@given("I have a coverage ChangeSetEntryRepository")
|
|
def step_create_coverage_entry_repo(context: Context) -> None:
|
|
"""Create a ChangeSetEntryRepository with real in-memory session."""
|
|
factory, session, engine = _make_session_factory()
|
|
context.cov_engine = engine
|
|
context.cov_entry_repo = ChangeSetEntryRepository(factory)
|
|
context._cleanup_handlers.append(lambda: session.close())
|
|
context._cleanup_handlers.append(lambda: engine.dispose())
|
|
|
|
|
|
@when("I try to coverage-save an entry with empty changeset_id")
|
|
def step_save_entry_empty_changeset_id(context: Context) -> None:
|
|
"""Attempt to save entry with empty changeset_id."""
|
|
entry = _make_change_entry()
|
|
try:
|
|
context.cov_entry_repo.save_entry("", entry)
|
|
context.cov_raised = None
|
|
except (ValueError, TypeError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I try to coverage-save a non-ChangeEntry object")
|
|
def step_save_entry_non_change_entry(context: Context) -> None:
|
|
"""Attempt to save a non-ChangeEntry object."""
|
|
try:
|
|
context.cov_entry_repo.save_entry("cs-valid", "not-an-entry") # type: ignore[arg-type]
|
|
context.cov_raised = None
|
|
except (ValueError, TypeError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I try to coverage-get entries for empty changeset_id")
|
|
def step_get_entries_empty_changeset_id(context: Context) -> None:
|
|
"""Attempt to get entries with empty changeset_id."""
|
|
try:
|
|
context.cov_entry_repo.get_entries_for_changeset("")
|
|
context.cov_raised = None
|
|
except (ValueError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I try to coverage-get entries for empty plan_id")
|
|
def step_get_entries_empty_plan_id(context: Context) -> None:
|
|
"""Attempt to get entries with empty plan_id."""
|
|
try:
|
|
context.cov_entry_repo.get_entries_for_plan("")
|
|
context.cov_raised = None
|
|
except (ValueError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I try to coverage-delete entries for empty changeset_id")
|
|
def step_delete_entries_empty_changeset_id(context: Context) -> None:
|
|
"""Attempt to delete entries with empty changeset_id."""
|
|
try:
|
|
context.cov_entry_repo.delete_for_changeset("")
|
|
context.cov_raised = None
|
|
except (ValueError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I try to coverage-delete entries for empty plan_id")
|
|
def step_delete_entries_empty_plan_id(context: Context) -> None:
|
|
"""Attempt to delete entries for empty plan_id."""
|
|
try:
|
|
context.cov_entry_repo.delete_for_plan("")
|
|
context.cov_raised = None
|
|
except (ValueError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
# ------ ChangeSetEntryRepository database error steps ------
|
|
|
|
|
|
@given("I have a coverage ChangeSetEntryRepository with a broken session")
|
|
def step_create_broken_entry_repo(context: Context) -> None:
|
|
"""Create a ChangeSetEntryRepository with a broken session factory."""
|
|
context.cov_entry_repo = ChangeSetEntryRepository(lambda: _BrokenSession()) # type: ignore[return-value]
|
|
|
|
|
|
@when("I try to coverage-save an entry that triggers an OperationalError")
|
|
def step_save_entry_operational_error(context: Context) -> None:
|
|
"""Attempt to save an entry that triggers OperationalError."""
|
|
entry = _make_change_entry()
|
|
try:
|
|
context.cov_entry_repo.save_entry("cs-broken", entry)
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@given("I have a coverage ChangeSetEntryRepository with a broken query session")
|
|
def step_create_broken_query_entry_repo(context: Context) -> None:
|
|
"""Create a ChangeSetEntryRepository with broken query session."""
|
|
context.cov_entry_repo = ChangeSetEntryRepository(lambda: _BrokenQuerySession()) # type: ignore[return-value]
|
|
|
|
|
|
@when('I try to coverage-get entries for changeset "{changeset_id}"')
|
|
def step_get_entries_broken_changeset(context: Context, changeset_id: str) -> None:
|
|
"""Attempt to get entries with broken query session."""
|
|
try:
|
|
context.cov_entry_repo.get_entries_for_changeset(changeset_id)
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when('I try to coverage-get entries for plan "{plan_id}"')
|
|
def step_get_entries_broken_plan(context: Context, plan_id: str) -> None:
|
|
"""Attempt to get entries for plan with broken session."""
|
|
try:
|
|
context.cov_entry_repo.get_entries_for_plan(plan_id)
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@given("I have a coverage ChangeSetEntryRepository with a broken delete session")
|
|
def step_create_broken_delete_entry_repo(context: Context) -> None:
|
|
"""Create a ChangeSetEntryRepository with broken delete session."""
|
|
context.cov_entry_repo = ChangeSetEntryRepository(lambda: _BrokenDeleteSession()) # type: ignore[return-value]
|
|
|
|
|
|
@when('I try to coverage-delete entries for changeset "{changeset_id}"')
|
|
def step_delete_entries_broken_changeset(context: Context, changeset_id: str) -> None:
|
|
"""Attempt to delete entries for changeset with broken session."""
|
|
try:
|
|
context.cov_entry_repo.delete_for_changeset(changeset_id)
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when('I try to coverage-delete entries for plan "{plan_id}"')
|
|
def step_delete_entries_broken_plan(context: Context, plan_id: str) -> None:
|
|
"""Attempt to delete entries for plan with broken session."""
|
|
try:
|
|
context.cov_entry_repo.delete_for_plan(plan_id)
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
# ------ ChangeSetEntryRepository _to_domain steps ------
|
|
|
|
|
|
@given("I have a coverage ChangeSetEntryModel row with null hashes and modes")
|
|
def step_create_row_null_hashes(context: Context) -> None:
|
|
"""Create a ChangeSetEntryModel with null hashes and modes."""
|
|
context.cov_row = ChangeSetEntryModel(
|
|
entry_id="01ENTRY0000000000000000001",
|
|
changeset_id="01CSET00000000000000000001",
|
|
plan_id="01PLAN00000000000000000001",
|
|
resource_id="01RES000000000000000000001",
|
|
tool_name="file-write",
|
|
operation="modify",
|
|
path="src/test.py",
|
|
before_hash=None,
|
|
after_hash=None,
|
|
before_mode=None,
|
|
after_mode=None,
|
|
timestamp=datetime.now(UTC).isoformat(),
|
|
)
|
|
|
|
|
|
@given("I have a coverage ChangeSetEntryModel row with hashes and modes")
|
|
def step_create_row_with_hashes(context: Context) -> None:
|
|
"""Create a ChangeSetEntryModel with populated hashes and modes."""
|
|
context.cov_row = ChangeSetEntryModel(
|
|
entry_id="01ENTRY0000000000000000002",
|
|
changeset_id="01CSET00000000000000000002",
|
|
plan_id="01PLAN00000000000000000002",
|
|
resource_id="01RES000000000000000000002",
|
|
tool_name="file-write",
|
|
operation="modify",
|
|
path="src/test.py",
|
|
before_hash="abc123",
|
|
after_hash="def456",
|
|
before_mode=33188,
|
|
after_mode=33261,
|
|
timestamp=datetime.now(UTC).isoformat(),
|
|
)
|
|
|
|
|
|
@given("I have a coverage ChangeSetEntryModel row with empty string hashes")
|
|
def step_create_row_empty_string_hashes(context: Context) -> None:
|
|
"""Create a ChangeSetEntryModel with empty string hashes."""
|
|
context.cov_row = ChangeSetEntryModel(
|
|
entry_id="01ENTRY0000000000000000003",
|
|
changeset_id="01CSET00000000000000000003",
|
|
plan_id="01PLAN00000000000000000003",
|
|
resource_id="01RES000000000000000000003",
|
|
tool_name="file-write",
|
|
operation="modify",
|
|
path="src/test.py",
|
|
before_hash="",
|
|
after_hash="",
|
|
before_mode=None,
|
|
after_mode=None,
|
|
timestamp=datetime.now(UTC).isoformat(),
|
|
)
|
|
|
|
|
|
@when("I convert the coverage row to a domain ChangeEntry")
|
|
def step_convert_row_to_domain_entry(context: Context) -> None:
|
|
"""Convert the model row to a domain ChangeEntry."""
|
|
context.cov_domain_entry = ChangeSetEntryRepository._to_domain(context.cov_row)
|
|
|
|
|
|
@then("the domain entry before_hash should be None")
|
|
def step_assert_before_hash_none(context: Context) -> None:
|
|
"""Assert before_hash is None."""
|
|
assert context.cov_domain_entry.before_hash is None, (
|
|
f"Expected before_hash=None, got {context.cov_domain_entry.before_hash!r}"
|
|
)
|
|
|
|
|
|
@then("the domain entry after_hash should be None")
|
|
def step_assert_after_hash_none(context: Context) -> None:
|
|
"""Assert after_hash is None."""
|
|
assert context.cov_domain_entry.after_hash is None, (
|
|
f"Expected after_hash=None, got {context.cov_domain_entry.after_hash!r}"
|
|
)
|
|
|
|
|
|
@then("the domain entry before_mode should be None")
|
|
def step_assert_before_mode_none(context: Context) -> None:
|
|
"""Assert before_mode is None."""
|
|
assert context.cov_domain_entry.before_mode is None, (
|
|
f"Expected before_mode=None, got {context.cov_domain_entry.before_mode!r}"
|
|
)
|
|
|
|
|
|
@then("the domain entry after_mode should be None")
|
|
def step_assert_after_mode_none(context: Context) -> None:
|
|
"""Assert after_mode is None."""
|
|
assert context.cov_domain_entry.after_mode is None, (
|
|
f"Expected after_mode=None, got {context.cov_domain_entry.after_mode!r}"
|
|
)
|
|
|
|
|
|
@then('the domain entry before_hash should be "{expected}"')
|
|
def step_assert_before_hash_value(context: Context, expected: str) -> None:
|
|
"""Assert before_hash equals expected value."""
|
|
assert context.cov_domain_entry.before_hash == expected, (
|
|
f"Expected before_hash={expected!r}, got {context.cov_domain_entry.before_hash!r}"
|
|
)
|
|
|
|
|
|
@then('the domain entry after_hash should be "{expected}"')
|
|
def step_assert_after_hash_value(context: Context, expected: str) -> None:
|
|
"""Assert after_hash equals expected value."""
|
|
assert context.cov_domain_entry.after_hash == expected, (
|
|
f"Expected after_hash={expected!r}, got {context.cov_domain_entry.after_hash!r}"
|
|
)
|
|
|
|
|
|
@then("the domain entry before_mode should be {expected:d}")
|
|
def step_assert_before_mode_value(context: Context, expected: int) -> None:
|
|
"""Assert before_mode equals expected value."""
|
|
assert context.cov_domain_entry.before_mode == expected, (
|
|
f"Expected before_mode={expected}, got {context.cov_domain_entry.before_mode!r}"
|
|
)
|
|
|
|
|
|
@then("the domain entry after_mode should be {expected:d}")
|
|
def step_assert_after_mode_value(context: Context, expected: int) -> None:
|
|
"""Assert after_mode equals expected value."""
|
|
assert context.cov_domain_entry.after_mode == expected, (
|
|
f"Expected after_mode={expected}, got {context.cov_domain_entry.after_mode!r}"
|
|
)
|
|
|
|
|
|
# ------ ToolInvocationRepository validation steps ------
|
|
|
|
|
|
@given("I have a coverage ToolInvocationRepository")
|
|
def step_create_coverage_inv_repo(context: Context) -> None:
|
|
"""Create a ToolInvocationRepository with real in-memory session."""
|
|
factory, session, engine = _make_session_factory()
|
|
context.cov_engine = engine
|
|
context.cov_inv_repo = ToolInvocationRepository(factory)
|
|
context._cleanup_handlers.append(lambda: session.close())
|
|
context._cleanup_handlers.append(lambda: engine.dispose())
|
|
|
|
|
|
@when("I try to coverage-save a non-ToolInvocation object")
|
|
def step_save_non_tool_invocation(context: Context) -> None:
|
|
"""Attempt to save a non-ToolInvocation object."""
|
|
try:
|
|
context.cov_inv_repo.save_invocation("not-an-invocation") # type: ignore[arg-type]
|
|
context.cov_raised = None
|
|
except (ValueError, TypeError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I try to coverage-get invocations for empty plan_id")
|
|
def step_get_invocations_empty_plan_id(context: Context) -> None:
|
|
"""Attempt to get invocations with empty plan_id."""
|
|
try:
|
|
context.cov_inv_repo.get_invocations_for_plan("")
|
|
context.cov_raised = None
|
|
except (ValueError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I try to coverage-delete invocations for empty plan_id")
|
|
def step_delete_invocations_empty_plan_id(context: Context) -> None:
|
|
"""Attempt to delete invocations for empty plan_id."""
|
|
try:
|
|
context.cov_inv_repo.delete_for_plan("")
|
|
context.cov_raised = None
|
|
except (ValueError, DatabaseError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
# ------ ToolInvocationRepository database error steps ------
|
|
|
|
|
|
@given("I have a coverage ToolInvocationRepository with a broken session")
|
|
def step_create_broken_inv_repo(context: Context) -> None:
|
|
"""Create a ToolInvocationRepository with a broken session factory."""
|
|
context.cov_inv_repo = ToolInvocationRepository(lambda: _BrokenSession()) # type: ignore[return-value]
|
|
|
|
|
|
@when("I try to coverage-save an invocation that triggers an OperationalError")
|
|
def step_save_invocation_operational_error(context: Context) -> None:
|
|
"""Attempt to save invocation that triggers OperationalError."""
|
|
inv = ToolInvocation(
|
|
plan_id="plan-broken",
|
|
tool_name="file-write",
|
|
arguments={"path": "test.py"},
|
|
success=True,
|
|
)
|
|
try:
|
|
context.cov_inv_repo.save_invocation(inv, changeset_id="cs-broken")
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@given("I have a coverage ToolInvocationRepository with a broken query session")
|
|
def step_create_broken_query_inv_repo(context: Context) -> None:
|
|
"""Create a ToolInvocationRepository with broken query session."""
|
|
context.cov_inv_repo = ToolInvocationRepository(lambda: _BrokenQuerySession()) # type: ignore[return-value]
|
|
|
|
|
|
@when('I try to coverage-get invocations for plan "{plan_id}"')
|
|
def step_get_invocations_broken_plan(context: Context, plan_id: str) -> None:
|
|
"""Attempt to get invocations with broken query session."""
|
|
try:
|
|
context.cov_inv_repo.get_invocations_for_plan(plan_id)
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@given("I have a coverage ToolInvocationRepository with a broken delete session")
|
|
def step_create_broken_delete_inv_repo(context: Context) -> None:
|
|
"""Create a ToolInvocationRepository with broken delete session."""
|
|
context.cov_inv_repo = ToolInvocationRepository(lambda: _BrokenDeleteSession()) # type: ignore[return-value]
|
|
|
|
|
|
@when('I try to coverage-delete invocations for plan "{plan_id}"')
|
|
def step_delete_invocations_broken_plan(context: Context, plan_id: str) -> None:
|
|
"""Attempt to delete invocations with broken session."""
|
|
try:
|
|
context.cov_inv_repo.delete_for_plan(plan_id)
|
|
context.cov_raised = None
|
|
except (DatabaseError, OperationalError) as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
# ------ ToolInvocationRepository save with optional fields ------
|
|
|
|
|
|
@given("I have a coverage ToolInvocationRepository with real session")
|
|
def step_create_real_inv_repo(context: Context) -> None:
|
|
"""Create a ToolInvocationRepository with real in-memory session."""
|
|
factory, session, engine = _make_session_factory()
|
|
context.cov_engine = engine
|
|
context.cov_session_factory = factory
|
|
context.cov_inv_repo = ToolInvocationRepository(factory)
|
|
context._cleanup_handlers.append(lambda: session.close())
|
|
context._cleanup_handlers.append(lambda: engine.dispose())
|
|
|
|
|
|
@when("I coverage-save an invocation with result, completed_at, and provider_metadata")
|
|
def step_save_invocation_all_optional(context: Context) -> None:
|
|
"""Save invocation with all optional fields populated."""
|
|
context.cov_plan_id = "plan-all-opts"
|
|
inv = ToolInvocation(
|
|
plan_id=context.cov_plan_id,
|
|
tool_name="file-write",
|
|
skill_name="code-skill",
|
|
arguments={"path": "test.py", "content": "hello"},
|
|
result={"status": "ok", "bytes_written": 5},
|
|
error=None,
|
|
success=True,
|
|
duration_ms=42.5,
|
|
completed_at=datetime.now(UTC),
|
|
change_ids=["chg-1", "chg-2"],
|
|
sequence_number=3,
|
|
sandbox_path="/tmp/sandbox",
|
|
resource_refs=["res-1"],
|
|
provider_metadata={"model": "gpt-4", "latency_ms": 100},
|
|
)
|
|
context.cov_inv_repo.save_invocation(inv, changeset_id="cs-all-opts")
|
|
# Commit the session
|
|
session = context.cov_session_factory()
|
|
session.commit()
|
|
|
|
|
|
@then("the coverage-saved invocation should round-trip with all fields intact")
|
|
def step_assert_roundtrip_all_fields(context: Context) -> None:
|
|
"""Assert round-trip preserves all optional fields."""
|
|
invocations = context.cov_inv_repo.get_invocations_for_plan(context.cov_plan_id)
|
|
assert len(invocations) == 1, f"Expected 1 invocation, got {len(invocations)}"
|
|
inv = invocations[0]
|
|
assert inv.result is not None, "Expected result to not be None"
|
|
assert inv.result["status"] == "ok", (
|
|
f"Expected result status 'ok', got {inv.result}"
|
|
)
|
|
assert inv.completed_at is not None, "Expected completed_at to not be None"
|
|
assert inv.provider_metadata is not None, (
|
|
"Expected provider_metadata to not be None"
|
|
)
|
|
assert inv.provider_metadata["model"] == "gpt-4", (
|
|
f"Expected provider_metadata model 'gpt-4', got {inv.provider_metadata}"
|
|
)
|
|
assert inv.skill_name == "code-skill", (
|
|
f"Expected skill_name 'code-skill', got {inv.skill_name}"
|
|
)
|
|
assert inv.duration_ms == 42.5, f"Expected duration_ms 42.5, got {inv.duration_ms}"
|
|
assert inv.sequence_number == 3, (
|
|
f"Expected sequence_number 3, got {inv.sequence_number}"
|
|
)
|
|
assert inv.sandbox_path == "/tmp/sandbox", (
|
|
f"Expected sandbox_path, got {inv.sandbox_path}"
|
|
)
|
|
assert inv.change_ids == ["chg-1", "chg-2"], (
|
|
f"Expected change_ids, got {inv.change_ids}"
|
|
)
|
|
assert inv.resource_refs == ["res-1"], (
|
|
f"Expected resource_refs, got {inv.resource_refs}"
|
|
)
|
|
|
|
|
|
@when(
|
|
"I coverage-save an invocation with no result, no completed_at, and no provider_metadata"
|
|
)
|
|
def step_save_invocation_none_optional(context: Context) -> None:
|
|
"""Save invocation with all optional fields as None."""
|
|
context.cov_plan_id = "plan-no-opts"
|
|
inv = ToolInvocation(
|
|
plan_id=context.cov_plan_id,
|
|
tool_name="file-read",
|
|
arguments={},
|
|
result=None,
|
|
success=True,
|
|
completed_at=None,
|
|
provider_metadata=None,
|
|
)
|
|
context.cov_inv_repo.save_invocation(inv, changeset_id="cs-no-opts")
|
|
session = context.cov_session_factory()
|
|
session.commit()
|
|
|
|
|
|
@then("the coverage-saved invocation should round-trip with None optional fields")
|
|
def step_assert_roundtrip_none_fields(context: Context) -> None:
|
|
"""Assert round-trip preserves None optional fields."""
|
|
invocations = context.cov_inv_repo.get_invocations_for_plan(context.cov_plan_id)
|
|
assert len(invocations) == 1, f"Expected 1 invocation, got {len(invocations)}"
|
|
inv = invocations[0]
|
|
assert inv.result is None, f"Expected result=None, got {inv.result!r}"
|
|
assert inv.completed_at is None, (
|
|
f"Expected completed_at=None, got {inv.completed_at!r}"
|
|
)
|
|
assert inv.provider_metadata is None, (
|
|
f"Expected provider_metadata=None, got {inv.provider_metadata!r}"
|
|
)
|
|
|
|
|
|
@when("I coverage-save an invocation without changeset_id")
|
|
def step_save_invocation_no_changeset(context: Context) -> None:
|
|
"""Save invocation without changeset_id."""
|
|
context.cov_plan_id = "plan-no-cs"
|
|
inv = ToolInvocation(
|
|
plan_id=context.cov_plan_id,
|
|
tool_name="file-write",
|
|
arguments={"path": "test.py"},
|
|
success=True,
|
|
)
|
|
context.cov_inv_repo.save_invocation(inv, changeset_id=None)
|
|
session = context.cov_session_factory()
|
|
session.commit()
|
|
|
|
|
|
@then("the coverage-saved invocation for plan should be retrievable")
|
|
def step_assert_invocation_retrievable(context: Context) -> None:
|
|
"""Assert invocation is retrievable without changeset_id."""
|
|
invocations = context.cov_inv_repo.get_invocations_for_plan(context.cov_plan_id)
|
|
assert len(invocations) == 1, f"Expected 1 invocation, got {len(invocations)}"
|
|
|
|
|
|
# ------ ToolInvocationRepository _to_domain steps ------
|
|
|
|
|
|
@given("I have a coverage ToolInvocationModel row with null JSON fields")
|
|
def step_create_inv_row_null_json(context: Context) -> None:
|
|
"""Create a ToolInvocationModel with null JSON fields."""
|
|
context.cov_inv_row = ToolInvocationModel(
|
|
invocation_id="01INV000000000000000000001",
|
|
changeset_id=None,
|
|
plan_id="01PLAN00000000000000000001",
|
|
tool_name="file-write",
|
|
skill_name=None,
|
|
arguments_json=None,
|
|
result_json=None,
|
|
error=None,
|
|
success=True,
|
|
duration_ms=10.0,
|
|
started_at=datetime.now(UTC).isoformat(),
|
|
completed_at=None,
|
|
change_ids_json=None,
|
|
sequence_number=1,
|
|
sandbox_path=None,
|
|
resource_refs_json=None,
|
|
provider_metadata_json=None,
|
|
)
|
|
|
|
|
|
@given("I have a coverage ToolInvocationModel row with populated JSON fields")
|
|
def step_create_inv_row_populated_json(context: Context) -> None:
|
|
"""Create a ToolInvocationModel with populated JSON fields."""
|
|
context.cov_inv_row = ToolInvocationModel(
|
|
invocation_id="01INV000000000000000000002",
|
|
changeset_id="01CSET00000000000000000001",
|
|
plan_id="01PLAN00000000000000000002",
|
|
tool_name="file-write",
|
|
skill_name="code-skill",
|
|
arguments_json=json.dumps({"path": "test.py"}),
|
|
result_json=json.dumps({"status": "ok"}),
|
|
error=None,
|
|
success=True,
|
|
duration_ms=55.5,
|
|
started_at=datetime.now(UTC).isoformat(),
|
|
completed_at=datetime.now(UTC).isoformat(),
|
|
change_ids_json=json.dumps(["chg-1", "chg-2"]),
|
|
sequence_number=5,
|
|
sandbox_path="/tmp/sandbox",
|
|
resource_refs_json=json.dumps(["res-1"]),
|
|
provider_metadata_json=json.dumps({"model": "gpt-4"}),
|
|
)
|
|
|
|
|
|
@given("I have a coverage ToolInvocationModel row with null numeric fields")
|
|
def step_create_inv_row_null_numerics(context: Context) -> None:
|
|
"""Create a ToolInvocationModel with null numeric fields."""
|
|
context.cov_inv_row = ToolInvocationModel(
|
|
invocation_id="01INV000000000000000000003",
|
|
changeset_id=None,
|
|
plan_id="01PLAN00000000000000000003",
|
|
tool_name="file-read",
|
|
skill_name=None,
|
|
arguments_json=json.dumps({}),
|
|
result_json=None,
|
|
error=None,
|
|
success=True,
|
|
duration_ms=None,
|
|
started_at=datetime.now(UTC).isoformat(),
|
|
completed_at=None,
|
|
change_ids_json=json.dumps([]),
|
|
sequence_number=None,
|
|
sandbox_path=None,
|
|
resource_refs_json=json.dumps([]),
|
|
provider_metadata_json=None,
|
|
)
|
|
|
|
|
|
@given("I have a coverage ToolInvocationModel row with null completed_at")
|
|
def step_create_inv_row_null_completed(context: Context) -> None:
|
|
"""Create a ToolInvocationModel with null completed_at."""
|
|
context.cov_inv_row = ToolInvocationModel(
|
|
invocation_id="01INV000000000000000000004",
|
|
changeset_id=None,
|
|
plan_id="01PLAN00000000000000000004",
|
|
tool_name="file-write",
|
|
skill_name=None,
|
|
arguments_json=json.dumps({}),
|
|
result_json=None,
|
|
error=None,
|
|
success=True,
|
|
duration_ms=10.0,
|
|
started_at=datetime.now(UTC).isoformat(),
|
|
completed_at=None,
|
|
change_ids_json=json.dumps([]),
|
|
sequence_number=0,
|
|
sandbox_path=None,
|
|
resource_refs_json=json.dumps([]),
|
|
provider_metadata_json=None,
|
|
)
|
|
|
|
|
|
@given("I have a coverage ToolInvocationModel row with populated completed_at")
|
|
def step_create_inv_row_populated_completed(context: Context) -> None:
|
|
"""Create a ToolInvocationModel with populated completed_at."""
|
|
context.cov_inv_row = ToolInvocationModel(
|
|
invocation_id="01INV000000000000000000005",
|
|
changeset_id=None,
|
|
plan_id="01PLAN00000000000000000005",
|
|
tool_name="file-write",
|
|
skill_name=None,
|
|
arguments_json=json.dumps({}),
|
|
result_json=None,
|
|
error=None,
|
|
success=True,
|
|
duration_ms=10.0,
|
|
started_at=datetime.now(UTC).isoformat(),
|
|
completed_at=datetime.now(UTC).isoformat(),
|
|
change_ids_json=json.dumps([]),
|
|
sequence_number=0,
|
|
sandbox_path=None,
|
|
resource_refs_json=json.dumps([]),
|
|
provider_metadata_json=None,
|
|
)
|
|
|
|
|
|
@when("I convert the coverage row to a domain ToolInvocation")
|
|
def step_convert_row_to_domain_invocation(context: Context) -> None:
|
|
"""Convert the model row to a domain ToolInvocation."""
|
|
context.cov_domain_inv = ToolInvocationRepository._to_domain(context.cov_inv_row)
|
|
|
|
|
|
@then("the domain invocation arguments should be an empty dict")
|
|
def step_assert_inv_args_empty(context: Context) -> None:
|
|
"""Assert arguments is empty dict."""
|
|
assert context.cov_domain_inv.arguments == {}, (
|
|
f"Expected empty dict, got {context.cov_domain_inv.arguments!r}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation result should be None")
|
|
def step_assert_inv_result_none(context: Context) -> None:
|
|
"""Assert result is None."""
|
|
assert context.cov_domain_inv.result is None, (
|
|
f"Expected result=None, got {context.cov_domain_inv.result!r}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation change_ids should be an empty list")
|
|
def step_assert_inv_change_ids_empty(context: Context) -> None:
|
|
"""Assert change_ids is empty list."""
|
|
assert context.cov_domain_inv.change_ids == [], (
|
|
f"Expected empty list, got {context.cov_domain_inv.change_ids!r}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation resource_refs should be an empty list")
|
|
def step_assert_inv_resource_refs_empty(context: Context) -> None:
|
|
"""Assert resource_refs is empty list."""
|
|
assert context.cov_domain_inv.resource_refs == [], (
|
|
f"Expected empty list, got {context.cov_domain_inv.resource_refs!r}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation provider_metadata should be None")
|
|
def step_assert_inv_provider_meta_none(context: Context) -> None:
|
|
"""Assert provider_metadata is None."""
|
|
assert context.cov_domain_inv.provider_metadata is None, (
|
|
f"Expected None, got {context.cov_domain_inv.provider_metadata!r}"
|
|
)
|
|
|
|
|
|
@then('the domain invocation arguments should have key "{key}"')
|
|
def step_assert_inv_args_has_key(context: Context, key: str) -> None:
|
|
"""Assert arguments dict has given key."""
|
|
assert key in context.cov_domain_inv.arguments, (
|
|
f"Expected key '{key}' in arguments, got {context.cov_domain_inv.arguments!r}"
|
|
)
|
|
|
|
|
|
@then('the domain invocation result should have key "{key}"')
|
|
def step_assert_inv_result_has_key(context: Context, key: str) -> None:
|
|
"""Assert result dict has given key."""
|
|
assert context.cov_domain_inv.result is not None, "Expected result to not be None"
|
|
assert key in context.cov_domain_inv.result, (
|
|
f"Expected key '{key}' in result, got {context.cov_domain_inv.result!r}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation change_ids should have {count:d} entries")
|
|
def step_assert_inv_change_ids_count(context: Context, count: int) -> None:
|
|
"""Assert change_ids has expected number of entries."""
|
|
assert len(context.cov_domain_inv.change_ids) == count, (
|
|
f"Expected {count} change_ids, got {len(context.cov_domain_inv.change_ids)}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation resource_refs should have {count:d} entry")
|
|
def step_assert_inv_resource_refs_count(context: Context, count: int) -> None:
|
|
"""Assert resource_refs has expected count."""
|
|
assert len(context.cov_domain_inv.resource_refs) == count, (
|
|
f"Expected {count} resource_refs, got {len(context.cov_domain_inv.resource_refs)}"
|
|
)
|
|
|
|
|
|
@then('the domain invocation provider_metadata should have key "{key}"')
|
|
def step_assert_inv_provider_meta_has_key(context: Context, key: str) -> None:
|
|
"""Assert provider_metadata dict has given key."""
|
|
assert context.cov_domain_inv.provider_metadata is not None, (
|
|
"Expected provider_metadata to not be None"
|
|
)
|
|
assert key in context.cov_domain_inv.provider_metadata, (
|
|
f"Expected key '{key}' in provider_metadata, got {context.cov_domain_inv.provider_metadata!r}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation duration_ms should be 0.0")
|
|
def step_assert_inv_duration_zero(context: Context) -> None:
|
|
"""Assert duration_ms is 0.0."""
|
|
assert context.cov_domain_inv.duration_ms == 0.0, (
|
|
f"Expected duration_ms=0.0, got {context.cov_domain_inv.duration_ms}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation sequence_number should be 0")
|
|
def step_assert_inv_seq_zero(context: Context) -> None:
|
|
"""Assert sequence_number is 0."""
|
|
assert context.cov_domain_inv.sequence_number == 0, (
|
|
f"Expected sequence_number=0, got {context.cov_domain_inv.sequence_number}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation completed_at should be None")
|
|
def step_assert_inv_completed_none(context: Context) -> None:
|
|
"""Assert completed_at is None."""
|
|
assert context.cov_domain_inv.completed_at is None, (
|
|
f"Expected completed_at=None, got {context.cov_domain_inv.completed_at!r}"
|
|
)
|
|
|
|
|
|
@then("the domain invocation completed_at should be a datetime")
|
|
def step_assert_inv_completed_datetime(context: Context) -> None:
|
|
"""Assert completed_at is a datetime."""
|
|
assert isinstance(context.cov_domain_inv.completed_at, datetime), (
|
|
f"Expected datetime, got {type(context.cov_domain_inv.completed_at)}"
|
|
)
|
|
|
|
|
|
# ------ SqliteChangeSetStore steps ------
|
|
|
|
|
|
@given("I have a coverage sqlite changeset store")
|
|
def step_create_coverage_store(context: Context) -> None:
|
|
"""Create a SqliteChangeSetStore backed by in-memory SQLite."""
|
|
factory, session, engine = _make_session_factory()
|
|
context.cov_engine = engine
|
|
context.cov_session_factory = factory
|
|
context.cov_store = SqliteChangeSetStore(factory)
|
|
context._cleanup_handlers.append(lambda: session.close())
|
|
context._cleanup_handlers.append(lambda: engine.dispose())
|
|
|
|
|
|
@when("I coverage-get a changeset with empty string")
|
|
def step_get_changeset_empty(context: Context) -> None:
|
|
"""Get a changeset with empty string ID."""
|
|
context.cov_get_result = context.cov_store.get("")
|
|
|
|
|
|
@when('I coverage-get a changeset with id "{changeset_id}"')
|
|
def step_get_changeset_by_id(context: Context, changeset_id: str) -> None:
|
|
"""Get a changeset by specific ID."""
|
|
context.cov_get_result = context.cov_store.get(changeset_id)
|
|
|
|
|
|
@when('I coverage-start a changeset for plan "{plan_id}"')
|
|
def step_start_changeset_coverage(context: Context, plan_id: str) -> None:
|
|
"""Start a changeset for a plan."""
|
|
context.cov_changeset_id = context.cov_store.start(plan_id)
|
|
context.cov_plan_id = plan_id
|
|
|
|
|
|
@when("I coverage-get the started changeset")
|
|
def step_get_started_changeset(context: Context) -> None:
|
|
"""Get the changeset that was just started."""
|
|
context.cov_get_result = context.cov_store.get(context.cov_changeset_id)
|
|
|
|
|
|
@when("I coverage-record a create entry in the started changeset")
|
|
def step_record_entry_coverage(context: Context) -> None:
|
|
"""Record a create entry in the started changeset."""
|
|
entry = _make_change_entry(
|
|
plan_id=context.cov_plan_id,
|
|
operation=ChangeOperation.CREATE,
|
|
path="src/new_cov.py",
|
|
)
|
|
context.cov_store.record(context.cov_changeset_id, entry)
|
|
# Commit the session so the entry is visible
|
|
session = context.cov_session_factory()
|
|
session.commit()
|
|
|
|
|
|
@when("I try to coverage-record an entry with empty changeset_id")
|
|
def step_record_empty_changeset_id(context: Context) -> None:
|
|
"""Attempt to record entry with empty changeset_id."""
|
|
entry = _make_change_entry()
|
|
try:
|
|
context.cov_store.record("", entry)
|
|
context.cov_raised = None
|
|
except ValueError as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when("I coverage-get_for_plan with empty plan_id")
|
|
def step_get_for_plan_empty(context: Context) -> None:
|
|
"""Get changesets for empty plan_id."""
|
|
context.cov_for_plan_result = context.cov_store.get_for_plan("")
|
|
|
|
|
|
@when('I coverage-get_for_plan with plan_id "{plan_id}"')
|
|
def step_get_for_plan_by_id(context: Context, plan_id: str) -> None:
|
|
"""Get changesets for specific plan_id."""
|
|
context.cov_for_plan_result = context.cov_store.get_for_plan(plan_id)
|
|
|
|
|
|
@when("I coverage-summarize changeset with empty string")
|
|
def step_summarize_empty(context: Context) -> None:
|
|
"""Summarize changeset with empty string ID."""
|
|
context.cov_summarize_result = context.cov_store.summarize("")
|
|
|
|
|
|
@when("I coverage-summarize the started changeset")
|
|
def step_summarize_started(context: Context) -> None:
|
|
"""Summarize the changeset that was started."""
|
|
context.cov_summarize_result = context.cov_store.summarize(context.cov_changeset_id)
|
|
|
|
|
|
@when("I try to coverage-delete_for_plan with empty plan_id")
|
|
def step_delete_for_plan_empty(context: Context) -> None:
|
|
"""Attempt to delete for empty plan_id."""
|
|
try:
|
|
context.cov_store.delete_for_plan("")
|
|
context.cov_raised = None
|
|
except ValueError as exc:
|
|
context.cov_raised = exc
|
|
|
|
|
|
@when('I coverage-delete_for_plan "{plan_id}"')
|
|
def step_delete_for_plan_coverage(context: Context, plan_id: str) -> None:
|
|
"""Delete all entries for a plan."""
|
|
context.cov_delete_count = context.cov_store.delete_for_plan(plan_id)
|
|
|
|
|
|
@then("the coverage-get result should be None")
|
|
def step_assert_get_result_none(context: Context) -> None:
|
|
"""Assert get result is None."""
|
|
assert context.cov_get_result is None, (
|
|
f"Expected None, got {context.cov_get_result!r}"
|
|
)
|
|
|
|
|
|
@then("the coverage-get result should be a SpecChangeSet with {count:d} entries")
|
|
def step_assert_get_result_spec_changeset(context: Context, count: int) -> None:
|
|
"""Assert get result is a SpecChangeSet with expected entries."""
|
|
from cleveragents.domain.models.core.change import SpecChangeSet
|
|
|
|
assert context.cov_get_result is not None, "Expected SpecChangeSet, got None"
|
|
assert isinstance(context.cov_get_result, SpecChangeSet), (
|
|
f"Expected SpecChangeSet, got {type(context.cov_get_result)}"
|
|
)
|
|
assert len(context.cov_get_result.entries) == count, (
|
|
f"Expected {count} entries, got {len(context.cov_get_result.entries)}"
|
|
)
|
|
|
|
|
|
@then('the coverage-get result plan_id should be "{expected}"')
|
|
def step_assert_get_result_plan_id(context: Context, expected: str) -> None:
|
|
"""Assert get result plan_id matches expected."""
|
|
assert context.cov_get_result.plan_id == expected, (
|
|
f"Expected plan_id={expected!r}, got {context.cov_get_result.plan_id!r}"
|
|
)
|
|
|
|
|
|
@then("the coverage-get_for_plan result should be an empty list")
|
|
def step_assert_for_plan_empty(context: Context) -> None:
|
|
"""Assert get_for_plan result is empty list."""
|
|
assert context.cov_for_plan_result == [], (
|
|
f"Expected empty list, got {context.cov_for_plan_result!r}"
|
|
)
|
|
|
|
|
|
@then("the coverage-summarize result should be an empty dict")
|
|
def step_assert_summarize_empty(context: Context) -> None:
|
|
"""Assert summarize result is empty dict."""
|
|
assert context.cov_summarize_result == {}, (
|
|
f"Expected empty dict, got {context.cov_summarize_result!r}"
|
|
)
|
|
|
|
|
|
@then("the coverage-summarize result total should be {count:d}")
|
|
def step_assert_summarize_total(context: Context, count: int) -> None:
|
|
"""Assert summarize result total matches expected."""
|
|
assert context.cov_summarize_result.get("total") == count, (
|
|
f"Expected total={count}, got {context.cov_summarize_result!r}"
|
|
)
|
|
|
|
|
|
@then("the coverage-delete count should be {count:d}")
|
|
def step_assert_delete_count(context: Context, count: int) -> None:
|
|
"""Assert delete count matches expected."""
|
|
assert context.cov_delete_count == count, (
|
|
f"Expected delete count={count}, got {context.cov_delete_count}"
|
|
)
|
|
|
|
|
|
# ------ Shared assertion steps ------
|
|
|
|
|
|
@then('a coverage ValueError should be raised with message "{message}"')
|
|
def step_assert_coverage_value_error(context: Context, message: str) -> None:
|
|
"""Assert a ValueError was raised with expected message."""
|
|
assert isinstance(context.cov_raised, ValueError), (
|
|
f"Expected ValueError, got {type(context.cov_raised).__name__}: {context.cov_raised!r}"
|
|
)
|
|
assert message in str(context.cov_raised), (
|
|
f"Expected message containing '{message}', got '{context.cov_raised}'"
|
|
)
|
|
|
|
|
|
@then('a coverage TypeError should be raised with message "{message}"')
|
|
def step_assert_coverage_type_error(context: Context, message: str) -> None:
|
|
"""Assert a TypeError was raised with expected message."""
|
|
assert isinstance(context.cov_raised, TypeError), (
|
|
f"Expected TypeError, got {type(context.cov_raised).__name__}: {context.cov_raised!r}"
|
|
)
|
|
assert message in str(context.cov_raised), (
|
|
f"Expected message containing '{message}', got '{context.cov_raised}'"
|
|
)
|
|
|
|
|
|
@then('a coverage DatabaseError should be raised with message "{message}"')
|
|
def step_assert_coverage_db_error(context: Context, message: str) -> None:
|
|
"""Assert a DatabaseError was raised with expected message."""
|
|
assert isinstance(context.cov_raised, (DatabaseError, OperationalError)), (
|
|
f"Expected DatabaseError or OperationalError, got {type(context.cov_raised).__name__}: {context.cov_raised!r}"
|
|
)
|
|
assert message in str(context.cov_raised), (
|
|
f"Expected message containing '{message}', got '{context.cov_raised}'"
|
|
)
|