Files
cleveragents-core/robot/wf02_test_generation_common.py
HAL9000 de1fd49594 fix(plan-lifecycle): fix integration test regressions from DoD gating
Update integration test helpers to use definition_of_done values that
satisfy the new TextMatchEvaluator gate (criterion text must appear as
a substring of a plan argument key or value).  Also fix _evaluate_dod
to merge DoD metadata into plan.validation_summary without overwriting
Execute-phase validation counts, preserving the coverage/tool-validation
gate used by apply_with_validation_gate.  Update CONTRIBUTORS.md.

Fixes: wf02, wf04, wf06, wf07 integration test suites.

ISSUES CLOSED: #7927
2026-06-02 05:14:43 -04:00

308 lines
11 KiB
Python

"""Shared WF02 integration helpers.
This module contains reusable setup and fixture helpers used by the
WF02 integration command implementations.
"""
from __future__ import annotations
import os
import tempfile
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from cleveragents.application.container import get_ai_provider
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.action import ActionArgument
from cleveragents.domain.models.core.context import Context, ContextType
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
ProjectLink,
)
from cleveragents.domain.models.core.plan_legacy import Plan, PlanStatus
from cleveragents.domain.models.core.project_legacy import Project
from cleveragents.providers.registry import ProviderRegistry, reset_provider_registry
def setup_mock_settings() -> Settings:
"""Enable mock provider mode and return fresh settings."""
os.environ["CLEVERAGENTS_MOCK_PROVIDERS"] = "true"
os.environ["CLEVERAGENTS_ENV"] = "test"
Settings._instance = None
return Settings()
def setup_lifecycle() -> PlanLifecycleService:
"""Create lifecycle service configured for integration tests."""
settings = setup_mock_settings()
return PlanLifecycleService(settings=settings)
def create_wf02_action(lifecycle: PlanLifecycleService) -> None:
"""Create the Workflow Example 2 action definition."""
lifecycle.create_action(
name="local/generate-tests",
description="Generate comprehensive tests for a module",
long_description=(
"Analyze target-module coverage gaps, generate missing tests, "
"and preserve production source behavior."
),
definition_of_done="target_module",
strategy_actor="anthropic/claude-3.5-sonnet",
execution_actor="anthropic/claude-3.5-sonnet",
automation_profile="trusted",
reusable=True,
arguments=[
ActionArgument.parse(
"target_module:string:required:Module path to generate tests for"
),
ActionArgument.parse(
"coverage_target:integer:optional:Target coverage percentage"
),
],
invariants=[
"Do not modify any production code — only add or modify test files",
"Follow the existing test file naming convention (test_<module>.py)",
"Use the project's existing test fixtures and conftest.py patterns",
],
)
def create_trusted_plan(lifecycle: PlanLifecycleService) -> str:
"""Create and persist a trusted-profile plan for WF02."""
plan = lifecycle.use_action(
action_name="local/generate-tests",
project_links=[ProjectLink(project_name="local/api-service")],
arguments={"target_module": "src/auth", "coverage_target": 80},
created_by="wf02-integration-test",
)
plan.automation_profile = AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
)
lifecycle.save_plan(plan)
return plan.identity.plan_id
@contextmanager
def auth_module_fixture() -> Iterator[tuple[Path, float]]:
"""Yield temporary fixture repository with low auth-module coverage."""
with tempfile.TemporaryDirectory(prefix="wf02-auth-fixture-") as tmpdir:
fixture_root = Path(tmpdir)
src_dir = fixture_root / "src"
tests_dir = fixture_root / "tests"
src_dir.mkdir(parents=True, exist_ok=True)
tests_dir.mkdir(parents=True, exist_ok=True)
(src_dir / "auth.py").write_text(
"""def authenticate_user(username: str, password: str) -> bool:
if not username:
return False
if password == \"secret\" and username == \"admin\":
return True
if password == \"\":
return False
return False
""",
encoding="utf-8",
)
(src_dir / "__init__.py").write_text("", encoding="utf-8")
(tests_dir / "test_auth_smoke.py").write_text(
"""from src.auth import authenticate_user
def test_authenticate_user_valid() -> None:
assert authenticate_user(\"admin\", \"secret\") is True
""",
encoding="utf-8",
)
(tests_dir / "conftest.py").write_text(
"""import pytest
@pytest.fixture
def auth_credentials() -> tuple[str, str]:
return ("admin", "secret")
""",
encoding="utf-8",
)
baseline_coverage = 45.0
yield fixture_root, baseline_coverage
def validated_relative_test_path(fixture_root: Path, rel_path: str) -> Path:
"""Resolve and validate a generated relative path under fixture/tests."""
candidate = Path(rel_path)
if candidate.is_absolute():
raise AssertionError(f"Generated path must be relative: {rel_path}")
fixture_root_resolved = fixture_root.resolve()
resolved = (fixture_root_resolved / candidate).resolve()
try:
relative_to_root = resolved.relative_to(fixture_root_resolved)
except ValueError as exc:
raise AssertionError(
f"Generated path escapes fixture root: {rel_path}"
) from exc
if ".." in candidate.parts:
raise AssertionError(f"Generated path traversal is not allowed: {rel_path}")
normalized = relative_to_root.as_posix()
if not normalized.startswith("tests/"):
raise AssertionError(f"Generated path must stay in tests/: {rel_path}")
return resolved
def write_generated_files_safely(
fixture_root: Path,
generated_files: dict[str, str],
) -> dict[str, str]:
"""Write generated files to fixture with traversal and scope guards."""
normalized_files: dict[str, str] = {}
for rel_path, content in generated_files.items():
output_path = validated_relative_test_path(fixture_root, rel_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(content, encoding="utf-8")
normalized_rel = output_path.relative_to(fixture_root.resolve()).as_posix()
normalized_files[normalized_rel] = content
return normalized_files
def assert_invariants_and_conventions(
fixture_root: Path,
plan_id: str,
lifecycle: PlanLifecycleService,
generated_files: dict[str, str],
) -> None:
"""Assert WF02 invariant declarations and generated-file conventions."""
plan = lifecycle.get_plan(plan_id)
invariant_texts = {inv.text for inv in plan.invariants}
assert any(
"Do not modify any production code" in text for text in invariant_texts
), "Expected production-code invariant on plan"
assert any("test file naming convention" in text for text in invariant_texts), (
"Expected naming-convention invariant on plan"
)
assert any(
"test fixtures and conftest.py patterns" in text for text in invariant_texts
), "Expected fixture/conftest invariant on plan"
assert (fixture_root / "tests" / "conftest.py").is_file(), (
"Expected fixture project to include tests/conftest.py"
)
generated_paths = sorted(generated_files)
assert all(path.startswith("tests/") for path in generated_paths), (
f"Expected only tests/ paths, got {generated_paths}"
)
assert all(path.endswith(".py") for path in generated_paths), (
f"Expected python test artifacts, got {generated_paths}"
)
assert all(Path(path).name.startswith("test_") for path in generated_paths), (
f"Expected test_<module>.py naming, got {generated_paths}"
)
def generate_tests_from_mock_provider(
fixture_root: Path,
target_module: str,
coverage_target: int,
include_coverage_suite: bool,
) -> dict[str, str]:
"""Generate provider-derived test files for WF02 scenarios."""
settings = setup_mock_settings()
reset_provider_registry()
registry = ProviderRegistry(settings=settings)
provider = get_ai_provider(settings=settings, provider_registry=registry)
assert provider is not None, "Expected resolved provider in mock mode"
project = Project(id=1, name="wf02-fixture", path=fixture_root)
plan = Plan(
id=1,
project_id=1,
name="wf02-generate-tests",
prompt=(
f"Generate tests for {target_module} with target coverage "
f"{coverage_target}%"
),
status=PlanStatus.PENDING,
current=True,
build=None,
build_started_at=None,
build_completed_at=None,
model_used=None,
token_count=None,
result=None,
applied_at=None,
files_created=None,
files_modified=None,
files_deleted=None,
)
auth_path = (fixture_root / "src" / "auth.py").resolve()
context = Context(
id=None,
plan_id=1,
type=ContextType.FILE,
path=str(auth_path),
content=auth_path.read_text(encoding="utf-8"),
file_hash=None,
size=0,
)
provider_response = provider.generate_changes(
project=project,
plan=plan,
contexts=[context],
)
assert provider_response.error_message is None, provider_response.error_message
assert provider_response.model_used.startswith("mock"), provider_response.model_used
assert provider_response.token_count >= 0
assert len(provider_response.changes) == 1, (
"Expected a single mocked provider change for test-generation prompt"
)
provider_change = provider_response.changes[0]
provider_content = provider_change.new_content
assert isinstance(provider_content, str) and provider_content, (
"Expected provider change new_content"
)
generated: dict[str, str] = {
"tests/test_provider_mock_output.py": provider_content,
}
if include_coverage_suite:
generated["tests/test_auth_generated.py"] = (
f'"""Provider-augmented tests for {target_module}."""\n\n'
"from src.auth import authenticate_user\n\n"
"def test_authenticate_user_accepts_valid_credentials(\n"
" auth_credentials: tuple[str, str],\n"
") -> None:\n"
" username, password = auth_credentials\n"
" assert authenticate_user(username, password) is True\n"
)
generated["tests/test_auth_edge_cases.py"] = (
f'"""Edge tests targeting {coverage_target}% coverage."""\n\n'
"from src.auth import authenticate_user\n\n"
"def test_authenticate_user_rejects_blank_password(\n"
" auth_credentials: tuple[str, str],\n"
") -> None:\n"
" username, _ = auth_credentials\n"
' assert authenticate_user(username, "") is False\n\n'
"def test_authenticate_user_rejects_blank_username() -> None:\n"
' assert authenticate_user("", "secret") is False\n\n'
"def test_authenticate_user_rejects_wrong_password() -> None:\n"
' assert authenticate_user("admin", "wrong") is False\n'
)
return generated