cleveragents-core/features/steps/changeset_capture_steps.py

"""Step definitions for changeset capture feature tests."""

import os
import tempfile
from datetime import UTC, datetime

from behave import given, then, when

from cleveragents.domain.models.core.change import (
    ChangeEntry,
    ChangeOperation,
    InMemoryChangeSetStore,
    SpecChangeSet,
)
from cleveragents.tool.builtins.changeset import (
    ChangeSetCapture,
    ChangeSetEntry,
)

# ---- ChangeOperation enum ----


@given("I import the ChangeOperation enum")
def step_import_change_operation(context):
    """Import ChangeOperation enum."""
    context.enum_cls = ChangeOperation


@then("it should have members CREATE, MODIFY, DELETE, RENAME")
def step_verify_enum_members(context):
    """Verify enum members."""
    assert hasattr(context.enum_cls, "CREATE")
    assert hasattr(context.enum_cls, "MODIFY")
    assert hasattr(context.enum_cls, "DELETE")
    assert hasattr(context.enum_cls, "RENAME")
    assert context.enum_cls.CREATE == "create"
    assert context.enum_cls.MODIFY == "modify"
    assert context.enum_cls.DELETE == "delete"
    assert context.enum_cls.RENAME == "rename"


# ---- ChangeEntry model ----


@given("I import the ChangeEntry model")
def step_import_change_entry(context):
    """Import ChangeEntry model."""
    context.entry_cls = ChangeEntry


@when("I create a ChangeEntry for a create operation")
def step_create_entry_create(context):
    """Create a ChangeEntry for create."""
    context.entry = ChangeEntry(
        plan_id="01HXYZ",
        resource_id="res-1",
        tool_name="builtin/file-write",
        operation=ChangeOperation.CREATE,
        path="src/new_file.py",
        after_hash="abc123",
    )


@then("the entry should have a ULID entry_id")
def step_entry_has_ulid(context):
    """Verify ULID entry_id."""
    assert context.entry.entry_id
    assert len(context.entry.entry_id) > 0


@then('the entry should have operation "{op}"')
def step_entry_operation(context, op):
    """Verify operation value."""
    assert context.entry.operation == op


@then("the entry should have a UTC timestamp")
def step_entry_utc_timestamp(context):
    """Verify UTC timestamp."""
    ts = context.entry.timestamp
    assert isinstance(ts, datetime)
    assert ts.tzinfo is not None
    assert ts.tzinfo in (UTC, UTC)


@then("the before_hash should be None")
def step_before_hash_none(context):
    """Verify before_hash is None."""
    assert context.entry.before_hash is None


@when("I create a ChangeEntry for a modify operation with hashes")
def step_create_entry_modify(context):
    """Create a ChangeEntry for modify with hashes."""
    context.entry = ChangeEntry(
        plan_id="01HXYZ",
        resource_id="res-1",
        tool_name="builtin/file-edit",
        operation=ChangeOperation.MODIFY,
        path="src/existing.py",
        before_hash="hash-before",
        after_hash="hash-after",
    )


@then("the before_hash should not be None")
def step_before_hash_not_none(context):
    """Verify before_hash is set."""
    assert context.entry.before_hash is not None


@then("the after_hash should not be None")
def step_after_hash_not_none(context):
    """Verify after_hash is set."""
    assert context.entry.after_hash is not None


@when("I create a ChangeEntry for a delete operation")
def step_create_entry_delete(context):
    """Create a ChangeEntry for delete."""
    context.entry = ChangeEntry(
        plan_id="01HXYZ",
        resource_id="res-1",
        tool_name="builtin/file-delete",
        operation=ChangeOperation.DELETE,
        path="src/old_file.py",
        before_hash="hash-of-deleted",
    )


@then("the after_hash should be None")
def step_after_hash_none(context):
    """Verify after_hash is None."""
    assert context.entry.after_hash is None


@when("I create a ChangeEntry for a rename operation")
def step_create_entry_rename(context):
    """Create a ChangeEntry for rename."""
    context.entry = ChangeEntry(
        plan_id="01HXYZ",
        resource_id="res-1",
        tool_name="builtin/file-move",
        operation=ChangeOperation.RENAME,
        path="src/old_name.py",
        before_hash="hash-x",
        after_hash="hash-x",
    )


# ---- SpecChangeSet model ----


@given("I import the SpecChangeSet model")
def step_import_spec_changeset(context):
    """Import SpecChangeSet."""
    context.cs_cls = SpecChangeSet


