temp/robot/helper_cli_formats.py

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

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

from __future__ import annotations

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

import yaml

# 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
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
    ProjectLink,
)

runner = CliRunner()
_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
_ENVELOPE_KEYS = frozenset(
    {"command", "status", "exit_code", "data", "timing", "messages"}
)


def _unwrap_envelope(parsed: Any) -> Any:
    """Return the ``data`` field if *parsed* is a spec envelope, else *parsed* as-is."""
    if isinstance(parsed, dict) and _ENVELOPE_KEYS.issubset(parsed.keys()):
        return parsed["data"]
    return parsed


def _mock_action(name: str = "local/fmt-smoke") -> Action:
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="Smoke test action",
        long_description=None,
        definition_of_done="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(name: str = "local/fmt-smoke-plan") -> Plan:
    now = datetime.now()
    return Plan(
        identity=PlanIdentity(plan_id=_ULID),
        namespaced_name=NamespacedName.parse(name),
        description="Smoke plan",
        definition_of_done="Tests pass",
        action_name="local/fmt-smoke",
        phase=PlanPhase.STRATEGIZE,
        processing_state=ProcessingState.QUEUED,
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        project_links=[ProjectLink(project_name="proj-a")],
        timestamps=PlanTimestamps(created_at=now, updated_at=now),
        created_by=None,
        reusable=True,
        read_only=False,
    )


def action_list_json() -> None:
    svc = MagicMock()
    svc.list_actions.return_value = [_mock_action("local/a1"), _mock_action("local/a2")]
    with patch(
        "cleveragents.cli.commands.action._get_lifecycle_service",
        return_value=svc,
    ):
        result = runner.invoke(action_app, ["list", "--format", "json"])
    assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
    parsed = json.loads(result.output.strip())
    # Output is wrapped in spec-required envelope; unwrap the data field
    data = _unwrap_envelope(parsed)
    assert isinstance(data, list) and len(data) == 2
    print("cli-formats-action-list-json-ok")


def action_show_yaml() -> None:
    svc = MagicMock()
    svc.get_action_by_name.return_value = _mock_action()
    with patch(
        "cleveragents.cli.commands.action._get_lifecycle_service",
        return_value=svc,
    ):
        result = runner.invoke(
            action_app, ["show", "local/fmt-smoke", "--format", "yaml"]
        )
    assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
    parsed = yaml.safe_load(result.output.strip())
    # Output is wrapped in spec-required envelope; unwrap the data field
    data = _unwrap_envelope(parsed)
    assert isinstance(data, dict), f"Expected dict, got {type(data)}"
    assert "namespaced_name" in data
    print("cli-formats-action-show-yaml-ok")


def plan_list_json() -> None:
    svc = MagicMock()
    svc.list_plans.return_value = [_mock_plan()]
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=svc,
    ):
        result = runner.invoke(plan_app, ["list", "--format", "json"])
    assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
    parsed = json.loads(result.output.strip())
    # Output is wrapped in spec-required envelope; unwrap the data field
    data = _unwrap_envelope(parsed)
    assert isinstance(data, list)
    assert "plan_id" in data[0]
    print("cli-formats-plan-list-json-ok")


def plan_status_plain() -> None:
    svc = MagicMock()
    svc.get_plan.return_value = _mock_plan()
    with patch(
        "cleveragents.cli.commands.plan._get_lifecycle_service",
        return_value=svc,
    ):
        result = runner.invoke(plan_app, ["status", _ULID, "--format", "plain"])
    assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
    assert "plan_id:" in result.output
    assert "processing_state:" in result.output
    print("cli-formats-plan-status-plain-ok")


_COMMANDS = {
    "action-list-json": action_list_json,
    "action-show-yaml": action_show_yaml,
    "plan-list-json": plan_list_json,
    "plan-status-plain": plan_status_plain,
}

if __name__ == "__main__":
    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]]()