apipilot/tests/test_pilot.py

from __future__ import annotations

import pytest

from apipilot.adapters.openai import OpenAIAdapter
from apipilot.errors import ToolNotFoundError
from apipilot.tool import CompositeTool

from .conftest import build_sample_app


@pytest.fixture
async def pilot():
    _, pilot = build_sample_app(add_composite=True, add_deny=True)
    return pilot


@pytest.mark.asyncio
async def test_tools_for_returns_known_tools(pilot):
    tools = await pilot.tools_for()
    names = {t["name"] for t in tools}
    assert "list_items" in names
    assert "get_item" in names
    assert "create_item" in names
    assert "find_cheapest_item" in names


@pytest.mark.asyncio
async def test_tools_for_respects_allow_tool(pilot):
    user = {"id": "bob", "role": "user"}
    tools = await pilot.tools_for(user)
    names = {t["name"] for t in tools}
    assert "list_items" in names
    assert "create_item" not in names


@pytest.mark.asyncio
async def test_execute_endpoint_query_params(pilot):
    res = await pilot.execute(name="list_items", arguments={"category": "active"})
    assert res["success"] is True
    data = res["data"]
    assert data["total"] == 2
    assert {it["id"] for it in data["items"]} == {1, 2}


@pytest.mark.asyncio
async def test_execute_endpoint_path_param(pilot):
    res = await pilot.execute(name="get_item", arguments={"item_id": 1})
    assert res["success"] is True
    assert res["data"]["name"] == "Item A"


@pytest.mark.asyncio
async def test_execute_endpoint_path_param_not_found(pilot):
    res = await pilot.execute(name="get_item", arguments={"item_id": 999})
    assert res["success"] is False
    err = res["error"]
    assert err["code"] == "EXECUTION_ERROR"
    assert err["status_code"] == 404


@pytest.mark.asyncio
async def test_execute_endpoint_body_params(pilot):
    res = await pilot.execute(
        name="create_item",
        arguments={"name": "New Item", "category": "active", "price": 42},
    )
    assert res["success"] is True
    assert res["data"]["name"] == "New Item"
    assert res["data"]["price"] == 42


@pytest.mark.asyncio
async def test_execute_unknown_tool_raises(pilot):
    with pytest.raises(ToolNotFoundError):
        await pilot.execute(name="does_not_exist", arguments={})


@pytest.mark.asyncio
async def test_execute_invalid_arguments_returns_error(pilot):
    res = await pilot.execute(name="get_item", arguments={"item_id": "not-a-number"})
    assert res["success"] is False
    assert res["error"]["code"] == "INVALID_ARGUMENTS"


@pytest.mark.asyncio
async def test_execute_denied_tool_user_not_allowed(pilot):
    user = {"id": "bob", "role": "user"}
    with pytest.raises(ToolNotFoundError):
        await pilot.execute(user=user, name="create_item", arguments={"name": "x"})


@pytest.mark.asyncio
async def test_execute_timeout(pilot):
    res = await pilot.execute(name="get_slow_item", arguments={"item_id": 1})
    assert res["success"] is False
    assert res["error"]["code"] in {"TIMEOUT", "EXECUTION_ERROR"}


@pytest.mark.asyncio
async def test_execute_huge_response_truncates(pilot):
    res = await pilot.execute(name="get_huge_items", arguments={})
    assert res["success"] is True
    assert res["meta"].get("truncated") is True or res["data"].get("_truncated") is True
    import json

    serialized = json.dumps(res["data"])
    assert len(serialized) <= 1024 + 2048


@pytest.mark.asyncio
async def test_composite_tool(pilot):
    res = await pilot.execute(
        name="find_cheapest_item", arguments={"category": "active"}
    )
    assert res["success"] is True
    assert res["data"]["cheapest"]["id"] == 1


@pytest.mark.asyncio
async def test_execute_empty_result_is_none(pilot):
    res = await pilot.execute(name="list_items", arguments={"category": "nothing-here"})
    assert res["success"] is True
    assert res["data"]["total"] == 0
    assert res["data"]["items"] == []


def test_openai_adapter_tools_format():
    definitions = [
        {
            "name": "list_items",
            "description": "Elenca items",
            "parameters": {
                "type": "object",
                "properties": {"category": {"type": "string"}},
                "required": [],
            },
        }
    ]
    tools = OpenAIAdapter.to_tools(definitions)
    assert tools[0]["type"] == "function"
    assert tools[0]["function"]["name"] == "list_items"
    assert tools[0]["function"]["parameters"]["properties"]["category"]["type"] == "string"


def test_openai_adapter_parse_tool_call():
    tc = {
        "id": "call_123",
        "type": "function",
        "function": {"name": "list_items", "arguments": '{"category":"active"}'},
    }
    tc_id, name, args = OpenAIAdapter.parse_tool_call(tc)
    assert tc_id == "call_123"
    assert name == "list_items"
    assert args == {"category": "active"}


def test_openai_adapter_make_tool_result_success():
    from apipilot.types import ExecuteResponse

    resp: ExecuteResponse = {
        "success": True,
        "data": {"items": [1, 2, 3]},
        "error": None,
        "meta": {"tool": "list_items"},
    }
    tr = OpenAIAdapter.make_tool_result("call_1", "list_items", resp)
    assert tr["role"] == "tool"
    assert tr["tool_call_id"] == "call_1"
    assert tr["name"] == "list_items"
    import json

    assert json.loads(tr["content"]) == {"items": [1, 2, 3]}


def test_openai_adapter_make_tool_result_error():
    from apipilot.types import ExecuteResponse

    resp: ExecuteResponse = {
        "success": False,
        "data": None,
        "error": {"code": "X", "message": "fail"},
        "meta": {"tool": "x"},
    }
    tr = OpenAIAdapter.make_tool_result("call_2", "x", resp)
    import json

    assert json.loads(tr["content"]) == {"error": {"code": "X", "message": "fail"}}