"""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")