tests/test_server.py - master - pickling/mcp-sourcecontrol-python

tests/test_server.py
93 строки4 KB
pickling-21
Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP
29 июл 2026, 02:18
29 июл 2026, 02:18bc8bde4
"""Интеграция через FastMCP: регистрация, схемы, вызовы, ошибки (аналог tests/server.test.ts)."""

from __future__ import annotations

import json

import httpx
from mcp.shared.memory import create_connected_server_and_client_session

import mcp_sourcecontrol.runtime as runtime
from mcp_sourcecontrol.server import create_server
from mcp_sourcecontrol.tools import TOOL_REGISTRY
from tests.conftest import TEST_CONFIG, json_response, make_mock_transport


class TestCreateServer:
    async def test_registers_all_tools(self):
        server = create_server(TEST_CONFIG)
        tools = await server.list_tools()
        assert {tool.name for tool in tools} == {tool.name for tool in TOOL_REGISTRY}

    async def test_schema_matches_ts_version(self):
        server = create_server(TEST_CONFIG)
        tools = {tool.name: tool for tool in await server.list_tools()}

        branches_schema = tools["git_list_branches"].inputSchema
        assert set(branches_schema["properties"]) == {"project", "repository", "page", "limit"}
        assert branches_schema["required"] == ["project", "repository"]
        assert branches_schema["properties"]["limit"]["maximum"] == 100

        merge_schema = tools["git_merge_pull_request"].inputSchema
        assert merge_schema["properties"]["mergeMethod"]["enum"] == [
            "merge",
            "squash",
            "rebase",
            "rebase-merge",
            "manually-merged",
        ]

        assert tools["git_ping"].annotations.readOnlyHint is True
        assert tools["git_delete_branch"].annotations.destructiveHint is True

    async def test_works_without_token(self):
        config = TEST_CONFIG.__class__(**{**TEST_CONFIG.__dict__, "token": None})
        server = create_server(config)
        assert len(await server.list_tools()) == 20


class TestCallToolOverWire:
    async def test_ping_over_memory_session(self):
        server = create_server(TEST_CONFIG)
        async with create_connected_server_and_client_session(server._mcp_server) as client:
            result = await client.call_tool("git_ping", {})
        assert result.isError is False
        payload = json.loads(result.content[0].text)
        assert payload["status"] == "OK"
        assert payload["server"] == "mcp-sourcecontrol"

    async def test_list_branches_over_memory_session(self):
        server = create_server(TEST_CONFIG)
        transport, captured = make_mock_transport(json_response(200, {"values": [], "isLastPage": True}))
        runtime.set_transport_override(transport)
        async with create_connected_server_and_client_session(server._mcp_server) as client:
            result = await client.call_tool("git_list_branches", {"project": "P", "repository": "REPO"})
        assert result.isError is False
        # slug репозитория приводится к нижнему регистру (как .toLowerCase() в zod-схеме)
        assert "/repos/test-tenant-id/P/repo/branches" in str(captured[0].url)

    async def test_api_error_becomes_tool_error(self):
        server = create_server(TEST_CONFIG)
        transport, _ = make_mock_transport(httpx.Response(401, text="Unauthorized"))
        runtime.set_transport_override(transport)
        async with create_connected_server_and_client_session(server._mcp_server) as client:
            result = await client.call_tool("git_list_branches", {"project": "P", "repository": "r"})
        assert result.isError is True
        assert "HTTP 401: Unauthorized" in result.content[0].text

    async def test_network_error_becomes_unreachable_message(self):
        def raise_connect_error(request: httpx.Request) -> httpx.Response:
            raise httpx.ConnectError("refused", request=request)

        server = create_server(TEST_CONFIG)
        runtime.set_transport_override(httpx.MockTransport(raise_connect_error))
        async with create_connected_server_and_client_session(server._mcp_server) as client:
            result = await client.call_tool("git_list_branches", {"project": "P", "repository": "r"})
        assert result.isError is True
        assert "SourceControl unreachable (ConnectError)." in result.content[0].text

    async def test_validation_error_is_reported(self):
        server = create_server(TEST_CONFIG)
        async with create_connected_server_and_client_session(server._mcp_server) as client:
            result = await client.call_tool("git_list_branches", {"project": "", "repository": "r"})
        assert result.isError is True