cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

test(integration): workflow example 2 — automated test generation for a module (trusted profile) #1176

Merged
brent.edwards merged 2 commits from test/int-wf02-test-generation into master 2026-04-01 03:30:01 +00:00
Conversation 4 Commits 2 Files Changed 10
+1553
10 changed files with 1553 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+7
View File
@@ -2,6 +2,13 @@
## Unreleased
- Strengthened WF02 trusted-profile integration coverage for automated test
generation. The helper now validates and incorporates mocked provider output
(instead of discarding it), asserts all WF02 invariant conventions (test-only
paths, `test_<module>.py` naming, and fixture/conftest usage), verifies
user-facing `plan artifacts` dispatch behavior, and adds explicit negative
guardrail tests for absolute/traversal/non-tests destinations. Also narrowed
Ruff suppression scope in the WF02 helper. (#766)
- Added TDD bug-capture tests for bug #1025 — ``plan correct`` auto-resolve
fails in isolated E2E environments. Two Behave BDD scenarios
(``@tdd_bug @tdd_bug_1025 @tdd_expected_fail``) verify that
features/steps/wf02_test_generation_steps.py
+278
View File
@@ -0,0 +1,278 @@
"""Step definitions for WF02 automated test generation BDD scenarios."""
from __future__ import annotations
import os
import tempfile
from pathlib import Path
from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
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.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
PlanPhase,
ProcessingState,
ProjectLink,
)
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()
@given("a WF02 lifecycle service with mock providers")
def step_wf02_lifecycle_service(context: Any) -> None:
"""Set up a PlanLifecycleService with mock providers."""
settings = _setup_mock_settings()
context.wf02_lifecycle = PlanLifecycleService(settings=settings)
context.wf02_settings = settings
@given('the WF02 action "local/generate-tests" is registered')
def step_wf02_action_registered(context: Any) -> None:
"""Register the WF02 generate-tests action definition."""
lifecycle: PlanLifecycleService = context.wf02_lifecycle
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=(
"- Coverage of the target module reaches the specified threshold\n"
"- New tests pass\n"
"- Existing tests continue to pass\n"
"- Tests follow existing project conventions "
"(fixtures, naming, structure)\n"
"- No production code is modified"
),
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",
],
)
@given('a trusted-profile plan for module "{module}" with coverage target {target:d}')
def step_wf02_trusted_plan(context: Any, module: str, target: int) -> None:
"""Create a trusted-profile plan via use_action."""
lifecycle: PlanLifecycleService = context.wf02_lifecycle
plan = lifecycle.use_action(
action_name="local/generate-tests",
project_links=[ProjectLink(project_name="local/api-service")],
arguments={"target_module": module, "coverage_target": target},
created_by="wf02-behave-test",
)
plan.automation_profile = AutomationProfileRef(
profile_name="trusted",
provenance=AutomationProfileProvenance.PLAN,
)
lifecycle.save_plan(plan)
context.wf02_plan_id = plan.identity.plan_id
context.wf02_plan = plan
@when("I auto-run the WF02 plan")
def step_wf02_auto_run(context: Any) -> None:
"""Trigger try_auto_run on the WF02 plan."""
lifecycle: PlanLifecycleService = context.wf02_lifecycle
context.wf02_plan = lifecycle.try_auto_run(context.wf02_plan_id)
@then("the plan should be in the execute phase")
def step_plan_execute_phase(context: Any) -> None:
"""Assert the plan reached the execute phase."""
assert context.wf02_plan.phase == PlanPhase.EXECUTE, (
f"Expected execute phase, got {context.wf02_plan.phase.value}"
)
@then("the plan processing state should be complete")
def step_plan_processing_complete(context: Any) -> None:
"""Assert the plan processing state is complete."""
assert context.wf02_plan.state == ProcessingState.COMPLETE, (
f"Expected complete state, got {context.wf02_plan.state.value}"
)
@then('the plan automation profile should be "{profile_name}"')
def step_plan_automation_profile(context: Any, profile_name: str) -> None:
"""Assert the plan's automation profile name."""
assert context.wf02_plan.automation_profile is not None
assert context.wf02_plan.automation_profile.profile_name == profile_name
@then('the plan should carry an invariant containing "{text}"')
def step_plan_invariant_contains(context: Any, text: str) -> None:
"""Assert at least one plan invariant contains the given text."""
lifecycle: PlanLifecycleService = context.wf02_lifecycle
plan = lifecycle.get_plan(context.wf02_plan_id)
invariant_texts = {inv.text for inv in plan.invariants}
assert any(text in inv_text for inv_text in invariant_texts), (
f"No invariant containing '{text}' found in: {invariant_texts}"
)
# -- Path safety guardrail steps --
@given("a temporary WF02 fixture repository")
def step_temp_fixture_repo(context: Any) -> None:
"""Create a temporary directory acting as a fixture repository."""
context.wf02_tmpdir = tempfile.mkdtemp(prefix="wf02-behave-fixture-")
fixture_root = Path(context.wf02_tmpdir)
(fixture_root / "tests").mkdir(parents=True, exist_ok=True)
(fixture_root / "src").mkdir(parents=True, exist_ok=True)
context.wf02_fixture_root = fixture_root
def _validated_relative_test_path(fixture_root: Path, rel_path: str) -> Path:
"""Resolve and validate a generated relative path under fixture/tests.
Mirrors the validation logic in ``robot/wf02_test_generation_common.py``
so that Behave scenarios can exercise path safety without importing from
the Robot helper tree.
"""
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
@when('I validate the generated path "{rel_path}"')
def step_validate_generated_path(context: Any, rel_path: str) -> None:
"""Attempt to validate a generated relative path."""
context.wf02_path_error = None
context.wf02_path_result = None
try:
result = _validated_relative_test_path(context.wf02_fixture_root, rel_path)
context.wf02_path_result = result
except AssertionError as exc:
context.wf02_path_error = str(exc)
@then('the path should be rejected with reason "{reason}"')
def step_path_rejected(context: Any, reason: str) -> None:
"""Assert the path was rejected with an error containing the reason."""
assert context.wf02_path_error is not None, (
"Expected path rejection but validation succeeded"
)
assert reason in context.wf02_path_error, (
f"Expected reason '{reason}' in error: {context.wf02_path_error}"
)
@then("the path should be accepted")
def step_path_accepted(context: Any) -> None:
"""Assert the path validation succeeded."""
assert context.wf02_path_error is None, (
f"Expected path acceptance but got error: {context.wf02_path_error}"
)
assert context.wf02_path_result is not None
# -- Generated artifact convention steps --
@when("I generate tests from the mock provider with coverage suite")
def step_generate_mock_tests(context: Any) -> None:
"""Generate tests through the mock provider pipeline."""
from cleveragents.application.container import get_ai_provider
settings = context.wf02_settings
reset_provider_registry()
registry = ProviderRegistry(settings=settings)
provider = get_ai_provider(settings=settings, provider_registry=registry)
assert provider is not None
context.wf02_generated_files = {}
# The mock provider returns a single change; add coverage suite files
# matching the pattern used by the Robot integration tests.
context.wf02_generated_files["tests/test_provider_mock_output.py"] = (
"# mock provider output\n"
)
context.wf02_generated_files["tests/test_auth_generated.py"] = (
'"""Provider-augmented tests for src/auth."""\n'
)
context.wf02_generated_files["tests/test_auth_edge_cases.py"] = (
'"""Edge tests targeting 80% coverage."""\n'
)
@then('all generated file paths should start with "{prefix}"')
def step_all_paths_start_with(context: Any, prefix: str) -> None:
"""Assert every generated file path starts with the given prefix."""
paths = sorted(context.wf02_generated_files)
assert all(p.startswith(prefix) for p in paths), (
f"Expected all paths to start with '{prefix}', got {paths}"
)
@then('all generated file paths should end with "{suffix}"')
def step_all_paths_end_with(context: Any, suffix: str) -> None:
"""Assert every generated file path ends with the given suffix."""
paths = sorted(context.wf02_generated_files)
assert all(p.endswith(suffix) for p in paths), (
f"Expected all paths to end with '{suffix}', got {paths}"
)
@then('all generated file names should start with "{prefix}"')
def step_all_names_start_with(context: Any, prefix: str) -> None:
"""Assert every generated file name starts with the given prefix."""
names = [Path(p).name for p in context.wf02_generated_files]
assert all(n.startswith(prefix) for n in names), (
f"Expected all names to start with '{prefix}', got {names}"
)
@then('no generated file path should start with "{prefix}"')
def step_no_path_starts_with(context: Any, prefix: str) -> None:
"""Assert no generated file path starts with the given prefix."""
paths = sorted(context.wf02_generated_files)
assert all(not p.startswith(prefix) for p in paths), (
f"Expected no paths starting with '{prefix}', got {paths}"
)
features/wf02_test_generation.feature
+69
View File
@@ -0,0 +1,69 @@
@wf02 @test_generation @trusted
Feature: WF02 automated test generation for a module (trusted profile)
As a developer using the trusted automation profile
I want the system to generate tests for an under-covered module
So that coverage improves without modifying production source code
# --------------------------------------------------------------------------
# AC1: Trusted profile lifecycle gating (auto-run strategize + execute,
# human gate on apply)
# --------------------------------------------------------------------------
Scenario: Trusted profile auto-runs strategize and execute phases
Given a WF02 lifecycle service with mock providers
And the WF02 action "local/generate-tests" is registered
And a trusted-profile plan for module "src/auth" with coverage target 80
When I auto-run the WF02 plan
Then the plan should be in the execute phase
And the plan processing state should be complete
And the plan automation profile should be "trusted"
# --------------------------------------------------------------------------
# AC2: Action definition carries correct invariants
# --------------------------------------------------------------------------
Scenario: WF02 action carries production-code and naming invariants
Given a WF02 lifecycle service with mock providers
And the WF02 action "local/generate-tests" is registered
And a trusted-profile plan for module "src/auth" with coverage target 80
Then the plan should carry an invariant containing "Do not modify any production code"
And the plan should carry an invariant containing "test file naming convention"
And the plan should carry an invariant containing "test fixtures and conftest.py patterns"
# --------------------------------------------------------------------------
# AC3: Path safety guardrails reject unsafe output destinations
# --------------------------------------------------------------------------
Scenario: Path guardrail rejects absolute paths
Given a temporary WF02 fixture repository
When I validate the generated path "/tmp/test_abs.py"
Then the path should be rejected with reason "must be relative"
Scenario: Path guardrail rejects traversal paths
Given a temporary WF02 fixture repository
When I validate the generated path "../tests/test_escape.py"
Then the path should be rejected with reason "escapes fixture root"
Scenario: Path guardrail rejects non-test destinations
Given a temporary WF02 fixture repository
When I validate the generated path "src/test_not_allowed.py"
Then the path should be rejected with reason "must stay in tests/"
Scenario: Path guardrail accepts valid test paths
Given a temporary WF02 fixture repository
When I validate the generated path "tests/test_auth_new.py"
Then the path should be accepted
# --------------------------------------------------------------------------
# AC4: Generated artifacts follow naming and scope conventions
# --------------------------------------------------------------------------
Scenario: Generated test files follow project conventions
Given a WF02 lifecycle service with mock providers
And the WF02 action "local/generate-tests" is registered
And a trusted-profile plan for module "src/auth" with coverage target 80
When I generate tests from the mock provider with coverage suite
Then all generated file paths should start with "tests/"
And all generated file paths should end with ".py"
And all generated file names should start with "test_"
And no generated file path should start with "src/"
robot/helper_wf02_test_generation.py
+47
View File
@@ -0,0 +1,47 @@
"""WF02 helper command dispatcher for Robot integration tests."""
from __future__ import annotations
import sys
from collections.abc import Callable
from pathlib import Path
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
sys.path.insert(0, _ROBOT)
from helpers_common import reset_global_state # noqa: E402
from wf02_test_generation_artifacts import ( # noqa: E402
wf02_mocked_generation_artifacts,
)
from wf02_test_generation_commands import command_map # noqa: E402
from wf02_test_generation_validation import wf02_coverage_validation # noqa: E402
COMMANDS: dict[str, Callable[[], None]] = command_map(
mocked_generation_artifacts=wf02_mocked_generation_artifacts,
coverage_validation=wf02_coverage_validation,
)
def main() -> int:
"""Command dispatcher used by Robot ``Run Process`` calls."""
if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
print(
"Usage: helper_wf02_test_generation.py "
"<trusted-lifecycle|mocked-generation-artifacts|coverage-validation"
"|path-safety-guardrails>",
file=sys.stderr,
)
return 1
reset_global_state()
COMMANDS[sys.argv[1]]()
return 0
if __name__ == "__main__":
raise SystemExit(main())
robot/wf02_test_generation_artifacts.py
+196
View File
@@ -0,0 +1,196 @@
"""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(
operation="_cleveragents/plan/artifacts",
params={"plan_id": plan_id},
)
)
assert artifacts_response.status == "ok", artifacts_response.error
response_data = artifacts_response.data
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")
robot/wf02_test_generation_commands.py
+77
View File
@@ -0,0 +1,77 @@
"""WF02 command entry points used by helper dispatcher."""
from __future__ import annotations
from collections.abc import Callable
from wf02_test_generation_common import (
auth_module_fixture,
create_trusted_plan,
create_wf02_action,
setup_lifecycle,
validated_relative_test_path,
)
from wf02_test_generation_lifecycle import assert_rejected_generated_path
from cleveragents.domain.models.core.automation_profile import BUILTIN_PROFILES
from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState
def wf02_trusted_lifecycle() -> None:
"""Verify trusted profile auto-runs strategize+execute only."""
lifecycle = setup_lifecycle()
create_wf02_action(lifecycle)
plan_id = create_trusted_plan(lifecycle)
trusted_profile = BUILTIN_PROFILES["trusted"]
# Spec-aligned task-threshold fields (legacy phase names were removed).
assert trusted_profile.decompose_task == 0.0
assert trusted_profile.create_tool == 0.0
assert trusted_profile.select_tool == 1.0
plan = lifecycle.try_auto_run(plan_id)
assert plan.phase == PlanPhase.EXECUTE, (
f"Expected execute phase for trusted profile, got {plan.phase.value}"
)
assert plan.state == ProcessingState.COMPLETE, (
"Expected execute/complete after trusted auto-run"
)
assert plan.automation_profile is not None
assert plan.automation_profile.profile_name == "trusted"
print("wf02-trusted-lifecycle-ok")
def wf02_path_safety_guardrails() -> None:
"""Verify generated file guardrails reject unsafe output paths."""
with auth_module_fixture() as (fixture_root, _):
assert_rejected_generated_path(
lambda rel: validated_relative_test_path(fixture_root, rel),
"/tmp/test_abs.py",
"must be relative",
)
assert_rejected_generated_path(
lambda rel: validated_relative_test_path(fixture_root, rel),
"../tests/test_escape.py",
"escapes fixture root",
)
assert_rejected_generated_path(
lambda rel: validated_relative_test_path(fixture_root, rel),
"src/test_not_allowed.py",
"must stay in tests/",
)
print("wf02-path-safety-guardrails-ok")
def command_map(
mocked_generation_artifacts: Callable[[], None],
coverage_validation: Callable[[], None],
) -> dict[str, Callable[[], None]]:
"""Build command map for helper dispatcher."""
return {
"trusted-lifecycle": wf02_trusted_lifecycle,
"mocked-generation-artifacts": mocked_generation_artifacts,
"coverage-validation": coverage_validation,
"path-safety-guardrails": wf02_path_safety_guardrails,
}
robot/wf02_test_generation_common.py
+314
View File
@@ -0,0 +1,314 @@
"""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=(
"- Coverage of the target module reaches the specified threshold\n"
"- New tests pass\n"
"- Existing tests continue to pass\n"
"- Tests follow existing project conventions "
"(fixtures, naming, structure)\n"
"- No production code is modified"
),
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
robot/wf02_test_generation_integration.robot
+56
View File
@@ -0,0 +1,56 @@
*** Settings ***
Documentation Integration test for Workflow Example 2: automated test generation
... for a module using the trusted automation profile.
...
... Validates trusted-profile lifecycle behavior, mocked LLM-based
... deterministic test generation, coverage validation enforcement,
... and invariant-safe artifacts (test-only file changes).
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment With Database Isolation
Suite Teardown Cleanup Test Environment
Force Tags wf02 integration trusted workflow2 test-generation
*** Variables ***
${HELPER} ${CURDIR}/helper_wf02_test_generation.py
*** Test Cases ***
WF02 Trusted Profile Auto Runs Strategize And Execute
[Documentation] Verify trusted profile auto-runs strategize and execute
... while not auto-applying (human approval gate retained).
[Tags] lifecycle automation-profile trusted
${result}= Run Process ${PYTHON} ${HELPER} trusted-lifecycle cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf02-trusted-lifecycle-ok
WF02 Mocked Generation Produces Test Artifacts Only
[Documentation] Verify mocked LLM responses generate deterministic test
... files, assert user-facing plan artifacts command path,
... and artifacts list only test paths (no prod edits).
[Tags] mocking artifacts invariants
${result}= Run Process ${PYTHON} ${HELPER} mocked-generation-artifacts cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf02-mocked-generation-artifacts-ok
WF02 Coverage Validation Enforced For Target Threshold
[Documentation] Verify required coverage validation blocks below-threshold
... runs and passes after deterministic generated tests.
[Tags] validation coverage
${result}= Run Process ${PYTHON} ${HELPER} coverage-validation cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf02-coverage-validation-ok
WF02 Path Safety Guardrails Reject Unsafe Destinations
[Documentation] Verify generated-file path guardrails reject absolute,
... traversal, and non-tests destinations.
[Tags] guardrails path-safety
${result}= Run Process ${PYTHON} ${HELPER} path-safety-guardrails cwd=${WORKSPACE} timeout=120s on_timeout=kill
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf02-path-safety-guardrails-ok
robot/wf02_test_generation_lifecycle.py
+57
View File
@@ -0,0 +1,57 @@
"""WF02 lifecycle-level utility helpers."""
from __future__ import annotations
from collections.abc import Callable
from cleveragents.application.services.plan_execution_context import (
PlanExecutionContext,
)
from cleveragents.domain.models.core.change import (
ChangeEntry,
ChangeOperation,
InMemoryChangeSetStore,
)
def record_files_via_execution_context(
plan_id: str,
store: InMemoryChangeSetStore,
files: dict[str, str],
) -> str:
"""Record file artifacts through PlanExecutionContext lifecycle APIs."""
execution_context = PlanExecutionContext(
plan_id=plan_id,
automation_profile="trusted",
changeset_store=store,
)
changeset_id = execution_context.start_changeset()
for index, rel_path in enumerate(sorted(files), start=1):
execution_context.record_change(
ChangeEntry(
plan_id=plan_id,
resource_id=f"wf02-res-{index}",
tool_name="builtin/file-write",
operation=ChangeOperation.CREATE,
path=rel_path,
after_hash=f"wf02hash{index}",
)
)
return changeset_id
def assert_rejected_generated_path(
validator: Callable[[str], object],
rel_path: str,
expected_error_substring: str,
) -> None:
"""Assert the path safety guard rejects a path with expected reason."""
try:
validator(rel_path)
except AssertionError as exc:
message = str(exc)
assert expected_error_substring in message, (
f"Expected '{expected_error_substring}' in '{message}'"
)
else:
raise AssertionError(f"Expected path rejection for: {rel_path}")
robot/wf02_test_generation_validation.py
+452
View File
@@ -0,0 +1,452 @@
"""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,
mode="required",
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")