cleveragents-core/robot/helper_apply_pipeline.py

"""Robot Framework helper for apply pipeline smoke tests.

Provides a CLI-style interface for Robot to invoke validation-gated
apply operations. Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_apply_pipeline.py apply-passes
    python robot/helper_apply_pipeline.py apply-blocked
    python robot/helper_apply_pipeline.py already-applied
    python robot/helper_apply_pipeline.py empty-changeset
    python robot/helper_apply_pipeline.py model-checks
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock

# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.application.services.plan_apply_service import (  # noqa: E402
    ApplyOutcome,
    ApplyResult,
    PlanApplyService,
)
from cleveragents.domain.models.core.change import (  # noqa: E402
    ChangeEntry,
    ChangeOperation,
    SpecChangeSet,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
)

_PLAN_ID = "01PLANTEST000000000000001"


def _make_plan(
    *,
    phase: PlanPhase = PlanPhase.EXECUTE,
    state: ProcessingState = ProcessingState.COMPLETE,
    changeset_id: str | None = "CS001",
    validation_summary: dict[str, Any] | None = None,
    is_terminal: bool = False,
) -> MagicMock:
    plan = MagicMock()
    plan.identity.plan_id = _PLAN_ID
    plan.phase = phase
    plan.processing_state = state
    plan.changeset_id = changeset_id
    plan.validation_summary = validation_summary
    plan.is_terminal = is_terminal
    plan.error_details = None
    plan.timestamps = PlanTimestamps()
    return plan


def _make_changeset(n: int = 1) -> SpecChangeSet:
    cs = SpecChangeSet(plan_id=_PLAN_ID)
    for i in range(n):
        cs.add_change(
            ChangeEntry(
                plan_id=_PLAN_ID,
                resource_id="RES001",
                tool_name="builtin/test-tool",
                operation=ChangeOperation.MODIFY,
                path=f"src/file_{i}.py",
                before_hash=f"b{i}",
                after_hash=f"a{i}",
            )
        )
    return cs


def _make_service(
    plan: MagicMock, changeset: SpecChangeSet | None = None
) -> PlanApplyService:
    lifecycle = MagicMock()
    lifecycle.get_plan.return_value = plan
    lifecycle.complete_apply.return_value = plan
    lifecycle.constrain_apply.return_value = plan
    lifecycle.commit_plan = MagicMock()
    store = MagicMock()
    store.get.return_value = changeset
    return PlanApplyService(lifecycle_service=lifecycle, changeset_store=store)


def _cmd_apply_passes() -> int:
    """Apply with all validations passing."""
    cs = _make_changeset(3)
    plan = _make_plan(
        changeset_id=cs.changeset_id,
        validation_summary={
            "total": 2,
            "required_passed": 2,
            "required_failed": 0,
        },
    )
    svc = _make_service(plan, cs)
    result = svc.apply_with_validation_gate(_PLAN_ID)
    assert result.outcome == ApplyOutcome.APPLIED, f"Got {result.outcome}"
    assert result.files_changed == 3
    assert "applied successfully" in result.message
    print("apply-passes-ok")
    return 0


def _cmd_apply_blocked() -> int:
    """Apply blocked by validation failure."""
    cs = _make_changeset(2)
    plan = _make_plan(
        changeset_id=cs.changeset_id,
        validation_summary={
            "total": 3,
            "required_passed": 1,
            "required_failed": 2,
        },
    )
    svc = _make_service(plan, cs)
    result = svc.apply_with_validation_gate(_PLAN_ID)
    assert result.outcome == ApplyOutcome.CONSTRAINED, f"Got {result.outcome}"
    assert result.validations_failed == 2
    assert "Apply refused" in result.message
    print("apply-blocked-ok")
    return 0


def _cmd_already_applied() -> int:
    """Re-apply on already applied plan."""
    cs = _make_changeset(1)
    plan = _make_plan(
        phase=PlanPhase.APPLY,
        state=ProcessingState.APPLIED,
        is_terminal=True,
    )
    svc = _make_service(plan, cs)
    result = svc.apply_with_validation_gate(_PLAN_ID)
    assert result.outcome == ApplyOutcome.ALREADY_APPLIED, f"Got {result.outcome}"
    assert "already been applied" in result.message
    print("already-applied-ok")
    return 0


def _cmd_empty_changeset() -> int:
    """Apply blocked on empty changeset."""
    cs = _make_changeset(0)
    plan = _make_plan(changeset_id=cs.changeset_id)
    svc = _make_service(plan, cs)
    result = svc.apply_with_validation_gate(_PLAN_ID)
    assert result.outcome == ApplyOutcome.BLOCKED_EMPTY, f"Got {result.outcome}"
    assert "empty ChangeSet" in result.message
    print("empty-changeset-ok")
    return 0


def _cmd_model_checks() -> int:
    """Verify ApplyOutcome and ApplyResult models."""
    # Check all enum values exist
    assert ApplyOutcome.APPLIED == "applied"
    assert ApplyOutcome.CONSTRAINED == "constrained"
    assert ApplyOutcome.ALREADY_APPLIED == "already_applied"
    assert ApplyOutcome.BLOCKED_EMPTY == "blocked_empty"

    # Check ApplyResult construction
    result = ApplyResult(
        outcome=ApplyOutcome.APPLIED,
        plan_id="TEST",
        message="ok",
        files_changed=5,
        validations_total=3,
        validations_passed=3,
        validations_failed=0,
    )
    assert result.outcome == ApplyOutcome.APPLIED
    assert result.plan_id == "TEST"
    assert result.files_changed == 5

    print("model-checks-ok")
    return 0


def main() -> int:
    if len(sys.argv) < 2:
        print(
            "Usage: helper_apply_pipeline.py "
            "<apply-passes|apply-blocked|already-applied|"
            "empty-changeset|model-checks>"
        )
        return 1

    cmd = sys.argv[1]
    if cmd == "apply-passes":
        return _cmd_apply_passes()
    if cmd == "apply-blocked":
        return _cmd_apply_blocked()
    if cmd == "already-applied":
        return _cmd_already_applied()
    if cmd == "empty-changeset":
        return _cmd_empty_changeset()
    if cmd == "model-checks":
        return _cmd_model_checks()

    print(f"Unknown command: {cmd}")
    return 1


if __name__ == "__main__":
    sys.exit(main())