def _make_mixed_entries():
    """Build a list of entries with mixed operations."""
    base = {
        "plan_id": "plan-1",
        "resource_id": "res-A",
        "tool_name": "builtin/file-write",
    }
    return [
        ChangeEntry(
            **base,
            operation=ChangeOperation.CREATE,
            path="a.py",
        ),
        ChangeEntry(
            **base,
            operation=ChangeOperation.MODIFY,
            path="b.py",
        ),
        ChangeEntry(
            plan_id="plan-1",
            resource_id="res-B",
            tool_name="builtin/file-edit",
            operation=ChangeOperation.MODIFY,
            path="c.py",
        ),
        ChangeEntry(
            **base,
            operation=ChangeOperation.DELETE,
            path="d.py",
        ),
        ChangeEntry(
            **base,
            operation=ChangeOperation.RENAME,
            path="e.py",
        ),
    ]


@when("I create a SpecChangeSet with mixed operations")
def step_create_mixed_changeset(context):
    """Create a SpecChangeSet with mixed operations."""
    context.spec_cs = SpecChangeSet(
        plan_id="plan-1",
        entries=_make_mixed_entries(),
    )


@then("the creates count should be {n:d}")
def step_creates_count(context, n):
    """Verify creates count."""
    assert context.spec_cs.creates == n


@then("the modifies count should be {n:d}")
def step_modifies_count(context, n):
    """Verify modifies count."""
    assert context.spec_cs.modifies == n


@then("the deletes count should be {n:d}")
def step_deletes_count(context, n):
    """Verify deletes count."""
    assert context.spec_cs.deletes == n


@then("the renames count should be {n:d}")
def step_renames_count(context, n):
    """Verify renames count."""
    assert context.spec_cs.renames == n


@then("paths_changed should have {n:d} entries")
def step_paths_changed(context, n):
    """Verify paths_changed count."""
    assert len(context.spec_cs.paths_changed) == n


@then("resources_involved should have {n:d} entries")
def step_resources_involved(context, n):
    """Verify resources_involved count."""
    assert len(context.spec_cs.resources_involved) == n


@then("the summary dict should have correct totals")
def step_summary_dict(context):
    """Verify summary dict."""
    s = context.spec_cs.summary()
    assert s["total"] == 5
    assert s["creates"] == 1
    assert s["modifies"] == 2
    assert s["deletes"] == 1
    assert s["renames"] == 1
    assert s["paths_changed"] == 5
    assert s["resources_involved"] == 2


@when("I create an empty SpecChangeSet")
def step_create_empty_changeset(context):
    """Create an empty SpecChangeSet."""
    context.spec_cs = SpecChangeSet(plan_id="plan-empty")


# ---- InMemoryChangeSetStore ----


@given("I have an InMemoryChangeSetStore")
def step_create_store(context):
    """Create InMemoryChangeSetStore."""
    context.store = InMemoryChangeSetStore()
    context.changeset_ids = []


@when('I start a changeset for plan "{plan_id}"')
def step_start_changeset(context, plan_id):
    """Start a changeset."""
    cs_id = context.store.start(plan_id)
    context.changeset_ids.append(cs_id)
    context.last_cs_id = cs_id


@when("I record a create entry in the changeset")
def step_record_create_entry(context):
    """Record a create entry."""
    # Use the plan_id from the changeset so the entry matches
    # (InMemoryChangeSetStore.record now validates via add_change).
    cs = context.store.get(context.last_cs_id)
    plan_id = cs.plan_id if cs is not None else "plan-abc"
    entry = ChangeEntry(
        plan_id=plan_id,
        resource_id="res-1",
        tool_name="builtin/file-write",
        operation=ChangeOperation.CREATE,
        path="new.py",
    )
    context.store.record(context.last_cs_id, entry)


@then("I can get the changeset by ID")
def step_get_changeset(context):
    """Verify get returns the changeset."""
    cs = context.store.get(context.last_cs_id)
    assert cs is not None
    context.fetched_cs = cs


@then("the store changeset should have {n:d} entry")
def step_store_changeset_entry_count(context, n):
    """Verify entry count in store."""
    cs = context.store.get(context.last_cs_id)
    assert cs is not None
    assert len(cs.entries) == n


@then('get_for_plan "{plan_id}" should return {n:d} changesets')
def step_get_for_plan_count(context, plan_id, n):
    """Verify get_for_plan returns expected count."""
    result = context.store.get_for_plan(plan_id)
    assert len(result) == n


@then('get_for_plan "{plan_id}" should return {n:d} changeset')
def step_get_for_plan_singular(context, plan_id, n):
    """Verify get_for_plan returns expected count (singular)."""
    result = context.store.get_for_plan(plan_id)
    assert len(result) == n


@when("I record a modify entry in the changeset")
def step_record_modify_entry(context):
    """Record a modify entry."""
    entry = ChangeEntry(
        plan_id="plan-s",
        resource_id="res-1",
        tool_name="builtin/file-edit",
        operation=ChangeOperation.MODIFY,
        path="existing.py",
        before_hash="aaa",
        after_hash="bbb",
    )
    context.store.record(context.last_cs_id, entry)


@then("the summarized changeset should show {n:d} total")
def step_summarize_total(context, n):
    """Verify summarize total."""
    s = context.store.summarize(context.last_cs_id)
    assert s["total"] == n


