freemo
1d36449a98
CI / build (push) Has been cancelled
CI / e2e_tests (push) Has been cancelled
CI / typecheck (push) Has been cancelled
CI / integration_tests (push) Has been cancelled
CI / quality (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / benchmark-publish (push) Has been cancelled
CI / security (push) Has been cancelled
CI / helm (push) Has been cancelled
CI / unit_tests (push) Has been cancelled
CI / benchmark-regression (push) Has been cancelled
CI / coverage (push) Has been cancelled
CI / docker (push) Has been cancelled
CI / status-check (push) Has been cancelled
Remove the --mode/-m CLI flag from the validation attach command to align with the specification, which defines validation mode as an inherent property of the validation definition set at registration time via validation add, not as a per-attachment override. The service layer (ToolRegistryService.attach_validation) no longer accepts mode as a parameter; instead it reads the mode from the validation's registered definition. The ToolRegistryRepository _to_legacy_domain now exposes the mode field so the service can access it. All tests that passed --mode to the CLI or service have been updated or removed. The invalid-mode validation scenarios were removed since the mode is no longer caller-supplied. Coverage remains at 98.7%. ISSUES CLOSED: #913 Co-authored-by: Jeffrey Phillips Freeman <jeffrey.freeman@cleverthis.com> Co-committed-by: Jeffrey Phillips Freeman <jeffrey.freeman@cleverthis.com>
503 lines
20 KiB
Python
503 lines
20 KiB
Python
"""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=(
|
|
"- All lint issues are resolved\n"
|
|
"- Type checking passes\n"
|
|
"- Test coverage does not decrease\n"
|
|
"- Security scan passes\n"
|
|
"- All fixes are committed to the 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.STRATEGIZE
|
|
assert plan.state == ProcessingState.QUEUED
|
|
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).
|
|
# --- Phase-by-phase plan completion (spec Step 3) ---
|
|
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"},
|
|
)
|
|
cancelled = service.cancel_plan(cancelled_plan.identity.plan_id, reason="ci cancel")
|
|
assert cancelled.state == ProcessingState.CANCELLED
|
|
assert cancelled.is_terminal
|
|
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]]()
|