cleveragents-core/robot/helper_cli_extensions.py

"""Helper script for cli_extensions.robot smoke tests.

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

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.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
    AutomationProfileProvenance,
    AutomationProfileRef,
    InvariantSource,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanInvariant,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
)

runner = CliRunner()

_PLAN_ULID = "01KHDE6WWS2171PWW3GJEBXZ8S"


def _mock_plan(
    *,
    automation_profile: AutomationProfileRef | None = None,
    invariants: list[PlanInvariant] | None = None,
) -> Plan:
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=_PLAN_ULID),
        namespaced_name=NamespacedName.parse("local/test-plan"),
        description="Test plan for robot",
        definition_of_done="All tests pass",
        action_name="local/test-action",
        phase=PlanPhase.STRATEGIZE,
        processing_state=ProcessingState.QUEUED,
        project_links=[ProjectLink(project_name="my-project")],
        arguments={},
        arguments_order=[],
        automation_profile=automation_profile,
        invariants=invariants or [],
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        timestamps=PlanTimestamps(created_at=now, updated_at=now),
    )


def _mock_action() -> Action:
    return Action(
        namespaced_name=NamespacedName.parse("local/test-action"),
        description="Test action for robot",
        definition_of_done="All tests pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        state=ActionState.AVAILABLE,
        created_at=datetime.now(),
        updated_at=datetime.now(),
    )


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


def plan_use_with_invariants() -> None:
    """Verify plan use with --invariant flags works."""
    mock_svc = MagicMock()
    mock_svc.get_action_by_name.return_value = _mock_action()
    mock_svc.use_action.return_value = _mock_plan(
        invariants=[
            PlanInvariant(text="No warnings", source=InvariantSource.PLAN),
        ]
    )
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            plan_app,
            [
                "use",
                "local/test-action",
                "--invariant",
                "No warnings",
            ],
        )
    if result.exit_code == 0:
        print("cli-ext-plan-use-invariants-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_use_with_automation_profile() -> None:
    """Verify plan use with --automation-profile works."""
    mock_svc = MagicMock()
    mock_svc.get_action_by_name.return_value = _mock_action()
    mock_svc.use_action.return_value = _mock_plan(
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        )
    )
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            plan_app,
            [
                "use",
                "local/test-action",
                "--automation-profile",
                "trusted",
            ],
        )
    if result.exit_code == 0 and "trusted" in result.output:
        print("cli-ext-plan-use-profile-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_use_actor_validation_valid() -> None:
    """Verify valid namespaced actors are accepted."""
    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/test-action",
                "--strategy-actor",
                "openai/gpt-4",
            ],
        )
    if result.exit_code == 0:
        print("cli-ext-actor-valid-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


def plan_use_actor_validation_invalid() -> None:
    """Verify invalid actor format is rejected."""
    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/test-action",
                "--strategy-actor",
                "bad-format",
            ],
        )
    if result.exit_code != 0:
        print("cli-ext-actor-invalid-ok")
    else:
        print(f"FAIL: expected non-zero exit, got {result.exit_code}")
        sys.exit(1)


def plan_use_combined() -> None:
    """Verify plan use with profile + invariants together."""
    mock_svc = MagicMock()
    mock_svc.get_action_by_name.return_value = _mock_action()
    mock_svc.use_action.return_value = _mock_plan(
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        ),
        invariants=[
            PlanInvariant(text="No regressions", source=InvariantSource.PLAN),
        ],
    )
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_svc,
    ):
        result = runner.invoke(
            plan_app,
            [
                "use",
                "local/test-action",
                "--automation-profile",
                "trusted",
                "--invariant",
                "No regressions",
            ],
        )
    if result.exit_code == 0 and "trusted" in result.output:
        print("cli-ext-plan-combined-ok")
    else:
        print(f"FAIL: exit={result.exit_code} output={result.output}")
        sys.exit(1)


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

_COMMANDS: dict[str, object] = {
    "plan-use-invariants": plan_use_with_invariants,
    "plan-use-profile": plan_use_with_automation_profile,
    "actor-valid": plan_use_actor_validation_valid,
    "actor-invalid": plan_use_actor_validation_invalid,
    "plan-combined": plan_use_combined,
}

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]