cleveragents-core/robot/helper_actor_remove_cli.py

"""Helper script for Robot integration tests covering ``actor remove --format`` output.

Exercises the real ``agents`` CLI via subprocess — no mocking of any kind.
A test actor is seeded via ``agents actor add``, then removed via
``agents actor remove --format json``, and the resulting JSON envelope is
validated against the spec.

Usage::

    python helper_actor_remove_cli.py <command>

Where *command* is one of:

- ``remove-json`` — seed an actor, remove it with ``--format json``, validate envelope
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path

# Ensure src is importable when run from workspace root
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

# Ensure robot/ is on the import path for helper_e2e_common.
_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
    sys.path.insert(0, _ROBOT)

from helper_e2e_common import cleanup_workspace, run_cli, setup_workspace  # noqa: E402

_ACTOR_NAME = "local/robot-remove-actor"

_ACTOR_CONFIG: dict[str, object] = {
    "name": _ACTOR_NAME,
    "provider": "openai",
    "model": "gpt-4",
}


def _write_actor_config(workspace: str) -> str:
    """Write actor config JSON to a temp file and return its path."""
    config_path = os.path.join(workspace, "robot_remove_actor.json")
    with open(config_path, "w", encoding="utf-8") as fh:
        json.dump(_ACTOR_CONFIG, fh)
    return config_path


def test_remove_format_json() -> None:
    """Seed an actor via the real CLI, remove it with ``--format json``.

    Validates the resulting JSON envelope against the spec.
    """
    workspace = setup_workspace(prefix="robot_actor_remove_")
    try:
        config_path = _write_actor_config(workspace)

        # Step 1: Add the actor using the real CLI.
        add_result = run_cli(
            "actor",
            "add",
            _ACTOR_NAME,
            "--config",
            config_path,
            workspace=workspace,
        )
        assert add_result.returncode == 0, (
            f"actor add failed (rc={add_result.returncode}):\n"
            f"stdout: {add_result.stdout}\nstderr: {add_result.stderr}"
        )

        # Step 2: Remove the actor with --format json using the real CLI.
        remove_result = run_cli(
            "actor",
            "remove",
            _ACTOR_NAME,
            "--format",
            "json",
            workspace=workspace,
        )
        assert remove_result.returncode == 0, (
            f"actor remove --format json failed (rc={remove_result.returncode}):\n"
            f"stdout: {remove_result.stdout}\nstderr: {remove_result.stderr}"
        )

        # Step 3: Parse and validate the JSON envelope.
        output = remove_result.stdout.strip()
        assert output, (
            f"actor remove --format json produced no output.\n"
            f"stderr: {remove_result.stderr}"
        )

        payload = json.loads(output)

        assert payload["command"] == f"agents actor remove {_ACTOR_NAME}", (
            f"Unexpected command field: {payload.get('command')!r}"
        )
        assert payload["status"] == "ok", (
            f"Unexpected status: {payload.get('status')!r}"
        )
        assert payload["exit_code"] == 0, (
            f"Unexpected exit_code: {payload.get('exit_code')!r}"
        )

        data = payload["data"]

        removed = data.get("actor_removed", {})
        assert removed.get("name") == _ACTOR_NAME, (
            f"actor_removed.name mismatch: {removed.get('name')!r}"
        )
        assert removed.get("provider") == _ACTOR_CONFIG["provider"], (
            f"actor_removed.provider mismatch: {removed.get('provider')!r}"
        )
        assert removed.get("model") == _ACTOR_CONFIG["model"], (
            f"actor_removed.model mismatch: {removed.get('model')!r}"
        )

        impact = data.get("impact", {})
        assert "sessions" in impact, f"Missing 'sessions' in impact: {impact}"
        assert "active_plans" in impact, f"Missing 'active_plans' in impact: {impact}"
        assert "actions_referencing" in impact, (
            f"Missing 'actions_referencing' in impact: {impact}"
        )

        cleanup = data.get("cleanup", {})
        assert cleanup.get("config") == "kept on disk", (
            f"cleanup.config mismatch: {cleanup.get('config')!r}"
        )
        assert "contexts" in cleanup, f"Missing 'contexts' in cleanup: {cleanup}"

        messages = payload.get("messages", [])
        assert messages, f"Expected non-empty messages list, got: {messages}"
        assert messages[0].get("level") == "ok", (
            f"Unexpected message level: {messages[0].get('level')!r}"
        )
        assert "Actor removed" in messages[0].get("text", ""), (
            f"Expected 'Actor removed' in message text: {messages[0].get('text')!r}"
        )

        print("actor-remove-json-format-ok")

    finally:
        cleanup_workspace(workspace)


def main() -> None:
    command = sys.argv[1] if len(sys.argv) > 1 else "remove-json"
    dispatch: dict[str, object] = {
        "remove-json": test_remove_format_json,
    }
    handler = dispatch.get(command)
    if handler is None:
        print(f"Unknown command: {command}", file=sys.stderr)
        sys.exit(1)
    assert callable(handler)
    handler()


if __name__ == "__main__":
    main()