cleveragents-core/robot/wf02_test_generation_validation.py

"""WF02 coverage validation lifecycle assertions."""

from __future__ import annotations

import importlib
import importlib.util
import inspect
import json
import sys
import trace
import traceback
import uuid
from collections.abc import Callable
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from wf02_test_generation_common import (
    auth_module_fixture,
    create_trusted_plan,
    create_wf02_action,
    generate_tests_from_mock_provider,
    setup_lifecycle,
    write_generated_files_safely,
)
from wf02_test_generation_lifecycle import (
    record_files_via_execution_context,
)

from cleveragents.application.services.plan_apply_service import (
    ApplyOutcome,
    PlanApplyService,
)
from cleveragents.application.services.resource_registry_service import (
    ResourceRegistryService,
)
from cleveragents.application.services.tool_registry_service import ToolRegistryService
from cleveragents.application.services.validation_pipeline import (
    ValidationCommand,
    ValidationPipeline,
)
from cleveragents.domain.models.core.change import InMemoryChangeSetStore
from cleveragents.domain.models.core.tool import ValidationMode
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
    ToolRegistryRepository,
    ValidationAttachmentRepository,
)


class _NoCloseSession:
    """Thin proxy that delegates every attribute to a real SQLAlchemy Session
    but suppresses ``close()``.

    Repository helpers call ``session.close()`` after each operation.  In
    integration tests we share a single in-memory SQLite session across
    multiple repositories so that all writes are visible without a second
    connection.  Closing the session would invalidate the connection and
    lose all in-memory data.

    ``object.__setattr__`` / ``object.__getattribute__`` are used in the
    dunder overrides to access this wrapper's own ``_s`` slot without
    triggering the proxy's ``__getattr__`` / ``__setattr__``, which would
    recurse into the wrapped session instead.
    """

    def __init__(self, session: object) -> None:
        object.__setattr__(self, "_s", session)

    def close(self) -> None:
        pass

    def __setattr__(self, name: str, value: object) -> None:
        setattr(object.__getattribute__(self, "_s"), name, value)

    def __getattr__(self, name: str) -> object:
        return getattr(object.__getattribute__(self, "_s"), name)