@then("getting a non-existent changeset returns None")
def step_get_nonexistent(context):
    """Verify get returns None for unknown ID."""
    assert context.store.get("nonexistent-id") is None


@then("recording to a non-existent changeset raises KeyError")
def step_record_nonexistent(context):
    """Verify recording to unknown ID raises KeyError."""
    entry = ChangeEntry(
        plan_id="plan-x",
        resource_id="res-1",
        tool_name="t",
        operation=ChangeOperation.CREATE,
        path="x.py",
    )
    try:
        context.store.record("nonexistent-id", entry)
        raise AssertionError("Expected KeyError")
    except KeyError:
        pass


@then("summarizing a non-existent changeset returns empty dict")
def step_summarize_nonexistent(context):
    """Verify summarize returns {} for unknown ID."""
    assert context.store.summarize("nonexistent-id") == {}


# ---- ChangeSetCapture integration ----


@given('I have a ChangeSetCapture with resource_id "{rid}"')
def step_capture_with_resource(context, rid):
    """Create ChangeSetCapture with resource_id."""
    context.capture = ChangeSetCapture(
        plan_id="plan-cap",
        resource_id=rid,
    )


@when("I create a ChangeSetEntry via capture")
def step_create_via_capture(context):
    """Add an entry via the capture object."""
    entry = ChangeSetEntry(
        operation="create",
        path="test.py",
        resource_id=None,
        tool_name="builtin/file-write",
    )
    context.capture._entries.append(entry)
    # Resource ID comes from capture default
    context.last_entry = context.capture.get_changeset().entries[-1]


@then('the entry resource_id should be "{rid}"')
def step_entry_resource_id(context, rid):
    """Verify resource_id is empty — set on domain convert."""
    # The lightweight entry may have None; the capture stores
    # the resource_id on the capture object itself.
    assert context.capture._resource_id == rid


@then("the entry tool_name should be set")
def step_entry_tool_name_set(context):
    """Verify tool_name is set."""
    assert context.last_entry.tool_name is not None


@given("I have a ChangeSetCapture with sandbox_root")
def step_capture_with_sandbox(context):
    """Create capture with a sandbox root."""
    context.tmpdir = tempfile.mkdtemp()
    context.capture = ChangeSetCapture(
        plan_id="plan-sb",
        resource_id="res-sb",
        sandbox_root=context.tmpdir,
    )


@when("I capture a write to a nested path")
def step_capture_nested_write(context):
    """Simulate capturing a nested path write."""
    from cleveragents.tool.builtins.changeset import (
        _normalize_path,
    )

    context.normalized = _normalize_path(
        "sub/dir/file.py", context.capture._sandbox_root
    )


@then("the captured path should be repo-relative")
def step_path_repo_relative(context):
    """Verify the path is relative."""
    assert not os.path.isabs(context.normalized)
    assert context.normalized == os.path.join("sub", "dir", "file.py")


@when("I add several entries via capture")
def step_add_entries_via_capture(context):
    """Add several entries directly."""
    for op in ("create", "modify", "delete"):
        entry = ChangeSetEntry(
            operation=op,
            path=f"{op}_file.py",
            resource_id="res-2",
            tool_name="builtin/file-write",
        )
        context.capture._entries.append(entry)


@then("to_spec_changeset should return a SpecChangeSet")
def step_spec_changeset_conversion(context):
    """Verify spec changeset conversion."""
    spec = context.capture.to_spec_changeset()
    assert isinstance(spec, SpecChangeSet)
    assert len(spec.entries) == 3
    assert spec.plan_id == "plan-cap"


# ---- Multi-resource plan ----


@given("I have two ChangeSetCapture instances for different resources")
def step_two_captures(context):
    """Create two captures for different resources."""
    context.capture_a = ChangeSetCapture(
        plan_id="plan-multi",
        resource_id="res-A",
    )
    context.capture_b = ChangeSetCapture(
        plan_id="plan-multi",
        resource_id="res-B",
    )


@when("each capture records changes for its resource")
def step_record_per_resource(context):
    """Record one entry in each capture."""
    entry_a = ChangeSetEntry(
        operation="create",
        path="a.py",
        resource_id="res-A",
        tool_name="builtin/file-write",
    )
    entry_b = ChangeSetEntry(
        operation="modify",
        path="b.py",
        resource_id="res-B",
        tool_name="builtin/file-edit",
    )
    context.capture_a._entries.append(entry_a)
    context.capture_b._entries.append(entry_b)


@then("each changeset should only have its resource entries")
def step_verify_per_resource(context):
    """Verify each changeset has only its resource."""
    cs_a = context.capture_a.get_changeset()
    cs_b = context.capture_b.get_changeset()
    assert len(cs_a.entries) == 1
    assert cs_a.entries[0].resource_id == "res-A"
    assert len(cs_b.entries) == 1
    assert cs_b.entries[0].resource_id == "res-B"