CoreRasurae
629 lines
19 KiB
Python
629 lines
19 KiB
Python
"""Step definitions for ChangeSet models and invocation tracking."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.domain.models.core.change import (
|
|
ChangeEntry,
|
|
ChangeOperation,
|
|
ChangeType,
|
|
InMemoryInvocationTracker,
|
|
SpecChangeSet,
|
|
ToolInvocation,
|
|
normalize_change_path,
|
|
)
|
|
|
|
__all__: list[str] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ChangeType alias
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("I import ChangeType and ChangeOperation")
|
|
def step_import_change_type(context: Any) -> None:
|
|
context.change_type = ChangeType
|
|
context.change_operation = ChangeOperation
|
|
|
|
|
|
@then("ChangeType should be the same class as ChangeOperation")
|
|
def step_changetype_same(context: Any) -> None:
|
|
assert context.change_type is context.change_operation
|
|
|
|
|
|
@then('ChangeType CREATE should equal "{value}"')
|
|
def step_changetype_create_value(context: Any, value: str) -> None:
|
|
assert value == ChangeType.CREATE
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ChangeEntry validators
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _base_entry_kwargs() -> dict[str, Any]:
|
|
return {
|
|
"plan_id": "plan-test",
|
|
"resource_id": "res-test",
|
|
"tool_name": "builtin/file-write",
|
|
"path": "src/test.py",
|
|
}
|
|
|
|
|
|
@when('I create a ChangeEntry with CREATE and before_hash "{h}"')
|
|
def step_create_entry_create_before(context: Any, h: str) -> None:
|
|
try:
|
|
ChangeEntry(
|
|
**_base_entry_kwargs(),
|
|
operation=ChangeOperation.CREATE,
|
|
before_hash=h,
|
|
after_hash="xyz",
|
|
)
|
|
context.validation_error = None
|
|
except Exception as exc:
|
|
context.validation_error = str(exc)
|
|
|
|
|
|
@when('I create a ChangeEntry with DELETE and after_hash "{h}"')
|
|
def step_create_entry_delete_after(context: Any, h: str) -> None:
|
|
try:
|
|
ChangeEntry(
|
|
**_base_entry_kwargs(),
|
|
operation=ChangeOperation.DELETE,
|
|
before_hash="xyz",
|
|
after_hash=h,
|
|
)
|
|
context.validation_error = None
|
|
except Exception as exc:
|
|
context.validation_error = str(exc)
|
|
|
|
|
|
@then('a validation error should be raised mentioning "{text}"')
|
|
def step_validation_error(context: Any, text: str) -> None:
|
|
assert context.validation_error is not None, "No error was raised"
|
|
assert text in context.validation_error, (
|
|
f"Expected '{text}' in: {context.validation_error}"
|
|
)
|
|
|
|
|
|
@when('I create a valid CREATE ChangeEntry with after_hash "{h}"')
|
|
def step_valid_create_entry(context: Any, h: str) -> None:
|
|
context.entry = ChangeEntry(
|
|
**_base_entry_kwargs(),
|
|
operation=ChangeOperation.CREATE,
|
|
after_hash=h,
|
|
)
|
|
|
|
|
|
@when("I create a valid MODIFY ChangeEntry with both hashes")
|
|
def step_valid_modify_entry(context: Any) -> None:
|
|
context.entry = ChangeEntry(
|
|
**_base_entry_kwargs(),
|
|
operation=ChangeOperation.MODIFY,
|
|
before_hash="hash-a",
|
|
after_hash="hash-b",
|
|
)
|
|
|
|
|
|
@when("I create a RENAME ChangeEntry without hashes")
|
|
def step_rename_no_hashes(context: Any) -> None:
|
|
context.entry = ChangeEntry(
|
|
**_base_entry_kwargs(),
|
|
operation=ChangeOperation.RENAME,
|
|
)
|
|
|
|
|
|
@when("I create a CREATE ChangeEntry without any hashes")
|
|
def step_create_no_hashes(context: Any) -> None:
|
|
context.entry = ChangeEntry(
|
|
**_base_entry_kwargs(),
|
|
operation=ChangeOperation.CREATE,
|
|
)
|
|
|
|
|
|
@then('the change entry operation should be "{op}"')
|
|
def step_change_entry_operation(context: Any, op: str) -> None:
|
|
assert context.entry.operation == op
|
|
|
|
|
|
@then('the change entry after_hash should be "{h}"')
|
|
def step_change_entry_after_hash(context: Any, h: str) -> None:
|
|
assert context.entry.after_hash == h
|
|
|
|
|
|
@then("the change entry has_integrity_hashes should be true")
|
|
def step_change_entry_integrity_true(context: Any) -> None:
|
|
assert context.entry.has_integrity_hashes is True
|
|
|
|
|
|
@then("the change entry has_integrity_hashes should be false")
|
|
def step_change_entry_integrity_false(context: Any) -> None:
|
|
assert context.entry.has_integrity_hashes is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Path normalization
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a repo root "{root}"')
|
|
def step_repo_root(context: Any, root: str) -> None:
|
|
context.repo_root = root
|
|
|
|
|
|
@given("an empty repo root")
|
|
def step_empty_repo_root(context: Any) -> None:
|
|
context.repo_root = ""
|
|
|
|
|
|
@when('I normalize the path "{path}"')
|
|
def step_normalize_path(context: Any, path: str) -> None:
|
|
context.normalized = normalize_change_path(path, context.repo_root)
|
|
|
|
|
|
@then('the normalized path should be "{expected}"')
|
|
def step_normalized_result(context: Any, expected: str) -> None:
|
|
assert context.normalized == expected, (
|
|
f"Expected '{expected}', got '{context.normalized}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SpecChangeSet ordering
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_entry(resource_id: str, path: str, ts_offset: int = 0) -> ChangeEntry:
|
|
"""Create a ChangeEntry with specific ordering fields."""
|
|
return ChangeEntry(
|
|
plan_id="plan-order",
|
|
resource_id=resource_id,
|
|
tool_name="builtin/file-write",
|
|
operation=ChangeOperation.CREATE,
|
|
path=path,
|
|
after_hash="hash-new",
|
|
timestamp=datetime(2026, 1, 1, 0, 0, ts_offset, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
@given("a SpecChangeSet with entries in random order")
|
|
def step_changeset_random_order(context: Any) -> None:
|
|
entries = [
|
|
_make_entry("res-B", "z.py", 3),
|
|
_make_entry("res-A", "b.py", 1),
|
|
_make_entry("res-B", "a.py", 2),
|
|
_make_entry("res-A", "a.py", 0),
|
|
]
|
|
context.changeset = SpecChangeSet(plan_id="plan-order", entries=entries)
|
|
|
|
|
|
@when("I call sorted_entries")
|
|
def step_call_sorted(context: Any) -> None:
|
|
context.sorted = context.changeset.sorted_entries()
|
|
|
|
|
|
@then("entries should be sorted by resource_id then path then timestamp")
|
|
def step_verify_sorted(context: Any) -> None:
|
|
keys = [(e.resource_id, e.path, e.timestamp) for e in context.sorted]
|
|
assert keys == sorted(keys), f"Not sorted: {keys}"
|
|
|
|
|
|
@given("a SpecChangeSet with entries for multiple resources")
|
|
def step_changeset_multi_resource(context: Any) -> None:
|
|
entries = [
|
|
_make_entry("res-B", "y.py", 1),
|
|
_make_entry("res-A", "x.py", 0),
|
|
_make_entry("res-B", "w.py", 2),
|
|
_make_entry("res-A", "z.py", 3),
|
|
]
|
|
context.changeset = SpecChangeSet(plan_id="plan-grp", entries=entries)
|
|
|
|
|
|
@when("I call grouped_by_resource")
|
|
def step_call_grouped(context: Any) -> None:
|
|
context.groups = context.changeset.grouped_by_resource()
|
|
|
|
|
|
@then("entries should be grouped by resource_id")
|
|
def step_verify_grouped(context: Any) -> None:
|
|
assert set(context.groups.keys()) == {"res-A", "res-B"}
|
|
|
|
|
|
@then("each group should be sorted by path and timestamp")
|
|
def step_verify_group_sort(context: Any) -> None:
|
|
for entries in context.groups.values():
|
|
keys = [(e.path, e.timestamp) for e in entries]
|
|
assert keys == sorted(keys), f"Not sorted: {keys}"
|
|
|
|
|
|
@given("an empty SpecChangeSet")
|
|
def step_empty_changeset(context: Any) -> None:
|
|
context.changeset = SpecChangeSet(plan_id="plan-empty")
|
|
|
|
|
|
@when("I add a change entry to it")
|
|
def step_add_change(context: Any) -> None:
|
|
entry = ChangeEntry(
|
|
plan_id="plan-empty",
|
|
resource_id="res-1",
|
|
tool_name="builtin/file-write",
|
|
operation=ChangeOperation.CREATE,
|
|
path="new.py",
|
|
after_hash="hash-new",
|
|
)
|
|
context.changeset.add_change(entry)
|
|
|
|
|
|
@then("the spec changeset should contain {count:d} entry")
|
|
def step_spec_changeset_count(context: Any, count: int) -> None:
|
|
assert len(context.changeset.entries) == count
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ToolInvocation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a ToolInvocation for plan "{pid}" and tool "{tool}"')
|
|
def step_create_invocation(context: Any, pid: str, tool: str) -> None:
|
|
context.invocation = ToolInvocation(plan_id=pid, tool_name=tool)
|
|
|
|
|
|
@then("the invocation should have a ULID invocation_id")
|
|
def step_invocation_ulid(context: Any) -> None:
|
|
assert len(context.invocation.invocation_id) == 26
|
|
|
|
|
|
@then('the invocation plan_id should be "{pid}"')
|
|
def step_invocation_plan(context: Any, pid: str) -> None:
|
|
assert context.invocation.plan_id == pid
|
|
|
|
|
|
@then('the invocation tool_name should be "{tool}"')
|
|
def step_invocation_tool(context: Any, tool: str) -> None:
|
|
assert context.invocation.tool_name == tool
|
|
|
|
|
|
@then("the invocation success should be true")
|
|
def step_invocation_success_true(context: Any) -> None:
|
|
assert context.invocation.success is True
|
|
|
|
|
|
@when('I create a failed ToolInvocation with error "{msg}"')
|
|
def step_failed_invocation(context: Any, msg: str) -> None:
|
|
context.invocation = ToolInvocation(
|
|
plan_id="plan-fail",
|
|
tool_name="builtin/file-read",
|
|
success=False,
|
|
error=msg,
|
|
)
|
|
|
|
|
|
@then("the invocation success should be false")
|
|
def step_invocation_success_false(context: Any) -> None:
|
|
assert context.invocation.success is False
|
|
|
|
|
|
@then('the invocation error should be "{msg}"')
|
|
def step_invocation_error(context: Any, msg: str) -> None:
|
|
assert context.invocation.error == msg
|
|
|
|
|
|
@when("I create a ToolInvocation with change_ids")
|
|
def step_invocation_change_ids(context: Any) -> None:
|
|
context.invocation = ToolInvocation(
|
|
plan_id="plan-chg",
|
|
tool_name="builtin/file-write",
|
|
change_ids=["chg-1", "chg-2"],
|
|
)
|
|
|
|
|
|
@then("the invocation should have {count:d} change_ids")
|
|
def step_invocation_change_count(context: Any, count: int) -> None:
|
|
assert len(context.invocation.change_ids) == count
|
|
|
|
|
|
@when("I create a ToolInvocation with provider metadata")
|
|
def step_invocation_provider(context: Any) -> None:
|
|
context.invocation = ToolInvocation(
|
|
plan_id="plan-prov",
|
|
tool_name="builtin/file-write",
|
|
provider_metadata={
|
|
"model_name": "gpt-4",
|
|
"provider_id": "openai",
|
|
"latency_ms": 150.0,
|
|
},
|
|
)
|
|
|
|
|
|
@then("the invocation provider_metadata should contain model name")
|
|
def step_invocation_provider_meta(context: Any) -> None:
|
|
meta = context.invocation.provider_metadata
|
|
assert meta is not None
|
|
assert "model_name" in meta
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# InMemoryInvocationTracker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an InMemoryInvocationTracker")
|
|
def step_tracker(context: Any) -> None:
|
|
context.tracker = InMemoryInvocationTracker()
|
|
|
|
|
|
@when('I track {count:d} invocations for plan "{pid}"')
|
|
def step_track_n(context: Any, count: int, pid: str) -> None:
|
|
for i in range(count):
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id=pid,
|
|
tool_name=f"tool-{i}",
|
|
sequence_number=i,
|
|
)
|
|
)
|
|
|
|
|
|
@then('get_invocations for "{pid}" should return {count:d}')
|
|
def step_get_invocations_count(context: Any, pid: str, count: int) -> None:
|
|
result = context.tracker.get_invocations(pid)
|
|
assert len(result) == count, f"Expected {count}, got {len(result)}"
|
|
|
|
|
|
@when("I track invocations for different skills")
|
|
def step_track_skills(context: Any) -> None:
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-sk",
|
|
tool_name="t1",
|
|
skill_name="skill-A",
|
|
sequence_number=0,
|
|
)
|
|
)
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-sk",
|
|
tool_name="t2",
|
|
skill_name="skill-B",
|
|
sequence_number=1,
|
|
)
|
|
)
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-sk",
|
|
tool_name="t3",
|
|
skill_name="skill-A",
|
|
sequence_number=2,
|
|
)
|
|
)
|
|
|
|
|
|
@then("get_invocations_for_skill should filter correctly")
|
|
def step_filter_skill(context: Any) -> None:
|
|
a = context.tracker.get_invocations_for_skill("plan-sk", "skill-A")
|
|
assert len(a) == 2
|
|
b = context.tracker.get_invocations_for_skill("plan-sk", "skill-B")
|
|
assert len(b) == 1
|
|
|
|
|
|
@when("I track invocations with change_ids")
|
|
def step_track_with_changes(context: Any) -> None:
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-chg",
|
|
tool_name="t1",
|
|
change_ids=["c1", "c2"],
|
|
sequence_number=0,
|
|
)
|
|
)
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-chg",
|
|
tool_name="t2",
|
|
change_ids=["c3"],
|
|
sequence_number=1,
|
|
)
|
|
)
|
|
|
|
|
|
@then("get_changes should return all change_ids for the plan")
|
|
def step_get_all_changes(context: Any) -> None:
|
|
changes = context.tracker.get_changes("plan-chg")
|
|
assert changes == ["c1", "c2", "c3"]
|
|
|
|
|
|
@when("I track invocations out of order")
|
|
def step_track_out_of_order(context: Any) -> None:
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-ord",
|
|
tool_name="t3",
|
|
sequence_number=2,
|
|
)
|
|
)
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-ord",
|
|
tool_name="t1",
|
|
sequence_number=0,
|
|
)
|
|
)
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-ord",
|
|
tool_name="t2",
|
|
sequence_number=1,
|
|
)
|
|
)
|
|
|
|
|
|
@then("get_invocations should return them in sequence order")
|
|
def step_verify_sequence_order(context: Any) -> None:
|
|
invs = context.tracker.get_invocations("plan-ord")
|
|
seqs = [i.sequence_number for i in invs]
|
|
assert seqs == [0, 1, 2], f"Expected [0, 1, 2], got {seqs}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SpecChangeSet summary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a SpecChangeSet with mixed operation entries")
|
|
def step_changeset_mixed_ops(context: Any) -> None:
|
|
entries = [
|
|
ChangeEntry(
|
|
plan_id="plan-mix",
|
|
resource_id="res-A",
|
|
tool_name="builtin/file-write",
|
|
operation=ChangeOperation.CREATE,
|
|
path="new.py",
|
|
after_hash="h1",
|
|
),
|
|
ChangeEntry(
|
|
plan_id="plan-mix",
|
|
resource_id="res-A",
|
|
tool_name="builtin/file-edit",
|
|
operation=ChangeOperation.MODIFY,
|
|
path="edit.py",
|
|
before_hash="h2a",
|
|
after_hash="h2b",
|
|
),
|
|
ChangeEntry(
|
|
plan_id="plan-mix",
|
|
resource_id="res-B",
|
|
tool_name="builtin/file-delete",
|
|
operation=ChangeOperation.DELETE,
|
|
path="old.py",
|
|
before_hash="h3",
|
|
),
|
|
ChangeEntry(
|
|
plan_id="plan-mix",
|
|
resource_id="res-B",
|
|
tool_name="builtin/file-rename",
|
|
operation=ChangeOperation.RENAME,
|
|
path="moved.py",
|
|
),
|
|
]
|
|
context.changeset = SpecChangeSet(plan_id="plan-mix", entries=entries)
|
|
|
|
|
|
@when("I call summary on the changeset")
|
|
def step_call_summary(context: Any) -> None:
|
|
context.summary_result = context.changeset.summary()
|
|
|
|
|
|
@then("the summary total should be {count:d}")
|
|
def step_summary_total(context: Any, count: int) -> None:
|
|
assert context.summary_result["total"] == count, (
|
|
f"Expected total {count}, got {context.summary_result['total']}"
|
|
)
|
|
|
|
|
|
@then("the summary creates should be {count:d}")
|
|
def step_summary_creates(context: Any, count: int) -> None:
|
|
assert context.summary_result["creates"] == count
|
|
|
|
|
|
@then("the summary modifies should be {count:d}")
|
|
def step_summary_modifies(context: Any, count: int) -> None:
|
|
assert context.summary_result["modifies"] == count
|
|
|
|
|
|
@then("the summary deletes should be {count:d}")
|
|
def step_summary_deletes(context: Any, count: int) -> None:
|
|
assert context.summary_result["deletes"] == count
|
|
|
|
|
|
@then("the summary renames should be {count:d}")
|
|
def step_summary_renames(context: Any, count: int) -> None:
|
|
assert context.summary_result["renames"] == count
|
|
|
|
|
|
@then("the summary paths_changed should be {count:d}")
|
|
def step_summary_paths(context: Any, count: int) -> None:
|
|
assert context.summary_result["paths_changed"] == count
|
|
|
|
|
|
@then("the summary resources_involved should be {count:d}")
|
|
def step_summary_resources(context: Any, count: int) -> None:
|
|
assert context.summary_result["resources_involved"] == count
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# add_change plan_id validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I add a change entry with a different plan_id")
|
|
def step_add_mismatched_plan(context: Any) -> None:
|
|
entry = ChangeEntry(
|
|
plan_id="wrong-plan",
|
|
resource_id="res-1",
|
|
tool_name="builtin/file-write",
|
|
operation=ChangeOperation.CREATE,
|
|
path="file.py",
|
|
after_hash="hash-new",
|
|
)
|
|
try:
|
|
context.changeset.add_change(entry)
|
|
context.validation_error = None
|
|
except Exception as exc:
|
|
context.validation_error = str(exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Empty tracker queries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('get_changes for "{pid}" should return {count:d} ids')
|
|
def step_get_changes_empty(context: Any, pid: str, count: int) -> None:
|
|
changes = context.tracker.get_changes(pid)
|
|
assert len(changes) == count, f"Expected {count}, got {len(changes)}"
|
|
|
|
|
|
@then('get_invocations_for_skill "{pid}" "{skill}" should return {count:d}')
|
|
def step_get_skill_empty(context: Any, pid: str, skill: str, count: int) -> None:
|
|
result = context.tracker.get_invocations_for_skill(pid, skill)
|
|
assert len(result) == count, f"Expected {count}, got {len(result)}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_changes ordering proof
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I track change invocations in reverse sequence order")
|
|
def step_track_changes_reverse(context: Any) -> None:
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-rev",
|
|
tool_name="t2",
|
|
sequence_number=1,
|
|
change_ids=["c-second"],
|
|
)
|
|
)
|
|
context.tracker.track(
|
|
ToolInvocation(
|
|
plan_id="plan-rev",
|
|
tool_name="t1",
|
|
sequence_number=0,
|
|
change_ids=["c-first"],
|
|
)
|
|
)
|
|
|
|
|
|
@then("get_changes should return change_ids in sequence order")
|
|
def step_verify_changes_sequence(context: Any) -> None:
|
|
changes = context.tracker.get_changes("plan-rev")
|
|
assert changes == ["c-first", "c-second"], (
|
|
f"Expected ['c-first', 'c-second'], got {changes}"
|
|
)
|