def _setup_validation_db() -> tuple[Callable[[], _NoCloseSession], Session]:
    """Create in-memory DB and return (session_factory, session)."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine, expire_on_commit=False)()
    wrapper = _NoCloseSession(session)

    def factory() -> _NoCloseSession:
        return wrapper

    return factory, session


def _assert_validation_payload(
    validation_summary: dict[str, Any],
    *,
    expected_passed: bool,
    target: int,
) -> None:
    """Assert validation summary payload semantics for WF02 coverage gate."""
    assert validation_summary["total"] == 1
    assert validation_summary["informational_passed"] == 0
    assert validation_summary["informational_failed"] == 0

    if expected_passed:
        assert validation_summary["required_passed"] == 1
        assert validation_summary["required_failed"] == 0
        assert validation_summary["all_required_passed"] is True
    else:
        assert validation_summary["required_passed"] == 0
        assert validation_summary["required_failed"] == 1
        assert validation_summary["all_required_passed"] is False

    results = validation_summary.get("results")
    assert isinstance(results, list) and len(results) == 1
    result = results[0]
    assert result["validation_name"] == "local/auth-coverage"
    assert result["resource_name"] == "local/api-service-repo"
    assert result["mode"] == "required"
    assert result["passed"] is expected_passed

    data = result.get("data")
    assert isinstance(data, dict)
    measured = data.get("measured_coverage")
    assert isinstance(measured, float)
    assert data.get("target") == target

    if expected_passed:
        assert measured >= float(target)
    else:
        assert measured < float(target)


def _measure_auth_coverage_with_trace(
    fixture_root: Path,
) -> tuple[float, str, str, int]:
    """Measure auth.py line coverage by tracing test function execution."""
    import os
    from contextlib import redirect_stderr, redirect_stdout
    from io import StringIO

    target_file = (fixture_root / "src" / "auth.py").resolve()
    source_lines = target_file.read_text(encoding="utf-8").splitlines()
    executable_lines = {
        line_no
        for line_no, text in enumerate(source_lines, start=1)
        if text.strip() and not text.strip().startswith("#")
    }
    if not executable_lines:
        return 0.0, "", "No executable lines found in auth.py", 3

    tracer = trace.Trace(count=True, trace=False)
    old_cwd = Path.cwd()
    old_sys_path = list(sys.path)
    stdout_capture = StringIO()
    stderr_capture = StringIO()
    try:
        os.chdir(fixture_root)
        sys.path.insert(0, str(fixture_root))
        importlib.invalidate_caches()
        with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
            for test_file in sorted((fixture_root / "tests").glob("test_*.py")):
                if test_file.suffix != ".py":
                    continue
                if not test_file.name.startswith("test_auth"):
                    continue

                module_name = f"_wf02_test_{test_file.stem}_{uuid.uuid4().hex}"
                spec = importlib.util.spec_from_file_location(module_name, test_file)
                if spec is None or spec.loader is None:
                    return (
                        0.0,
                        stdout_capture.getvalue(),
                        f"Unable to load {test_file}",
                        4,
                    )
                module = importlib.util.module_from_spec(spec)
                spec.loader.exec_module(module)

                fixture_values: dict[str, object] = {
                    "auth_credentials": ("admin", "secret"),
                }

                for attr_name in dir(module):
                    if not attr_name.startswith("test_"):
                        continue
                    maybe_test = getattr(module, attr_name)
                    if not callable(maybe_test):
                        continue

                    signature = inspect.signature(maybe_test)
                    kwargs: dict[str, object] = {}
                    for param_name, param in signature.parameters.items():
                        if param.kind not in (
                            inspect.Parameter.POSITIONAL_OR_KEYWORD,
                            inspect.Parameter.KEYWORD_ONLY,
                        ):
                            raise AssertionError(
                                f"Unsupported test function parameter kind for "
                                f"{attr_name}: {param.kind}"
                            )

                        if param_name in fixture_values:
                            kwargs[param_name] = fixture_values[param_name]
                            continue

                        if param.default is inspect.Parameter.empty:
                            raise AssertionError(
                                "Unsupported required test fixture parameter "
                                f"'{param_name}' in {attr_name}"
                            )

                    tracer.runfunc(maybe_test, **kwargs)

                for loaded_name in list(sys.modules):
                    if loaded_name == module_name or loaded_name.startswith(
                        "_wf02_test_"
                    ):
                        sys.modules.pop(loaded_name, None)
                for cached_name in ("src.auth", "src"):
                    sys.modules.pop(cached_name, None)
                importlib.invalidate_caches()
    except Exception:
        return 0.0, stdout_capture.getvalue(), traceback.format_exc(), 1
    finally:
        sys.path = old_sys_path
        os.chdir(old_cwd)

    counts = tracer.results().counts
    covered_lines = {
        lineno
        for (filename, lineno), count in counts.items()
        if count > 0
        and Path(filename).name == "auth.py"
        and "src" in Path(filename).parts
    }
    measured_pct = (
        len(covered_lines & executable_lines) / len(executable_lines)
    ) * 100.0
    return measured_pct, stdout_capture.getvalue(), stderr_capture.getvalue(), 0


def _register_attach_and_run_validation(
    fixture_root: Path,
    plan_id: str,
    target: int,
) -> dict[str, Any]:
    """Run validation through registration/attachment/execution lifecycle."""
    session_factory, _ = _setup_validation_db()
    tool_repo = ToolRegistryRepository(session_factory=session_factory)
    attachment_repo = ValidationAttachmentRepository(session_factory=session_factory)
    tool_service = ToolRegistryService(
        tool_repo=tool_repo,
        attachment_repo=attachment_repo,
    )
    now = datetime.now(tz=UTC).isoformat()
    tool_service.register_tool(
        {
            "name": "local/auth-coverage",
            "description": "Auth module coverage gate",
            "tool_type": "validation",
            "source": "builtin",
            "timeout": 300,
            "created_at": now,
            "updated_at": now,
            "resource_bindings": [],
            "mode": "required",
        }
    )

    resource_service = ResourceRegistryService(session_factory=session_factory)
    resource_service.bootstrap_builtin_types()
    registered_resource = resource_service.register_resource(
        type_name="git-checkout",
        name="local/api-service-repo",
        location=str(fixture_root),
        description="WF02 coverage fixture repository",
    )

    resource_id = registered_resource.resource_id
    tool_service.attach_validation(
        validation_name="local/auth-coverage",
        resource_id=resource_id,
        plan_id=plan_id,
        args={"target": target},
    )

    attachments = tool_service.list_validations_for_resource(
        resource_id=resource_id,
        plan_id=plan_id,
    )
    assert attachments, "Expected at least one attached validation"

    commands: list[ValidationCommand] = []
    for attachment in attachments:
        args_json = attachment.get("args_json")
        args = json.loads(args_json) if isinstance(args_json, str) and args_json else {}
        mode_text = str(attachment.get("mode", "required"))
        mode = (
            ValidationMode.REQUIRED
            if mode_text == "required"
            else ValidationMode.INFORMATIONAL
        )
        commands.append(
            ValidationCommand(
                validation_name=str(attachment["validation_name"]),
                resource_id=str(attachment["resource_id"]),
                resource_name="local/api-service-repo",
                mode=mode,
                arguments=args,
                timeout_seconds=30.0,
            )
        )

    def executor(
        validation_name: str,
        arguments: dict[str, object],
    ) -> dict[str, object]:
        target_value = arguments.get("target")
        if not isinstance(target_value, int):
            raise TypeError("target must be an int")
        measured_pct, stdout, stderr, exit_code = _measure_auth_coverage_with_trace(
            fixture_root
        )
        return {
            "passed": exit_code == 0 and measured_pct >= float(target_value),
            "message": (
                f"{validation_name}: measured={measured_pct:.1f}% "
                f"target={target_value}% exit_code={exit_code}"
            ),
            "data": {
                "stdout": stdout,
                "stderr": stderr,
                "measured_coverage": measured_pct,
                "target": target_value,
            },
        }

    plan_metadata: dict[str, Any] = {}
    summary = ValidationPipeline(
        commands=commands,
        executor=executor,
        max_workers=1,
    ).run_for_plan(plan_metadata=plan_metadata)
    assert summary.results, "Expected at least one validation result"

    validation_summary = plan_metadata.get("validation_summary")
    assert isinstance(validation_summary, dict), "Expected validation_summary metadata"
    return validation_summary


def wf02_coverage_validation() -> None:
    """Verify coverage gate blocks failing coverage then applies passing coverage."""
    target = 80

    lifecycle_fail = setup_lifecycle()
    create_wf02_action(lifecycle_fail)
    fail_plan_id = create_trusted_plan(lifecycle_fail)
    lifecycle_fail.try_auto_run(fail_plan_id)
    store_fail = InMemoryChangeSetStore()
    apply_fail = PlanApplyService(
        lifecycle_service=lifecycle_fail, changeset_store=store_fail
    )

    with auth_module_fixture() as (fail_fixture_root, _):
        placeholder_path = "tests/test_auth_placeholder.py"
        (fail_fixture_root / placeholder_path).write_text(
            """def test_placeholder() -> None:
    assert True
