tests/test_toolservers.py

"""Testy externích tool serverů (MCP klient + manager + integrace)."""

import json

import pytest
from fastapi.testclient import TestClient

from tests.mock_mcp import MockMCPServer


@pytest.fixture()
def mcp():
    server = MockMCPServer()
    yield server
    server.stop()


@pytest.fixture()
def mcp_sse():
    server = MockMCPServer(sse=True)
    yield server
    server.stop()


@pytest.fixture()
def ts_module(env):
    import importlib
    import sys
    for name in ("presets", "settings", "toolservers"):
        sys.modules.pop(name, None)
    return importlib.import_module("toolservers")


# ---------------------------------------------------------------- parsování

def test_parse_sse_response(ts_module):
    body = ('event: message\n'
            'data: {"jsonrpc":"2.0","id":1,"result":{"ok":true}}\n\n')
    data = ts_module.parse_mcp_response("text/event-stream", body)
    assert data["result"] == {"ok": True}
    data = ts_module.parse_mcp_response("application/json",
                                        '{"jsonrpc":"2.0","id":1,"result":{}}')
    assert data["result"] == {}
    with pytest.raises(ValueError):
        ts_module.parse_mcp_response("text/event-stream", "data: nevalidni\n")


def test_parse_registry_payload(ts_module):
    payload = {"servers": [
        {"server": {"name": "a/b", "description": "x" * 300, "version": "1.0",
                    "remotes": [{"type": "streamable-http", "url": "https://a/mcp"}]}},
        {"server": {"name": "c/d", "description": "local only",
                    "packages": [{"registryType": "npm"}]}},
    ]}
    out = ts_module.parse_registry_payload(payload)
    assert out[0]["urls"] == ["https://a/mcp"] and out[0]["remote"] is True
    assert len(out[0]["description"]) == 200
    assert out[1]["remote"] is False


# ---------------------------------------------------------------- manager

def test_discover_and_call(ts_module, mcp):
    mgr = ts_module.ToolServerManager()
    mgr.add_or_update("Mock-Server", mcp.url)
    servers = mgr.list()
    assert servers[0]["name"] == "mock_server"
    assert servers[0]["status"] == "ok"
    assert servers[0]["tool_names"] == ["echo", "add"]

    tools = mgr.openai_tools()
    names = [t["function"]["name"] for t in tools]
    assert names == ["ext_mock_server_echo", "ext_mock_server_add"]
    assert tools[0]["function"]["parameters"]["required"] == ["text"]
    assert "[externí nástroj" in tools[0]["function"]["description"]

    res = mgr.call("ext_mock_server_echo", {"text": "ahoj"})
    assert res == {"result": "echo:ahoj"}
    res = mgr.call("ext_mock_server_add", {"a": 2, "b": 3})
    assert res["structured"] == {"sum": 5}


def test_sse_transport(ts_module, mcp_sse):
    mgr = ts_module.ToolServerManager()
    mgr.add_or_update("ssemock", mcp_sse.url)
    assert mgr.list()[0]["status"] == "ok"
    assert mgr.call("ext_ssemock_echo", {"text": "x"}) == {"result": "echo:x"}


def test_auth_token(ts_module):
    server = MockMCPServer(token="tajny")
    try:
        mgr = ts_module.ToolServerManager()
        mgr.add_or_update("secured", server.url)          # bez tokenu
        assert mgr.list()[0]["status"] == "error"
        mgr.add_or_update("secured", server.url, token="tajny")
        assert mgr.list()[0]["status"] == "ok"
        # maskovaný token z UI nepřepíše skutečný
        mgr.add_or_update("secured", server.url, token="********")
        assert mgr.list()[0]["status"] == "ok"
    finally:
        server.stop()


