cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity
Files
master
cleveragents-core/robot/helper_wf07_cicd.py
T
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

499 lines
20 KiB
Python
Raw Permalink Blame History

"""Helper script for wf07_cicd_integration.robot tests.
Each subcommand is a self-contained integration check that prints a sentinel
on success and exercises specification workflow example 7.
"""
from __future__ import annotations
import json
import sys
import tempfile
from collections.abc import Callable
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from sqlalchemy import create_engine # noqa: E402
from sqlalchemy.exc import IntegrityError # noqa: E402
from sqlalchemy.orm import sessionmaker # noqa: E402
from cleveragents.application.services.automation_profile_service import ( # noqa: E402
AutomationProfileService,
)
from cleveragents.application.services.config_service import ( # noqa: E402
ConfigService,
)
from cleveragents.application.services.plan_lifecycle_service import ( # noqa: E402
PlanLifecycleService,
)
from cleveragents.application.services.resource_registry_service import ( # noqa: E402
ResourceRegistryService,
)
from cleveragents.application.services.tool_registry_service import ( # noqa: E402
ToolRegistryService,
)
from cleveragents.application.services.validation_pipeline import ( # noqa: E402
ValidationCommand,
ValidationPipeline,
)
from cleveragents.config.settings import Settings # noqa: E402
from cleveragents.core.exceptions import DatabaseError # noqa: E402
from cleveragents.domain.models.core.action import ( # noqa: E402
ActionArgument,
ArgumentRequirement,
ArgumentType,
)
from cleveragents.domain.models.core.plan import ( # noqa: E402
InvariantSource,
PlanPhase,
ProcessingState,
ProjectLink,
)
from cleveragents.domain.models.core.project import ( # noqa: E402
NamespacedProject,
parse_namespaced_name,
)
from cleveragents.domain.models.core.tool import ValidationMode # noqa: E402
from cleveragents.infrastructure.database.models import Base # noqa: E402
from cleveragents.infrastructure.database.repositories import ( # noqa: E402
NamespacedProjectRepository,
ToolRegistryRepository,
ValidationAttachmentRepository,
)
# ---------------------------------------------------------------------------
# Database helpers (in-memory SQLite)
# ---------------------------------------------------------------------------
class _NoCloseSession:
"""Wrapper that prevents the session from being closed by service code."""
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_db() -> tuple[Any, Any]:
"""Create an in-memory SQLite 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() -> object:
return wrapper
return factory, session
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def config_ci_profile() -> None:
"""Set and get ci profile, json format, and log level via ConfigService."""
with tempfile.TemporaryDirectory() as tmpdir:
svc = ConfigService(config_dir=Path(tmpdir))
svc.set_value("core.automation-profile", "ci")
resolved = svc.resolve("core.automation-profile")
assert resolved.value == "ci", f"Expected 'ci', got '{resolved.value}'"
svc.set_value("core.format", "json")
resolved = svc.resolve("core.format")
assert resolved.value == "json", f"Expected 'json', got '{resolved.value}'"
svc.set_value("core.log.level", "WARN")
resolved = svc.resolve("core.log.level")
assert resolved.value == "WARN", f"Expected 'WARN', got '{resolved.value}'"
print("wf07-config-ci-profile-ok")
def resource_idempotent() -> None:
"""Register a git-checkout resource twice; verify only one exists."""
factory, _session = _setup_db()
svc = ResourceRegistryService(session_factory=factory)
svc.bootstrap_builtin_types()
# First registration — includes branch per spec Step 3
with tempfile.TemporaryDirectory() as repo_dir:
svc.register_resource(
type_name="git-checkout",
name="local/cicd-repo",
location=repo_dir,
description="CI/CD repository",
properties={"branch": "fix/handle-null-users"},
)
# Second registration mirrors spec idempotency; CI handles duplicates with || true.
duplicate_raised = False
with tempfile.TemporaryDirectory() as dup_dir:
try:
svc.register_resource(
type_name="git-checkout",
name="local/cicd-repo",
location=dup_dir,
description="CI/CD repository (updated)",
)
except (IntegrityError, DatabaseError):
duplicate_raised = True
# The DB unique constraint should prevent a second resource
assert duplicate_raised, (
"Expected IntegrityError or DatabaseError on duplicate resource registration"
)
resources = svc.list_resources(type_name="git-checkout")
assert len(resources) == 1, f"Expected 1 resource, got {len(resources)}"
assert resources[0].description == "CI/CD repository"
assert resources[0].properties.get("branch") == "fix/handle-null-users", (
f"Expected branch 'fix/handle-null-users', got {resources[0].properties}"
)
print("wf07-resource-idempotent-ok")
def project_idempotent() -> None:
"""Create a project twice; verify it exists exactly once."""
factory, _session = _setup_db()
proj_repo = NamespacedProjectRepository(session_factory=factory)
parsed = parse_namespaced_name("cicd-project")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
description="CI/CD integration project",
)
proj_repo.create(proj)
# Second creation with same name: expect DatabaseError (integrity violation).
# NOTE: NamespacedProjectRepository.create has @database_retry (3 attempts,
# 0.5s wait) so the expected IntegrityError→DatabaseError adds ~1s overhead.
duplicate_raised = False
try:
proj_dup = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
description="CI/CD integration project (dup)",
)
proj_repo.create(proj_dup)
except (DatabaseError, IntegrityError):
duplicate_raised = True
assert duplicate_raised, (
"Expected DatabaseError or IntegrityError on duplicate project creation"
)
projects = proj_repo.list_projects()
assert len(projects) == 1, f"Expected exactly 1 project, got {len(projects)}"
fetched = proj_repo.get("local/cicd-project")
assert fetched.name == "cicd-project", (
f"Expected name 'cicd-project', got '{fetched.name}'"
)
print("wf07-project-idempotent-ok")
def validation_attach() -> None:
"""Register validation tools, attach to a resource, and run pipeline."""
factory, _session = _setup_db()
tool_repo = ToolRegistryRepository(session_factory=factory)
attachment_repo = ValidationAttachmentRepository(session_factory=factory)
svc = ToolRegistryService(
tool_repo=tool_repo,
attachment_repo=attachment_repo,
)
# Spec Step 3 registers 3 validations: ci-lint, ci-typecheck, ci-tests
now = datetime.now(tz=timezone.utc).isoformat() # noqa: UP017
validation_names = ["local/ci-lint", "local/ci-typecheck", "local/ci-tests"]
for vname in validation_names:
svc.register_tool(
{
"name": vname,
"description": f"CI validation: {vname.rsplit('/', 1)[-1]}",
"tool_type": "validation",
"source": "builtin",
"timeout": 300,
"created_at": now,
"updated_at": now,
"resource_bindings": [],
"mode": "required",
}
)
# Verify all 3 tools are registered
for vname in validation_names:
registered = svc.get_tool(vname)
assert registered is not None, f"Validation tool {vname} must be registered"
# --- Step 2: Attach to resource (spec: ``validation attach``) ---
# Register a resource so the attachment references a real entity.
res_svc = ResourceRegistryService(session_factory=factory)
res_svc.bootstrap_builtin_types()
with tempfile.TemporaryDirectory(prefix="val-test-repo-") as val_tmpdir:
registered_resource = res_svc.register_resource(
type_name="git-checkout",
name="local/cicd-repo",
location=val_tmpdir,
description="Validation test repo",
)
resource_id = registered_resource.resource_id
# Attach all 3 validations per spec Step 3
for vname in validation_names:
attachment = svc.attach_validation(
validation_name=vname,
resource_id=resource_id,
project_name="local/ci-workspace",
)
assert attachment is not None, f"Attachment for {vname} must be created"
# Verify all 3 attachments exist
attachments = svc.list_validations_for_resource(resource_id)
assert len(attachments) == 3, f"Expected 3 attachments, got {len(attachments)}"
# --- Step 3: Verify pipeline execution with the validations ---
commands = [
ValidationCommand(
validation_name=vn,
resource_id=resource_id,
resource_name="local/cicd-repo",
mode=ValidationMode.REQUIRED,
arguments={"language": "python"},
timeout_seconds=30.0,
)
for vn in validation_names
]
def mock_executor(
validation_name: str,
arguments: dict[str, Any],
) -> dict[str, Any]:
return {"passed": True, "message": f"{validation_name} passed"}
pipeline = ValidationPipeline(
commands=commands,
executor=mock_executor,
max_workers=1,
)
summary = pipeline.run()
assert summary.total == 3, f"Expected 3 validations, got {summary.total}"
assert summary.all_required_passed, "Expected all required validations to pass"
print("wf07-validation-attach-ok")
def ci_plan_lifecycle() -> None:
"""Create action, verify phase transitions, and assert terminal states."""
settings = Settings()
service = PlanLifecycleService(settings=settings)
# Create action for PR review (spec Step 2). Actor names use local/
# prefixes (no real API connectivity in integration tests).
action = service.create_action(
name="local/review-pr",
description="Automatically review a PR and fix issues",
definition_of_done="pr_branch",
strategy_actor="local/ci-planner",
execution_actor="local/ci-executor",
automation_profile="ci",
reusable=True,
arguments=[
ActionArgument(
name="pr_branch",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.REQUIRED,
description="Branch name of the PR",
),
ActionArgument(
name="base_branch",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.OPTIONAL,
description="Base branch to compare against",
default_value="main",
),
],
invariants=[
"Only modify files that are already changed in the PR",
"Do not change the intent of any code"
" — only fix style, types, and test issues",
"All fixes must include a comment explaining what was changed and why",
],
)
assert str(action.namespaced_name) == "local/review-pr" # L6: correctness
assert action.automation_profile == "ci", (
f"Expected profile 'ci', got '{action.automation_profile}'"
)
assert len(action.arguments) == 2
assert len(action.invariants) == 3
assert action.reusable is True, "Expected reusable=True"
# Use the action to create a plan with spec-required arguments
plan = service.use_action(
action_name=str(action.namespaced_name),
project_links=[ProjectLink(project_name="cicd-project")],
created_by="ci-pipeline",
arguments={"pr_branch": "fix/handle-null-users", "base_branch": "main"},
)
assert plan.phase == PlanPhase.APPLY
assert plan.state == ProcessingState.APPLIED
plan_id = plan.identity.plan_id
assert plan_id, "Plan must have a plan_id"
# Verify arguments flowed to the plan
assert plan.arguments == {
"pr_branch": "fix/handle-null-users",
"base_branch": "main",
}, f"Plan arguments mismatch: {plan.arguments}"
# Verify invariants flowed from action to plan — count and content (M5)
action_invariants = [
inv for inv in plan.invariants if inv.source == InvariantSource.ACTION
]
assert len(action_invariants) == 3, (
f"Expected 3 action invariants on plan, got {len(action_invariants)}"
)
expected_inv_texts = sorted(
[
"Only modify files that are already changed in the PR",
"Do not change the intent of any code"
" — only fix style, types, and test issues",
"All fixes must include a comment explaining what was changed and why",
]
)
actual_inv_texts = sorted(inv.text for inv in action_invariants)
assert actual_inv_texts == expected_inv_texts, (
f"Invariant text mismatch: {actual_inv_texts}"
)
fetched = service.get_plan(plan_id)
assert fetched.identity.plan_id == plan_id
assert fetched.action_name == str(action.namespaced_name)
# Verify the automation profile resolved to ci
profile_svc = AutomationProfileService(repo=None)
ci_profile = profile_svc.resolve_profile(plan_profile="ci")
assert ci_profile.name == "ci"
# use_action now resolves automation profile precedence and can auto-progress
# beyond phase boundaries for non-manual profiles (e.g., ci auto-runs to Applied).
# --- Phase-by-phase plan completion (spec Step 3) ---
p = service.get_plan(plan_id)
if not p.is_terminal:
service.start_strategize(plan_id)
p = service.get_plan(plan_id)
assert p.state == ProcessingState.PROCESSING, (
f"Expected PROCESSING after start_strategize, got {p.state}"
)
service.complete_strategize(plan_id)
p = service.get_plan(plan_id)
assert (p.phase, p.state) in {
(PlanPhase.STRATEGIZE, ProcessingState.COMPLETE),
(PlanPhase.EXECUTE, ProcessingState.QUEUED),
}, (
"After complete_strategize expected strategize/complete or execute/queued, "
f"got {p.phase.value}/{p.state.value}"
)
if p.phase == PlanPhase.STRATEGIZE:
service.execute_plan(plan_id)
# Execute phase
service.start_execute(plan_id)
service.complete_execute(plan_id)
p = service.get_plan(plan_id)
assert (p.phase, p.state) in {
(PlanPhase.EXECUTE, ProcessingState.COMPLETE),
(PlanPhase.APPLY, ProcessingState.QUEUED),
(PlanPhase.APPLY, ProcessingState.APPLIED),
}, (
"After complete_execute expected execute/complete, apply/queued, "
"or apply/applied, "
f"got {p.phase.value}/{p.state.value}"
)
if p.phase == PlanPhase.EXECUTE:
service.apply_plan(plan_id)
# Apply phase
if p.phase == PlanPhase.APPLY and p.state == ProcessingState.QUEUED:
service.start_apply(plan_id)
service.complete_apply(plan_id)
# Verify terminal state — the polling loop exits on 'applied'
final = service.get_plan(plan_id)
assert final.state == ProcessingState.APPLIED, (
f"Expected APPLIED, got {final.state}"
)
cancelled_plan = service.use_action(
action_name=str(action.namespaced_name),
project_links=[ProjectLink(project_name="cicd-project")],
created_by="ci-pipeline",
arguments={"pr_branch": "fix/handle-null-users", "base_branch": "main"},
)
assert cancelled_plan.state == ProcessingState.APPLIED, (
"CI profile auto-runs plan to terminal APPLIED state"
)
print("wf07-ci-plan-lifecycle-ok")
def json_output() -> None:
"""Verify JSON dict from Action and Plan models via as_cli_dict()."""
settings = Settings()
service = PlanLifecycleService(settings=settings)
# Create an action with arguments to verify full JSON round-trip
action = service.create_action(
name="local/json-test",
description="JSON output test action",
definition_of_done="Verify JSON output",
strategy_actor="local/s",
execution_actor="local/e",
arguments=[
ActionArgument(
name="pr_branch",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.REQUIRED,
description="Branch name of the PR",
),
ActionArgument(
name="base_branch",
arg_type=ArgumentType.STRING,
requirement=ArgumentRequirement.OPTIONAL,
description="Base branch to compare against",
default_value="main",
),
],
)
# Get CLI dict representation (simulates --format json output)
parsed = json.loads(json.dumps(action.as_cli_dict()))
assert isinstance(parsed, dict), "Parsed JSON must be a dict"
assert parsed.get("name") == "local/json-test"
assert parsed.get("state") == "available"
assert "description" in parsed, "JSON must contain 'description' field"
# Create a plan with arguments to verify full JSON output per spec Step 3
plan = service.use_action(
action_name=str(action.namespaced_name),
project_links=[ProjectLink(project_name="json-project")],
arguments={"pr_branch": "fix/test", "base_branch": "main"},
)
plan_parsed = json.loads(json.dumps(plan.as_cli_dict()))
# NOTE: as_cli_dict() uses "arguments"/"projects" (plural/dict) while
# the spec sample JSON (line 39199) uses "args"/"project" (singular).
# as_cli_dict() also omits "attempt" and "resources" (CLI-level fields)
# and "automation_profile" (not yet propagated — TODO #1060).
assert "plan_id" in plan_parsed
assert plan_parsed.get("phase") == "strategize"
assert plan_parsed.get("state") == "queued"
assert plan_parsed.get("action") == "local/json-test"
assert plan_parsed["projects"][0]["name"] == "json-project"
assert plan_parsed["arguments"]["pr_branch"] == "fix/test"
print("wf07-json-output-ok")
# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"config-ci-profile": config_ci_profile,
"resource-idempotent": resource_idempotent,
"project-idempotent": project_idempotent,
"validation-attach": validation_attach,
"ci-plan-lifecycle": ci_plan_lifecycle,
"json-output": json_output,
}
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
print(
f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
file=sys.stderr,
)
sys.exit(1)
_COMMANDS[sys.argv[1]]()
Reference in New Issue View Git Blame Copy Permalink