""",
            encoding="utf-8",
        )
        fail_changeset_id = record_files_via_execution_context(
            plan_id=fail_plan_id,
            store=store_fail,
            files={placeholder_path: "placeholder"},
        )
        fail_plan = lifecycle_fail.get_plan(fail_plan_id)
        fail_plan.changeset_id = fail_changeset_id
        fail_plan.sandbox_refs = [f"sandbox-{fail_plan_id[:8]}"]
        fail_plan.validation_summary = _register_attach_and_run_validation(
            fail_fixture_root,
            fail_plan_id,
            target,
        )
        _assert_validation_payload(
            fail_plan.validation_summary,
            expected_passed=False,
            target=target,
        )
        lifecycle_fail.save_plan(fail_plan)

        lifecycle_fail.apply_plan(fail_plan_id)
        lifecycle_fail.start_apply(fail_plan_id)
        fail_result = apply_fail.apply_with_validation_gate(fail_plan_id)
        assert fail_result.outcome == ApplyOutcome.CONSTRAINED

        constrained_plan = lifecycle_fail.get_plan(fail_plan_id)
        assert constrained_plan.processing_state.value == "constrained"

    lifecycle_pass = setup_lifecycle()
    create_wf02_action(lifecycle_pass)
    pass_plan_id = create_trusted_plan(lifecycle_pass)
    lifecycle_pass.try_auto_run(pass_plan_id)
    store_pass = InMemoryChangeSetStore()
    apply_pass = PlanApplyService(
        lifecycle_service=lifecycle_pass, changeset_store=store_pass
    )

    with auth_module_fixture() as (pass_fixture_root, _):
        generated_files = generate_tests_from_mock_provider(
            pass_fixture_root,
            "src/auth",
            target,
            include_coverage_suite=True,
        )
        normalized_files = write_generated_files_safely(
            pass_fixture_root,
            generated_files,
        )
        pass_changeset_id = record_files_via_execution_context(
            plan_id=pass_plan_id,
            store=store_pass,
            files=normalized_files,
        )

        pass_plan = lifecycle_pass.get_plan(pass_plan_id)
        pass_plan.changeset_id = pass_changeset_id
        pass_plan.sandbox_refs = [f"sandbox-{pass_plan_id[:8]}"]
        pass_plan.validation_summary = _register_attach_and_run_validation(
            pass_fixture_root,
            pass_plan_id,
            target,
        )
        _assert_validation_payload(
            pass_plan.validation_summary,
            expected_passed=True,
            target=target,
        )
        lifecycle_pass.save_plan(pass_plan)

        lifecycle_pass.apply_plan(pass_plan_id)
        lifecycle_pass.start_apply(pass_plan_id)
        pass_result = apply_pass.apply_with_validation_gate(pass_plan_id)
        assert pass_result.outcome == ApplyOutcome.APPLIED

        applied_plan = lifecycle_pass.get_plan(pass_plan_id)
        assert applied_plan.processing_state.value == "applied"

    print("wf02-coverage-validation-ok")