def test_allowed_tools_filter_and_disable(ts_module, mcp):
    mgr = ts_module.ToolServerManager()
    mgr.add_or_update("mock", mcp.url, allowed_tools="echo")
    names = [t["function"]["name"] for t in mgr.openai_tools()]
    assert names == ["ext_mock_echo"]
    assert "error" in mgr.call("ext_mock_add", {"a": 1, "b": 1})

    mgr.add_or_update("mock", mcp.url, enabled=False, refresh=False)
    assert mgr.openai_tools() == []
    assert "error" in mgr.call("ext_mock_echo", {"text": "x"})


def test_tool_error_result(ts_module, mcp):
    mgr = ts_module.ToolServerManager()
    mgr.add_or_update("mock", mcp.url)
    # ruční route na neexistující nástroj serveru
    mgr._routing["ext_mock_ghost"] = ("mock", "ghost")
    res = mgr.call("ext_mock_ghost", {})
    assert "error" in res and "unknown tool" in res["error"]


def test_persistence_roundtrip(ts_module, mcp):
    mgr = ts_module.ToolServerManager()
    mgr.add_or_update("mock", mcp.url, token="tajny", allowed_tools="echo")
    mgr2 = ts_module.ToolServerManager(path=mgr.path)
    row = mgr2.list()[0]
    assert row["name"] == "mock" and row["allowed_tools"] == "echo"
    assert row["token"] == "********"                    # maskováno v list()
    assert row["tool_names"] == ["echo", "add"]          # cache nástrojů přežila
    # routing funguje i po restartu (lazy klient si udělá initialize)
    assert mgr2.call("ext_mock_echo", {"text": "hi"}) == {"result": "echo:hi"}


def test_validation(ts_module):
    mgr = ts_module.ToolServerManager()
    with pytest.raises(ValueError):
        mgr.add_or_update("x", "ftp://spatne")
    with pytest.raises(ValueError):
        mgr.add_or_update("###", "https://ok/mcp", refresh=False)


# ---------------------------------------------------------------- integrace

def test_exec_tool_routing_and_role_guard(app_module, mcp):
    app_module.TOOL_SERVERS.add_or_update("mock", mcp.url)
    # hlavní agent (má "external") smí
    res = app_module._exec_tool("ext_mock_echo", {"text": "hej"},
                                app_module.MAIN_ALLOWED)
    assert res == {"result": "echo:hej"}
    # sub-agent bez "external" nesmí
    res = app_module._exec_tool("ext_mock_echo", {"text": "hej"},
                                app_module.READONLY)
    assert "nejsou v této roli povoleny" in res["error"]
    # nástroje se objeví v hlavních tools
    names = [t["function"]["name"] for t in app_module.build_main_tools()]
    assert "ext_mock_echo" in names and "ext_mock_add" in names


def test_admin_api_crud(app_module, auth, mcp):
    c = TestClient(app_module.app)
    assert c.get("/admin/toolservers").status_code == 401

    r = c.post("/admin/toolservers", headers=auth,
               json={"name": "mock", "url": mcp.url})
    assert r.status_code == 200
    assert r.json()["summary"]["tools"] == 2

    r = c.get("/admin/toolservers", headers=auth)
    assert r.json()["servers"][0]["status"] == "ok"

    r = c.post("/admin/toolservers/refresh", headers=auth, json={"name": "mock"})
    assert "ok" in r.json()["results"]["mock"]

    r = c.get("/admin/toolservers/catalog", headers=auth)
    assert any(s["name"] == "context7" for s in r.json()["servers"])

    r = c.post("/admin/toolservers", headers=auth,
               json={"name": "bad", "url": "ftp://x"})
    assert r.status_code == 400

    r = c.post("/admin/toolservers/delete", headers=auth, json={"name": "mock"})
    assert r.status_code == 200
    assert c.get("/admin/toolservers", headers=auth).json()["servers"] == []


def test_health_reports_toolservers(app_module, auth, mcp):
    app_module.TOOL_SERVERS.add_or_update("mock", mcp.url)
    c = TestClient(app_module.app)
    data = c.get("/health").json()
    assert data["tool_servers"] == {"servers": 1, "enabled": 1, "tools": 2}