cleveragents-core/robot/helper_cli_lifecycle_e2e.py

"""Helper script for cli_lifecycle_e2e.robot end-to-end smoke tests.

Each subcommand is self-contained and prints a sentinel on success.
"""

from __future__ import annotations

import os
import sys
import tempfile
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 Action, ActionState  # noqa: E402
from cleveragents.domain.models.core.plan import (  # noqa: E402
    AutomationProfileProvenance,
    AutomationProfileRef,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
)

runner = CliRunner()

_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"

_VALID_YAML = """\
name: local/e2e-action
description: E2E lifecycle action
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: All e2e tests pass
"""


def _mock_action(name: str = "local/e2e-action") -> Action:
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="E2E lifecycle action",
        long_description=None,
        definition_of_done="All e2e tests pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        state=ActionState.AVAILABLE,
        reusable=True,
        read_only=False,
        created_at=datetime.now(),
        updated_at=datetime.now(),
        created_by=None,
    )


def _mock_plan(
    name: str = "local/e2e-plan",
    phase: PlanPhase = PlanPhase.STRATEGIZE,
    state: ProcessingState = ProcessingState.QUEUED,
    project_links: list[ProjectLink] | None = None,
    plan_id: str = _PLAN_ULID,
) -> Plan:
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=plan_id),
        namespaced_name=NamespacedName.parse(name),
        description="E2E lifecycle plan",
        definition_of_done="Tests pass",
        action_name="local/e2e-action",
        phase=phase,
        processing_state=state,
        project_links=project_links or [ProjectLink(project_name="proj-a")],
        arguments={"target_coverage": 80},
        arguments_order=["target_coverage"],
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        ),
        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 _write_yaml(content: str) -> str:
    fd, path = tempfile.mkstemp(suffix=".yaml")
    with os.fdopen(fd, "w") as fh:
        fh.write(content)
    return path


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


def action_create() -> None:
    """Verify action create from config file."""
    mock_service = MagicMock()
    mock_service.create_action.return_value = _mock_action()
    yaml_path = _write_yaml(_VALID_YAML)

    try:
        with patch(
            "cleveragents.cli.commands.action._get_lifecycle_service",
            return_value=mock_service,
        ):
            result = runner.invoke(action_app, ["create", "--config", yaml_path])
            if result.exit_code == 0:
                print("cli-lifecycle-action-create-ok")
            else:
                print(
                    f"FAIL: action create returned {result.exit_code}", file=sys.stderr
                )
                print(result.output, file=sys.stderr)
                sys.exit(1)
    finally:
        os.unlink(yaml_path)


