cleveragents-core/robot/helper_m1_sourcecode_smoke.py

"""Helper script for m1_sourcecode_smoke.robot E2E tests.

Each subcommand is a self-contained check that prints a sentinel on success.
Uses ``--format plain`` where possible to stabilise assertion strings.
"""

from __future__ import annotations

import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, patch

# 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 typer.testing import CliRunner  # noqa: E402

from cleveragents.cli.commands.action import app as action_app  # noqa: E402
from cleveragents.cli.commands.plan import app as plan_app  # noqa: E402
from cleveragents.domain.models.core.action import (  # noqa: E402
    Action,
    ActionState,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
)

runner = CliRunner()

_PLAN_ULID = "01M1SM0KE00000000000000001"
_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m1"


def _mock_plan(
    *,
    phase: PlanPhase = PlanPhase.STRATEGIZE,
    state: ProcessingState = ProcessingState.QUEUED,
    project_links: list[ProjectLink] | None = None,
) -> Plan:
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=_PLAN_ULID),
        namespaced_name=NamespacedName.parse("local/m1-smoke-plan"),
        description="M1 smoke test plan",
        definition_of_done="Source code reviewed",
        action_name="local/m1-source-review",
        phase=phase,
        processing_state=state,
        project_links=project_links or [],
        arguments={},
        arguments_order=[],
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        reusable=True,
        read_only=False,
        created_by=None,
        timestamps=PlanTimestamps(created_at=now, updated_at=now),
    )


def _mock_action(name: str = "local/m1-source-review") -> Action:
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="Minimal source-code review action",
        long_description=None,
        definition_of_done="Source code reviewed",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        reusable=True,
        read_only=True,
        state=ActionState.AVAILABLE,
        created_by=None,
        created_at=datetime.now(),
        updated_at=datetime.now(),
    )


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def action_create() -> None:
    """Create an action from M1 fixture YAML config."""
    config_path = _FIXTURES_DIR / "action_sourcecode.yaml"
    mock_svc = MagicMock()
    mock_svc.create_action.return_value = _mock_action()
    with patch(
        "cleveragents.cli.commands.action._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(action_app, ["create", "--config", str(config_path)])
    if result.exit_code == 0:
        print("m1-action-create-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_use() -> None:
    """Use action to create a plan in strategize phase."""
    mock_svc = MagicMock()
    mock_svc.get_action_by_name.return_value = _mock_action()
    mock_svc.use_action.return_value = _mock_plan()
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(plan_app, ["use", "local/m1-source-review"])
    if result.exit_code == 0 and "strategize" in result.output.lower():
        print("m1-plan-use-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_use_with_project() -> None:
    """Use action with a project argument."""
    mock_svc = MagicMock()
    mock_svc.get_action_by_name.return_value = _mock_action()
    mock_svc.use_action.return_value = _mock_plan(
        project_links=[ProjectLink(project_name="local/m1-smoke-proj")]
    )
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            plan_app,
            ["use", "local/m1-source-review", "local/m1-smoke-proj"],
        )
    if result.exit_code == 0:
        print("m1-plan-use-project-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_execute() -> None:
    """Execute a plan and verify phase transition."""
    mock_svc = MagicMock()
    strategize_plan = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE
    )
    mock_svc.list_plans.return_value = [strategize_plan]
    mock_svc.get_plan.return_value = strategize_plan
    mock_svc.execute_plan.return_value = _mock_plan(
        phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
    )
    with (
        patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service",
            return_value=mock_svc,
        ),
        patch(
            "cleveragents.cli.commands.plan._get_plan_executor",
            return_value=MagicMock(),
        ),
    ):
        result = runner.invoke(plan_app, ["execute"])
    if result.exit_code == 0 and "execute" in result.output.lower():
        print("m1-plan-execute-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_diff() -> None:
    """Show plan diff (changeset)."""
    mock_apply_svc = MagicMock()
    mock_apply_svc.diff.return_value = "No changes detected."
    with patch(
        "cleveragents.cli.commands.plan._get_apply_service",
        return_value=mock_apply_svc,
    ):
        result = runner.invoke(plan_app, ["diff", _PLAN_ULID])
    if result.exit_code == 0:
        print("m1-plan-diff-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_apply() -> None:
    """Apply a plan and verify terminal state."""
    mock_svc = MagicMock()
    mock_svc.apply_plan.return_value = _mock_plan(
        phase=PlanPhase.APPLY, state=ProcessingState.APPLIED
    )
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(plan_app, ["lifecycle-apply", "--yes", _PLAN_ULID])
    if result.exit_code == 0:
        print("m1-plan-apply-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def full_lifecycle() -> None:
    """End-to-end: action create -> plan use -> execute -> apply."""
    config_path = _FIXTURES_DIR / "action_sourcecode.yaml"
    mock_svc = MagicMock()
    mock_svc.create_action.return_value = _mock_action()
    mock_svc.get_action_by_name.return_value = _mock_action()
    mock_svc.use_action.return_value = _mock_plan()
    strategize_plan = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE
    )
    mock_svc.list_plans.return_value = [strategize_plan]
    mock_svc.get_plan.return_value = strategize_plan
    mock_svc.execute_plan.return_value = _mock_plan(
        phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
    )
    mock_svc.apply_plan.return_value = _mock_plan(
        phase=PlanPhase.APPLY, state=ProcessingState.APPLIED
    )

    with (
        patch(
            "cleveragents.cli.commands.action._get_lifecycle_service",
            return_value=mock_svc,
        ),
        patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service",
            return_value=mock_svc,
        ),
        patch(
            "cleveragents.cli.commands.plan._get_plan_executor",
            return_value=MagicMock(),
        ),
    ):
        # Step 1: action create
        r1 = runner.invoke(action_app, ["create", "--config", str(config_path)])
        if r1.exit_code != 0:
            print(f"FAIL step 1: exit={r1.exit_code} output={r1.output}")
            sys.exit(1)

        # Step 2: plan use
        r2 = runner.invoke(plan_app, ["use", "local/m1-source-review"])
        if r2.exit_code != 0:
            print(f"FAIL step 2: exit={r2.exit_code} output={r2.output}")
            sys.exit(1)

        # Step 3: plan execute
        r3 = runner.invoke(plan_app, ["execute"])
        if r3.exit_code != 0:
            print(f"FAIL step 3: exit={r3.exit_code} output={r3.output}")
            sys.exit(1)

        # Step 4: plan apply
        r4 = runner.invoke(plan_app, ["lifecycle-apply", "--yes", _PLAN_ULID])
        if r4.exit_code != 0:
            print(f"FAIL step 4: exit={r4.exit_code} output={r4.output}")
            sys.exit(1)

    print("m1-full-lifecycle-ok")


def plan_use_plain() -> None:
    """Plan use with --format plain to stabilise assertion strings."""
    mock_svc = MagicMock()
    mock_svc.get_action_by_name.return_value = _mock_action()
    mock_svc.use_action.return_value = _mock_plan()
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            plan_app,
            ["use", "local/m1-source-review", "--format", "plain"],
        )
    if result.exit_code == 0:
        print("m1-plan-use-plain-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, object] = {
    "action-create": action_create,
    "plan-use": plan_use,
    "plan-use-project": plan_use_with_project,
    "plan-execute": plan_execute,
    "plan-diff": plan_diff,
    "plan-apply": plan_apply,
    "full-lifecycle": full_lifecycle,
    "plan-use-plain": plan_use_plain,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
        sys.exit(1)
    fn = _COMMANDS[sys.argv[1]]
    fn()  # type: ignore[operator]