test(e2e): add M1 source-code plan lifecycle suite

robot/helper_m1_sourcecode_smoke.py

@@ -0,0 +1,299 @@
+"""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.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,
+    ):
+        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", _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.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,
+        ),
+    ):
+        # 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", _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]