cleveragents-core/robot/helper_actor_run_signature.py

"""Helper script for actor_run_signature.robot integration tests.

Tests that the ``actor run`` and ``actor-run`` CLI commands accept
positional NAME and PROMPT arguments while preserving --config fallback.
"""

from __future__ import annotations

import os
import sys
import tempfile
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, 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)

import typer  # noqa: E402
from typer.testing import CliRunner  # noqa: E402

from cleveragents.cli.commands.actor import app as actor_app  # noqa: E402
from cleveragents.cli.commands.actor_run import app as actor_run_app  # noqa: E402

runner = CliRunner()

_SIMPLE_YAML = """\
name: smoke-actor
type: custom
provider: openai
model: gpt-4
tools:
  - operation: identity
"""


def _write_yaml(content: str) -> str:
    fd, path = tempfile.mkstemp(suffix=".yaml")
    with os.fdopen(fd, "w", encoding="utf-8") as fh:
        fh.write(content)
    return path


def _make_app(*, result: str) -> MagicMock:
    app_exec = MagicMock()
    app_exec.config = SimpleNamespace(global_context={})
    app_exec.run_single_shot = AsyncMock(return_value=result)
    # last_run_tool_calls is an int property; set to 0 so comparisons work in tests
    app_exec.last_run_tool_calls = 0
    return app_exec


def positional_with_config() -> None:
    """Test positional NAME + PROMPT with --config fallback."""
    path = _write_yaml(_SIMPLE_YAML)
    app_exec = _make_app(result="config response")

    try:
        with patch(
            "cleveragents.cli.commands.actor.ReactiveCleverAgentsApp",
            return_value=app_exec,
        ):
            result = runner.invoke(
                actor_app,
                ["run", "--config", path, "my-actor", "hello world"],
            )
        if result.exit_code != 0:
            print(
                f"FAIL: expected exit code 0, got {result.exit_code}",
                file=sys.stderr,
            )
            print(result.output, file=sys.stderr)
            sys.exit(1)
        if "config response" not in result.output:
            print(
                f"FAIL: expected 'config response' in output, got: {result.output}",
                file=sys.stderr,
            )
            sys.exit(1)
        print("positional-with-config-ok")
    finally:
        Path(path).unlink(missing_ok=True)


def positional_from_registry() -> None:
    """Test positional NAME resolved from the actor registry."""
    path = _write_yaml(_SIMPLE_YAML)
    app_exec = _make_app(result="registry response")

    def _mock_resolve(name: str, config: list[Any]) -> list[Any]:
        return [Path(path)]

    try:
        with (
            patch(
                "cleveragents.cli.commands.actor_run._resolve_config_files",
                side_effect=_mock_resolve,
            ),
            patch(
                "cleveragents.cli.commands.actor_run.ReactiveCleverAgentsApp",
                return_value=app_exec,
            ),
        ):
            result = runner.invoke(
                actor_run_app,
                ["local/my-actor", "hello from registry"],
            )

        if result.exit_code != 0:
            print(
                f"FAIL: expected exit code 0, got {result.exit_code}",
                file=sys.stderr,
            )
            print(result.output, file=sys.stderr)
            sys.exit(1)
        if "registry response" not in result.output:
            print(
                f"FAIL: expected 'registry response' in output, got: {result.output}",
                file=sys.stderr,
            )
            sys.exit(1)
        print("positional-from-registry-ok")
    finally:
        Path(path).unlink(missing_ok=True)


def actor_app_registry_resolution() -> None:
    """Test positional NAME resolved from the actor registry via actor_app."""
    path = _write_yaml(_SIMPLE_YAML)
    app_exec = _make_app(result="actor-app registry response")

    def _mock_resolve(name: str, config: list[Any]) -> list[Any]:
        return [Path(path)]

    try:
        with (
            patch(
                "cleveragents.cli.commands.actor._resolve_config_files",
                side_effect=_mock_resolve,
            ),
            patch(
                "cleveragents.cli.commands.actor.ReactiveCleverAgentsApp",
                return_value=app_exec,
            ),
        ):
            result = runner.invoke(
                actor_app,
                ["run", "local/my-actor", "hello from actor_app registry"],
            )

        if result.exit_code != 0:
            print(
                f"FAIL: expected exit code 0, got {result.exit_code}",
                file=sys.stderr,
            )
            print(result.output, file=sys.stderr)
            sys.exit(1)
        if "actor-app registry response" not in result.output:
            print(
                "FAIL: expected 'actor-app registry response' "
                f"in output, got: {result.output}",
                file=sys.stderr,
            )
            sys.exit(1)
        print("actor-app-registry-ok")
    finally:
        Path(path).unlink(missing_ok=True)


def unknown_actor_name() -> None:
    """Test error when actor name is not found in registry."""

    def _not_found_resolve(name: str, config: list[Any]) -> list[Any]:
        typer.echo(
            f"Error: Actor '{name}' not found in registry and no --config provided.",
            err=True,
        )
        raise typer.Exit(code=2)

    with patch(
        "cleveragents.cli.commands.actor_run._resolve_config_files",
        side_effect=_not_found_resolve,
    ):
        result = runner.invoke(
            actor_run_app,
            ["nonexistent/actor", "test prompt"],
        )

    if result.exit_code != 2:
        print(
            f"FAIL: expected exit code 2, got {result.exit_code}",
            file=sys.stderr,
        )
        print(result.output, file=sys.stderr)
        sys.exit(1)
    combined = (result.output or "") + (getattr(result, "stderr", None) or "")
    if "not found in registry" not in combined:
        print(
            "FAIL: expected 'not found in registry' in output",
            file=sys.stderr,
        )
        print(combined, file=sys.stderr)
        sys.exit(1)
    print("unknown-actor-name-ok")


def actor_app_unknown_name() -> None:
    """Test error when actor name is not found in registry via actor_app."""

    def _not_found_resolve(name: str, config: list[Any]) -> list[Any]:
        typer.echo(
            f"Error: Actor '{name}' not found in registry and no --config provided.",
            err=True,
        )
        raise typer.Exit(code=2)

    with patch(
        "cleveragents.cli.commands.actor._resolve_config_files",
        side_effect=_not_found_resolve,
    ):
        result = runner.invoke(
            actor_app,
            ["run", "nonexistent/actor", "test prompt"],
        )

    if result.exit_code != 2:
        print(
            f"FAIL: expected exit code 2, got {result.exit_code}",
            file=sys.stderr,
        )
        print(result.output, file=sys.stderr)
        sys.exit(1)
    combined = (result.output or "") + (getattr(result, "stderr", None) or "")
    if "not found in registry" not in combined:
        print(
            "FAIL: expected 'not found in registry' in output",
            file=sys.stderr,
        )
        print(combined, file=sys.stderr)
        sys.exit(1)
    print("actor-app-unknown-name-ok")


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

_COMMANDS: dict[str, Any] = {
    "positional-with-config": positional_with_config,
    "positional-from-registry": positional_from_registry,
    "actor-app-registry": actor_app_registry_resolution,
    "unknown-actor-name": unknown_actor_name,
    "actor-app-unknown-name": actor_app_unknown_name,
}


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