aiagent/backend/tests/test_tools_api.py

"""
工具市场 API 集成测试
"""
from __future__ import annotations

import json
import pytest
from unittest.mock import patch, AsyncMock


class TestToolsAPI:
    """工具市场 API 测试"""

    @pytest.mark.api
    def test_list_public_tools(self, authenticated_client):
        resp = authenticated_client.get("/api/v1/tools")
        assert resp.status_code == 200
        assert isinstance(resp.json(), list)

    @pytest.mark.api
    def test_list_categories(self, authenticated_client):
        resp = authenticated_client.get("/api/v1/tools/categories")
        assert resp.status_code == 200
        cats = resp.json()
        assert isinstance(cats, list)
        # 应包含默认分类
        assert "数据处理" in cats or "网络请求" in cats

    @pytest.mark.api
    def test_list_without_auth(self, client):
        resp = client.get("/api/v1/tools")
        # 未认证时默认 scope=public 应返回 401 或公开工具
        assert resp.status_code == 401

    @pytest.mark.api
    def test_list_builtin(self, authenticated_client):
        resp = authenticated_client.get("/api/v1/tools/builtin")
        assert resp.status_code == 200
        tools = resp.json()
        assert isinstance(tools, list)
        assert len(tools) >= 10  # 期待至少 10 个内置工具

    @pytest.mark.api
    def test_create_and_get_tool(self, authenticated_client):
        # 创建 HTTP 工具
        create_resp = authenticated_client.post("/api/v1/tools", json={
            "name": "echo_test",
            "description": "Echo test tool",
            "category": "network",
            "function_schema": {
                "name": "echo_test",
                "description": "Returns the input",
                "parameters": {
                    "type": "object",
                    "properties": {"msg": {"type": "string"}},
                    "required": ["msg"],
                },
            },
            "implementation_type": "http",
            "implementation_config": {
                "url": "https://httpbin.org/post",
                "method": "POST",
                "headers": {},
                "timeout": 10,
            },
            "is_public": True,
        })
        assert create_resp.status_code == 201
        data = create_resp.json()
        assert data["name"] == "echo_test"
        assert data["implementation_type"] == "http"
        tool_id = data["id"]

        # 获取工具详情
        get_resp = authenticated_client.get(f"/api/v1/tools/{tool_id}")
        assert get_resp.status_code == 200
        assert get_resp.json()["name"] == "echo_test"

    @pytest.mark.api
    def test_create_code_tool(self, authenticated_client):
        create_resp = authenticated_client.post("/api/v1/tools", json={
            "name": "double_test",
            "description": "Double a number",
            "category": "math",
            "function_schema": {
                "name": "double_test",
                "description": "Doubles a number",
                "parameters": {
                    "type": "object",
                    "properties": {"n": {"type": "number"}},
                    "required": ["n"],
                },
            },
            "implementation_type": "code",
            "implementation_config": {
                "source": "def run(args):\n    n = args.get('n', 0)\n    return {'result': n * 2}",
                "language": "python",
            },
            "is_public": True,
        })
        assert create_resp.status_code == 201
        data = create_resp.json()
        assert data["name"] == "double_test"
        assert data["implementation_type"] == "code"

    @pytest.mark.api
    def test_create_duplicate_name(self, authenticated_client):
        # 先创建
        authenticated_client.post("/api/v1/tools", json={
            "name": "dup_tool",
            "description": "Dup",
            "function_schema": {"name": "dup_tool", "parameters": {"type": "object", "properties": {}}},
            "implementation_type": "code",
            "implementation_config": {"source": "def run(args):\n    return {}"},
            "is_public": False,
        })
        # 重复创建应报错
        resp = authenticated_client.post("/api/v1/tools", json={
            "name": "dup_tool",
            "description": "Dup again",
            "function_schema": {"name": "dup_tool", "parameters": {"type": "object", "properties": {}}},
            "implementation_type": "code",
            "implementation_config": {"source": "def run(args):\n    return {}"},
            "is_public": False,
        })
        assert resp.status_code == 400
        assert "已存在" in resp.json().get("detail", "")

    @pytest.mark.api
    def test_invalid_implementation_type(self, authenticated_client):
        resp = authenticated_client.post("/api/v1/tools", json={
            "name": "bad_tool",
            "description": "Bad",
            "function_schema": {"name": "bad_tool", "parameters": {"type": "object", "properties": {}}},
            "implementation_type": "invalid_type",
            "is_public": False,
        })
        assert resp.status_code == 400

    @pytest.mark.api
    def test_mine_scope(self, authenticated_client):
        resp = authenticated_client.get("/api/v1/tools?scope=mine")
        assert resp.status_code == 200

    @pytest.mark.api
    def test_get_nonexistent_tool(self, authenticated_client):
        resp = authenticated_client.get("/api/v1/tools/nonexistent-id")
        assert resp.status_code == 404

    @pytest.mark.api
    @pytest.mark.asyncio
    async def test_test_http_endpoint(self, authenticated_client):
        """测试 HTTP 工具的测试端点"""
        with patch("httpx.AsyncClient.request", new=AsyncMock()) as mock_request:
            mock_response = AsyncMock()
            mock_response.status_code = 200
            mock_response.text = '{"origin": "1.2.3.4"}'
            mock_request.return_value = mock_response

            resp = authenticated_client.post("/api/v1/tools/test/http", json={
                "url": "https://httpbin.org/get",
                "method": "GET",
                "headers": {},
                "body": None,
                "args": {"ip": "8.8.8.8"},
                "timeout": 5,
            })
            assert resp.status_code == 200
            data = resp.json()
            assert data["success"] is True

    @pytest.mark.api
    def test_test_code_endpoint(self, authenticated_client):
        resp = authenticated_client.post("/api/v1/tools/test/code", json={
            "source": "def run(args):\n    return args.get('x', 0) + args.get('y', 0)",
            "args": {"x": 10, "y": 20},
        })
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is True
        assert data["result"] == 30

    @pytest.mark.api
    def test_test_code_compile_error(self, authenticated_client):
        resp = authenticated_client.post("/api/v1/tools/test/code", json={
            "source": "invalid python {{{",
            "args": {},
        })
        assert resp.status_code == 200
        data = resp.json()
        assert data["success"] is False
        assert "error" in data

    @pytest.mark.api
    def test_use_count(self, authenticated_client):
        # 先创建工具
        create_resp = authenticated_client.post("/api/v1/tools", json={
            "name": "count_test",
            "description": "Count test",
            "function_schema": {"name": "count_test", "parameters": {"type": "object", "properties": {}}},
            "implementation_type": "code",
            "implementation_config": {"source": "def run(args):\n    return {}"},
            "is_public": True,
        })
        assert create_resp.status_code == 201
        tool_id = create_resp.json()["id"]

        # 使用计数
        use_resp = authenticated_client.post(f"/api/v1/tools/{tool_id}/use")
        assert use_resp.status_code == 200
        assert use_resp.json()["use_count"] == 1

        # 再次使用
        use_resp2 = authenticated_client.post(f"/api/v1/tools/{tool_id}/use")
        assert use_resp2.status_code == 200
        assert use_resp2.json()["use_count"] == 2

    @pytest.mark.api
    def test_delete_tool(self, authenticated_client):
        # 创建
        create_resp = authenticated_client.post("/api/v1/tools", json={
            "name": "del_test",
            "description": "Delete test",
            "function_schema": {"name": "del_test", "parameters": {"type": "object", "properties": {}}},
            "implementation_type": "code",
            "implementation_config": {"source": "def run(args):\n    return {}"},
            "is_public": False,
        })
        assert create_resp.status_code == 201
        tool_id = create_resp.json()["id"]

        # 删除
        del_resp = authenticated_client.delete(f"/api/v1/tools/{tool_id}")
        assert del_resp.status_code == 200

        # 确认已删除
        get_resp = authenticated_client.get(f"/api/v1/tools/{tool_id}")
        assert get_resp.status_code == 404

    @pytest.mark.api
    def test_update_tool(self, authenticated_client):
        create_resp = authenticated_client.post("/api/v1/tools", json={
            "name": "update_test",
            "description": "Original desc",
            "function_schema": {"name": "update_test", "parameters": {"type": "object", "properties": {}}},
            "implementation_type": "code",
            "implementation_config": {"source": "def run(args):\n    return {}"},
            "is_public": False,
        })
        assert create_resp.status_code == 201
        tool_id = create_resp.json()["id"]

        update_resp = authenticated_client.put(f"/api/v1/tools/{tool_id}", json={
            "name": "update_test",
            "description": "Updated desc",
            "function_schema": {"name": "update_test", "parameters": {"type": "object", "properties": {}}},
            "implementation_type": "code",
            "implementation_config": {"source": "def run(args):\n    return {'updated': True}"},
            "is_public": True,
        })
        assert update_resp.status_code == 200
        assert update_resp.json()["description"] == "Updated desc"
        assert update_resp.json()["is_public"] is True