def plan_use() -> None:
    """Verify plan use creates a plan in Strategize phase."""
    mock_service = MagicMock()
    mock_service.get_action_by_name.return_value = _mock_action()
    mock_service.use_action.return_value = _mock_plan()

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = runner.invoke(plan_app, ["use", "local/e2e-action", "proj-a"])
        if result.exit_code == 0:
            print("cli-lifecycle-plan-use-ok")
        else:
            print(f"FAIL: plan use returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def plan_execute() -> None:
    """Verify plan execute transitions to Execute phase."""
    mock_service = MagicMock()
    mock_service.get_plan.return_value = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE
    )
    mock_service.execute_plan.return_value = _mock_plan(
        phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED
    )

    with (
        patch(
            "cleveragents.cli.commands.plan._get_lifecycle_service",
            return_value=mock_service,
        ),
        patch(
            "cleveragents.cli.commands.plan._get_plan_executor",
            return_value=MagicMock(),
        ),
    ):
        result = runner.invoke(plan_app, ["execute", _PLAN_ULID])
        if result.exit_code == 0:
            print("cli-lifecycle-plan-execute-ok")
        else:
            print(f"FAIL: plan execute returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def plan_apply() -> None:
    """Verify plan apply transitions to Apply phase."""
    mock_service = MagicMock()
    mock_service.apply_plan.return_value = _mock_plan(
        phase=PlanPhase.APPLY, state=ProcessingState.QUEUED
    )

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID])
        if result.exit_code == 0:
            print("cli-lifecycle-plan-apply-ok")
        else:
            print(f"FAIL: plan apply returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def plan_status() -> None:
    """Verify plan status shows plan details."""
    mock_service = MagicMock()
    mock_service.get_plan.return_value = _mock_plan(
        project_links=[ProjectLink(project_name="proj-a", alias="api")]
    )

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = runner.invoke(plan_app, ["status", _PLAN_ULID])
        if result.exit_code == 0:
            print("cli-lifecycle-plan-status-ok")
        else:
            print(f"FAIL: plan status returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def plan_cancel() -> None:
    """Verify plan cancel cancels a non-terminal plan."""
    mock_service = MagicMock()
    mock_service.cancel_plan.return_value = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.CANCELLED
    )

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = runner.invoke(
            plan_app, ["cancel", _PLAN_ULID, "--reason", "E2E test cancel"]
        )
        if result.exit_code == 0:
            print("cli-lifecycle-plan-cancel-ok")
        else:
            print(f"FAIL: plan cancel returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def plan_list() -> None:
    """Verify list shows plan table."""
    mock_service = MagicMock()
    mock_service.list_plans.return_value = [
        _mock_plan(name="local/plan-a", plan_id="01KHDE6WWS2171PWW3GJEBXZ8A"),
        _mock_plan(name="local/plan-b", plan_id="01KHDE6WWS2171PWW3GJEBXZ8B"),
    ]

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = runner.invoke(plan_app, ["list"])
        if result.exit_code == 0:
            print("cli-lifecycle-plan-list-ok")
        else:
            print(f"FAIL: plan list returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def full_lifecycle() -> None:
    """End-to-end: action create -> plan use -> execute -> apply."""
    mock_service = MagicMock()
    action = _mock_action()
    mock_service.create_action.return_value = action
    mock_service.get_action_by_name.return_value = action

    plan_strat = _mock_plan(phase=PlanPhase.STRATEGIZE, state=ProcessingState.QUEUED)
    plan_exec = _mock_plan(phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED)
    plan_apply = _mock_plan(phase=PlanPhase.APPLY, state=ProcessingState.QUEUED)

    mock_service.use_action.return_value = plan_strat
    mock_service.execute_plan.return_value = plan_exec
    mock_service.get_plan.return_value = _mock_plan(
        phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE
    )
    mock_service.apply_plan.return_value = plan_apply

    yaml_path = _write_yaml(_VALID_YAML)
    try:
        with (
            patch(
                "cleveragents.cli.commands.action._get_lifecycle_service",
                return_value=mock_service,
            ),
            patch(
                "cleveragents.cli.commands.plan._get_lifecycle_service",
                return_value=mock_service,
            ),
            patch(
                "cleveragents.cli.commands.plan._get_plan_executor",
                return_value=MagicMock(),
            ),
        ):
            # Step 1: Create action
            r1 = runner.invoke(action_app, ["create", "--config", yaml_path])
            assert r1.exit_code == 0, f"action create failed: {r1.output}"

            # Step 2: Plan use
            r2 = runner.invoke(plan_app, ["use", "local/e2e-action", "proj-a"])
            assert r2.exit_code == 0, f"plan use failed: {r2.output}"

            # Step 3: Plan execute
            r3 = runner.invoke(plan_app, ["execute", _PLAN_ULID])
            assert r3.exit_code == 0, f"plan execute failed: {r3.output}"

            # Step 4: Plan apply
            r4 = runner.invoke(plan_app, ["apply", "--yes", _PLAN_ULID])
            assert r4.exit_code == 0, f"plan apply failed: {r4.output}"

            print("cli-lifecycle-full-e2e-ok")

    except AssertionError as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        sys.exit(1)
    finally:
        os.unlink(yaml_path)


def plan_list_namespace() -> None:
    """Verify plan list with namespace regex filters plans by namespace.

    The CLI uses a positional regex argument (e.g. "^myteam/") to filter
    by namespace rather than a --namespace option.
    """
    mock_service = MagicMock()
    # Only plans in "myteam" namespace should be returned by the service
    mock_service.list_plans.return_value = [
        _mock_plan(name="myteam/plan-a", plan_id="01KHDE6WWS2171PWW3GJEBXZ8C"),
    ]

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        # Use regex pattern to filter by namespace (CLI spec: agents plan list
        # "^myteam/")
        result = runner.invoke(plan_app, ["list", "^myteam/"])
        if result.exit_code != 0:
            print(
                f"FAIL: plan list with namespace regex returned {result.exit_code}",
                file=sys.stderr,
            )
            print(result.output, file=sys.stderr)
            sys.exit(1)

        # Verify the plan name appears in the output (may be truncated in table)
        # The table truncates long names, so check for "mytea" prefix
        if "mytea" not in result.output and "plan-a" not in result.output:
            print(
                "FAIL: expected plan name containing 'mytea' or 'plan-a' in output",
                file=sys.stderr,
            )
            print(result.output, file=sys.stderr)
            sys.exit(1)

        print("cli-lifecycle-plan-list-namespace-ok")


def plan_list_namespace_short() -> None:
    """Verify plan list with namespace regex (short form) filters plans by namespace.

    The CLI uses a positional regex argument to filter by namespace.
    This test verifies the regex pattern works as a namespace filter.
    """
    mock_service = MagicMock()
    mock_service.list_plans.return_value = [
        _mock_plan(name="myteam/plan-b", plan_id="01KHDE6WWS2171PWW3GJEBXZ8D"),
    ]

    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        # Use regex pattern to filter by namespace
        result = runner.invoke(plan_app, ["list", "myteam"])
        if result.exit_code != 0:
            print(
                f"FAIL: plan list with namespace regex returned {result.exit_code}",
                file=sys.stderr,
            )
            print(result.output, file=sys.stderr)
            sys.exit(1)

        # Verify the plan name appears in the output (may be truncated in table)
        # The table truncates long names, so check for "mytea" prefix
        if "mytea" not in result.output and "plan-b" not in result.output:
            print(
                "FAIL: expected plan name containing 'mytea' or 'plan-b' in output",
                file=sys.stderr,
            )
            print(result.output, file=sys.stderr)
            sys.exit(1)

        print("cli-lifecycle-plan-list-namespace-short-ok")


# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------

_COMMANDS = {
    "action-create": action_create,
    "plan-use": plan_use,
    "plan-execute": plan_execute,
    "plan-apply": plan_apply,
    "plan-status": plan_status,
    "plan-cancel": plan_cancel,
    "plan-list": plan_list,
    "plan-list-namespace": plan_list_namespace,
    "plan-list-namespace-short": plan_list_namespace_short,
    "full-lifecycle": full_lifecycle,
}


def main() -> None:
    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(2)
    _COMMANDS[sys.argv[1]]()


if __name__ == "__main__":
    main()