cleveragents-core/robot/helper_plan_cli_spec.py

"""Helper script for plan_cli_spec.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 the local source tree takes priority over any installed copy.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
# Remove any *other* src entries (e.g. /app/src from the parent clone)
sys.path = [_SRC] + [
    p
    for p in sys.path
    if p != _SRC and not (p.endswith("/src") and "cleveragents" not in p.split("/")[-1])
]
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)
# Flush any already-loaded cleveragents modules so they reimport from _SRC
for _mod_name in sorted(sys.modules):
    if _mod_name.startswith("cleveragents"):
        del sys.modules[_mod_name]

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 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"


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


def _mock_plan(
    project_links: list[ProjectLink] | None = None,
    arguments: dict[str, object] | None = None,
) -> Plan:
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=_PLAN_ULID),
        namespaced_name=NamespacedName.parse("local/smoke-plan"),
        description="Smoke test plan",
        definition_of_done="All smoke tests pass",
        action_name="local/smoke-action",
        phase=PlanPhase.STRATEGIZE,
        processing_state=ProcessingState.QUEUED,
        project_links=project_links or [],
        arguments=dict(arguments) if arguments else {},
        arguments_order=list((arguments or {}).keys()),
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        ),
        estimation_actor="openai/gpt-4",
        invariant_actor="openai/gpt-4",
        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),
    )


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


def use_positional_projects() -> None:
    """Verify plan use accepts multiple positional projects."""
    mock_service = MagicMock()
    mock_service.get_action_by_name.return_value = _mock_action()
    mock_service.use_action.return_value = _mock_plan(
        project_links=[
            ProjectLink(project_name="proj-a"),
            ProjectLink(project_name="proj-b"),
        ],
    )
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = runner.invoke(
            plan_app, ["use", "local/smoke-action", "proj-a", "proj-b"]
        )
        if result.exit_code == 0:
            print("plan-cli-use-positional-ok")
        else:
            print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def use_automation_profile() -> None:
    """Verify plan use accepts --automation-profile."""
    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/smoke-action",
                "proj-a",
                "--automation-profile",
                "trusted",
            ],
        )
        if result.exit_code == 0:
            print("plan-cli-use-profile-ok")
        else:
            print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def use_invariant() -> None:
    """Verify plan use accepts repeatable --invariant."""
    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/smoke-action",
                "proj-a",
                "--invariant",
                "No warnings",
                "--invariant",
                "Keep compat",
            ],
        )
        if result.exit_code == 0:
            print("plan-cli-use-invariant-ok")
        else:
            print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def use_actor_overrides() -> None:
    """Verify plan use accepts actor override flags."""
    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/smoke-action",
                "proj-a",
                "--strategy-actor",
                "openai/gpt-4",
                "--execution-actor",
                "anthropic/claude-3",
                "--estimation-actor",
                "openai/gpt-4",
                "--invariant-actor",
                "openai/gpt-4",
            ],
        )
        if result.exit_code == 0:
            print("plan-cli-use-actors-ok")
        else:
            print(f"FAIL: use returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def list_filters() -> None:
    """Verify list accepts filter flags."""
    mock_service = MagicMock()
    mock_service.list_plans.return_value = [
        _mock_plan(project_links=[ProjectLink(project_name="proj-a")]),
    ]
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        r1 = runner.invoke(plan_app, ["list", "--phase", "strategize"])
        r2 = runner.invoke(plan_app, ["list", "--state", "queued"])
        r3 = runner.invoke(plan_app, ["list", "--project", "proj-a"])
        r4 = runner.invoke(plan_app, ["list", "--action", "local/smoke-action"])
        if all(r.exit_code == 0 for r in [r1, r2, r3, r4]):
            print("plan-cli-list-filters-ok")
        else:
            codes = [r1.exit_code, r2.exit_code, r3.exit_code, r4.exit_code]
            print(f"FAIL: list filters exit codes: {codes}", file=sys.stderr)
            sys.exit(1)


def list_columns() -> None:
    """Verify plan list rich output includes Name, Updated, and Invariants columns."""
    mock_service = MagicMock()
    mock_service.list_plans.return_value = [
        _mock_plan(project_links=[ProjectLink(project_name="proj-a")]),
    ]
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        # Use a wide terminal so column headers are not truncated
        wide_runner = CliRunner(mix_stderr=False)
        result = wide_runner.invoke(plan_app, ["list"], env={"COLUMNS": "200"})
        if result.exit_code != 0:
            print(f"FAIL: list returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)
        output = result.output
        # Check for column headers (rich may truncate in narrow terminals,
        # so we check for the first 5 chars which are always rendered)
        required_columns = ["Name", "Updat", "Invar"]
        missing = [col for col in required_columns if col not in output]
        if missing:
            print(f"FAIL: missing columns: {missing}", file=sys.stderr)
            print(output, file=sys.stderr)
            sys.exit(1)
        # Verify the plan name appears in the output
        if "local/smoke-plan" not in output and "local/smok" not in output:
            print("FAIL: plan name not in output", file=sys.stderr)
            print(output, file=sys.stderr)
            sys.exit(1)
        print("plan-cli-list-columns-ok")


def list_regex() -> None:
    """Verify list accepts a positional <REGEX> argument and filters by name."""
    mock_service = MagicMock()
    matching_plan = _mock_plan()  # name: local/smoke-plan, action: local/smoke-action
    now = matching_plan.timestamps.created_at
    non_matching_plan = Plan(
        identity=PlanIdentity(plan_id="01KHDE6WWS2171PWW3GJEBXZ8S"),
        namespaced_name=NamespacedName.parse("local/other-plan"),
        description="Other plan",
        definition_of_done="Done",
        action_name="local/other-action",
        phase=PlanPhase.STRATEGIZE,
        processing_state=ProcessingState.QUEUED,
        project_links=[],
        arguments={},
        arguments_order=[],
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        ),
        estimation_actor="openai/gpt-4",
        invariant_actor="openai/gpt-4",
        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),
    )
    mock_service.list_plans.return_value = [matching_plan, non_matching_plan]
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        # Use JSON format so full names are visible (rich table truncates them)
        result = runner.invoke(plan_app, ["list", "--format", "json", "smoke.*"])
        if result.exit_code != 0:
            print(f"FAIL: list regex returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)
        if "smoke-plan" not in result.output:
            print("FAIL: matching plan not in output", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)
        if "other-plan" in result.output:
            print("FAIL: non-matching plan appeared in output", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)
        print("plan-cli-list-regex-ok")


def status_fields() -> None:
    """Verify plan status renders required fields."""
    mock_service = MagicMock()
    plan = _mock_plan(
        project_links=[ProjectLink(project_name="proj-a", alias="api")],
        arguments={"coverage": 80},
    )
    mock_service.get_plan.return_value = plan
    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:
            output = result.output.lower()
            required = ["action", "phase", "processing state", "projects", "created"]
            missing = [f for f in required if f not in output]
            if missing:
                print(f"FAIL: missing fields: {missing}", file=sys.stderr)
                print(result.output, file=sys.stderr)
                sys.exit(1)
            print("plan-cli-status-fields-ok")
        else:
            print(f"FAIL: status returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


def cancel_reason() -> None:
    """Verify plan cancel accepts --reason."""
    mock_service = MagicMock()
    plan = _mock_plan()
    plan.processing_state = ProcessingState.CANCELLED
    mock_service.cancel_plan.return_value = plan
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=mock_service,
    ):
        result = runner.invoke(
            plan_app, ["cancel", _PLAN_ULID, "--reason", "Requirements changed"]
        )
        if result.exit_code == 0 and "Requirements changed" in result.output:
            print("plan-cli-cancel-reason-ok")
        else:
            print(f"FAIL: cancel returned {result.exit_code}", file=sys.stderr)
            print(result.output, file=sys.stderr)
            sys.exit(1)


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

_COMMANDS = {
    "use-positional-projects": use_positional_projects,
    "use-automation-profile": use_automation_profile,
    "use-invariant": use_invariant,
    "use-actor-overrides": use_actor_overrides,
    "list-filters": list_filters,
    "list-columns": list_columns,
    "list-regex": list_regex,
    "status-fields": status_fields,
    "cancel-reason": cancel_reason,
}


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()