cleveragents-core/robot/helper_builtin_adapter.py

"""Helper script for Robot Framework BuiltinAdapter integration tests."""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

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

from cleveragents.mcp.adapter import MCPToolAdapter
from cleveragents.tool.builtins import BuiltinAdapter
from cleveragents.tool.registry import ToolRegistry


def test_discover() -> None:
    """Verify BuiltinAdapter.discover() returns all built-in tools."""
    adapter = BuiltinAdapter()
    tools = adapter.discover()
    assert len(tools) == 11, f"Expected 11 tools, got {len(tools)}"
    names = {t.name for t in tools}
    assert "builtin/file-read" in names
    assert "builtin/git-status" in names
    assert "builtin/plan-subplan" in names
    # Calling discover again returns same result
    tools2 = adapter.discover()
    assert len(tools2) == 11
    print("discover-ok")


def test_register() -> None:
    """Verify BuiltinAdapter.register() populates a ToolRegistry."""
    adapter = BuiltinAdapter()
    registry = ToolRegistry()
    registered = adapter.register(registry)
    assert len(registered) == 11, f"Expected 11, got {len(registered)}"
    assert registry.get("builtin/file-read") is not None
    assert registry.get("builtin/git-status") is not None
    assert registry.get("builtin/plan-subplan") is not None
    print("register-ok")


def test_source() -> None:
    """Verify all tools registered via BuiltinAdapter have source=builtin."""
    adapter = BuiltinAdapter()
    registry = ToolRegistry()
    adapter.register(registry)
    for spec in registry.list_tools():
        assert spec.source == "builtin", (
            f"Tool '{spec.name}' has source='{spec.source}'"
        )
    print("source-ok")


def test_mcp_file() -> None:
    """Verify MCP resource slot inference for file_path parameter."""
    schema: dict[str, Any] = {
        "type": "object",
        "properties": {
            "file_path": {"type": "string"},
        },
    }
    slots = MCPToolAdapter.infer_resource_slots("test-read-file", schema)
    assert len(slots) == 1, f"Expected 1 slot, got {len(slots)}"
    assert slots[0].name == "file"
    assert slots[0].resource_type == "file"
    assert slots[0].access.value == "read_write"
    assert slots[0].binding.value == "parameter"
    print("mcp-file-ok")


def test_mcp_dir() -> None:
    """Verify MCP resource slot inference for directory parameter."""
    schema: dict[str, Any] = {
        "type": "object",
        "properties": {
            "directory": {"type": "string"},
        },
    }
    slots = MCPToolAdapter.infer_resource_slots("test-list-dir", schema)
    assert len(slots) == 1, f"Expected 1 slot, got {len(slots)}"
    assert slots[0].name == "directory"
    assert slots[0].resource_type == "directory"
    assert slots[0].access.value == "read_only"
    assert slots[0].binding.value == "parameter"
    print("mcp-dir-ok")


def test_mcp_repo() -> None:
    """Verify MCP resource slot inference for repo_path parameter."""
    schema: dict[str, Any] = {
        "type": "object",
        "properties": {
            "repo_path": {"type": "string"},
        },
    }
    slots = MCPToolAdapter.infer_resource_slots("test-git-status", schema)
    assert len(slots) == 1, f"Expected 1 slot, got {len(slots)}"
    assert slots[0].name == "repository"
    assert slots[0].resource_type == "git-checkout"
    assert slots[0].access.value == "read_write"
    assert slots[0].binding.value == "contextual"
    print("mcp-repo-ok")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "discover"
    dispatch: dict[str, Any] = {
        "discover": test_discover,
        "register": test_register,
        "source": test_source,
        "mcp_file": test_mcp_file,
        "mcp_dir": test_mcp_dir,
        "mcp_repo": test_mcp_repo,
    }
    fn = dispatch.get(cmd)
    if fn:
        fn()
    else:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)