cleveragents-core/robot/wf02_test_generation_artifacts.py

"""WF02 artifact lifecycle assertions.

Implements artifact generation assertions through lifecycle-produced
changesets and user-facing artifacts output.
"""

from __future__ import annotations

import json
import sys
from io import StringIO
from pathlib import Path

from wf02_test_generation_common import (
    assert_invariants_and_conventions,
    auth_module_fixture,
    create_trusted_plan,
    create_wf02_action,
    generate_tests_from_mock_provider,
    setup_lifecycle,
    validated_relative_test_path,
)

from cleveragents.a2a.facade import A2aLocalFacade
from cleveragents.a2a.models import A2aRequest
from cleveragents.application.services.plan_apply_service import PlanApplyService
from cleveragents.application.services.plan_execution_context import (
    PlanExecutionContext,
)
from cleveragents.domain.models.core.change import (
    ChangeEntry,
    ChangeOperation,
    InMemoryChangeSetStore,
)
from cleveragents.tool.builtins.changeset import ChangeSetCapture
from cleveragents.tool.builtins.file_tools import FILE_WRITE_SPEC


def _capture_changeset_from_file_lifecycle(
    plan_id: str,
    fixture_root: Path,
    generated_files: dict[str, str],
) -> tuple[str, InMemoryChangeSetStore]:
    """Produce a changeset through ChangeSetCapture + file-write lifecycle."""
    capture = ChangeSetCapture(
        plan_id=plan_id,
        resource_id="local/api-service-repo",
        sandbox_root=str(fixture_root),
    )
    wrapped_file_write = capture.wrap_tool(FILE_WRITE_SPEC)

    normalized: dict[str, str] = {}
    for rel_path, content in generated_files.items():
        output_path = validated_relative_test_path(fixture_root, rel_path)
        normalized_rel = output_path.relative_to(fixture_root.resolve()).as_posix()
        wrapped_file_write.handler(
            {
                "path": normalized_rel,
                "content": content,
                "create_dirs": True,
                "sandbox_root": str(fixture_root),
            }
        )
        normalized[normalized_rel] = content

    lifecycle_changeset = capture.to_spec_changeset()
    store = InMemoryChangeSetStore()
    execution_context = PlanExecutionContext(
        plan_id=plan_id,
        automation_profile="trusted",
        changeset_store=store,
    )
    changeset_id = execution_context.start_changeset()
    for entry in lifecycle_changeset.entries:
        operation = (
            ChangeOperation(entry.operation)
            if isinstance(entry.operation, str)
            else ChangeOperation(entry.operation.value)
        )
        execution_context.record_change(
            ChangeEntry(
                plan_id=plan_id,
                resource_id=entry.resource_id,
                tool_name=entry.tool_name,
                operation=operation,
                path=entry.path,
                before_hash=entry.before_hash,
                after_hash=entry.after_hash,
                timestamp=entry.timestamp,
            )
        )

    return changeset_id, store


def wf02_mocked_generation_artifacts() -> None:
    """Verify lifecycle-produced artifacts include provider-derived outputs."""
    lifecycle = setup_lifecycle()
    create_wf02_action(lifecycle)
    plan_id = create_trusted_plan(lifecycle)
    lifecycle.try_auto_run(plan_id)

    with auth_module_fixture() as (fixture_root, baseline_coverage):
        assert (fixture_root / "src" / "auth.py").is_file()
        assert baseline_coverage < 80.0

        generated_files = generate_tests_from_mock_provider(
            fixture_root,
            "src/auth",
            80,
            include_coverage_suite=True,
        )
        assert "tests/test_provider_mock_output.py" in generated_files

        changeset_id, store = _capture_changeset_from_file_lifecycle(
            plan_id,
            fixture_root,
            generated_files,
        )

        plan = lifecycle.get_plan(plan_id)
        plan.changeset_id = changeset_id
        plan.sandbox_refs = [f"sandbox-{plan_id[:8]}"]
        lifecycle.save_plan(plan)

        assert_invariants_and_conventions(
            fixture_root,
            plan_id,
            lifecycle,
            generated_files,
        )

        apply_service = PlanApplyService(
            lifecycle_service=lifecycle, changeset_store=store
        )
        capture = StringIO()
        old_stdout = sys.stdout
        sys.stdout = capture
        try:
            artifact_payload = apply_service.artifacts(plan_id, fmt="json")
        finally:
            sys.stdout = old_stdout

        rendered = artifact_payload or capture.getvalue().strip()
        parsed = json.loads(rendered)
        assert parsed["changeset_id"] == changeset_id
        raw_files = parsed["files_changed"]
        changed_files = {
            item["path"] if isinstance(item, dict) else str(item) for item in raw_files
        }
        expected_files = set(generated_files)
        assert len(changed_files) == len(expected_files), (
            "Artifact count mismatch: expected "
            f"{len(expected_files)} files, got {len(changed_files)}"
        )
        assert changed_files == expected_files, (
            "Artifact membership mismatch: expected "
            f"{sorted(expected_files)}, got {sorted(changed_files)}"
        )
        assert all(path.startswith("tests/") for path in changed_files), (
            f"Expected test-only artifacts, got {sorted(changed_files)}"
        )
        assert all(not path.startswith("src/") for path in changed_files), (
            "Invariant violated: production source path detected in artifacts"
        )

        persisted_changeset = store.get(changeset_id)
        assert persisted_changeset is not None
        persisted_paths = {entry.path for entry in persisted_changeset.entries}
        assert len(persisted_paths) == len(expected_files), (
            "Persisted artifact count mismatch: expected "
            f"{len(expected_files)} files, got {len(persisted_paths)}"
        )
        assert persisted_paths == expected_files, (
            "Persisted artifact membership mismatch: expected "
            f"{sorted(expected_files)}, got {sorted(persisted_paths)}"
        )

        facade = A2aLocalFacade(
            services={"plan_lifecycle_service": lifecycle},
        )
        artifacts_response = facade.dispatch(
            A2aRequest(
                method="_cleveragents/plan/artifacts",
                params={"plan_id": plan_id},
            )
        )
        assert artifacts_response.result is not None, artifacts_response.error
        response_data = artifacts_response.result
        assert response_data.get("plan_id") == plan_id
        assert response_data.get("changeset_id") == changeset_id
        sandbox_refs = response_data.get("sandbox_refs")
        assert isinstance(sandbox_refs, list)
        assert sandbox_refs, "Expected non-empty sandbox refs in artifacts response"

    print("wf02-mocked-generation-artifacts-ok")