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

tests/test_tools.py
123 строки5 KB
pickling-21
Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP
29 июл 2026, 02:18
29 июл 2026, 02:18bc8bde4
"""Реестр инструментов и поведение обработчиков (аналог tests/tools/*.test.ts)."""

from __future__ import annotations

import json

import mcp_sourcecontrol.runtime as runtime
from mcp_sourcecontrol.tools import TOOL_REGISTRY
from mcp_sourcecontrol.tools.annotations import DESTRUCTIVE, READ_ONLY, WRITE
from tests.conftest import TEST_CONFIG, json_response, make_mock_transport, text_response

EXPECTED_TOOLS = {
    "git_ping": ("read", READ_ONLY),
    "git_list_branches": ("read", READ_ONLY),
    "git_create_branch": ("write", WRITE),
    "git_get_branch": ("read", READ_ONLY),
    "git_delete_branch": ("write", DESTRUCTIVE),
    "git_list_commits": ("read", READ_ONLY),
    "git_list_pull_requests": ("read", READ_ONLY),
    "git_create_pull_request": ("write", WRITE),
    "git_get_pull_request": ("read", READ_ONLY),
    "git_merge_pull_request": ("write", DESTRUCTIVE),
    "git_decline_pull_request": ("write", DESTRUCTIVE),
    "git_update_pull_request": ("write", WRITE),
    "git_get_pull_request_diff": ("read", READ_ONLY),
    "git_list_pull_request_files": ("read", READ_ONLY),
    "git_list_pull_request_commits": ("read", READ_ONLY),
    "git_list_pull_request_comments": ("read", READ_ONLY),
    "git_add_pull_request_comment": ("write", WRITE),
    "git_get_file_content": ("read", READ_ONLY),
    "git_list_files": ("read", READ_ONLY),
    "git_get_user": ("read", READ_ONLY),
}


def tool_by_name(name: str):
    return next(tool for tool in TOOL_REGISTRY if tool.name == name)


def use_transport(*responses):
    """Подключает mock-транспорт и тестовую конфигурацию к runtime."""
    transport, captured = make_mock_transport(*responses)
    runtime.set_default_config(TEST_CONFIG)
    runtime.set_transport_override(transport)
    return captured


class TestRegistry:
    def test_has_20_tools(self):
        assert len(TOOL_REGISTRY) == 20

    def test_names_are_unique(self):
        names = [tool.name for tool in TOOL_REGISTRY]
        assert len(names) == len(set(names))

    def test_expected_names_groups_and_annotations(self):
        assert {tool.name for tool in TOOL_REGISTRY} == set(EXPECTED_TOOLS)
        for tool in TOOL_REGISTRY:
            group, annotations = EXPECTED_TOOLS[tool.name]
            assert tool.group == group, tool.name
            assert tool.annotations == annotations, tool.name

    def test_descriptions_present(self):
        for tool in TOOL_REGISTRY:
            assert tool.description.strip()


class TestHandlers:
    async def test_ping_does_not_touch_api(self):
        captured = use_transport(json_response(500, {}))
        result = await tool_by_name("git_ping").handler()
        assert json.loads(result) == {"status": "OK", "server": "mcp-sourcecontrol", "version": "1.0.0"}
        assert captured == []

    async def test_list_branches_defaults(self):
        captured = use_transport(json_response(200, {"values": [{"displayId": "main"}], "isLastPage": True}))
        result = await tool_by_name("git_list_branches").handler(project="TESTPROJ", repository="testrepo")
        assert "page=1&limit=30" in str(captured[0].url)
        assert "/repos/test-tenant-id/TESTPROJ/testrepo/branches" in str(captured[0].url)
        assert json.loads(result)["values"] == [{"displayId": "main"}]

    async def test_delete_branch_reports_deletion(self):
        captured = use_transport(json_response(200, {}))
        result = await tool_by_name("git_delete_branch").handler(
            project="P", repository="r", branchName="feature/old"
        )
        assert json.loads(result) == {"deleted": True, "branchName": "feature/old"}
        assert captured[0].method == "DELETE"

    async def test_diff_returns_plain_text(self):
        use_transport(text_response(200, "diff --git a/x b/x"))
        result = await tool_by_name("git_get_pull_request_diff").handler(
            project="P", repository="r", pullRequestIndex=42
        )
        assert result == "diff --git a/x b/x"

    async def test_update_pull_request_maps_target_branch(self):
        captured = use_transport(json_response(200, {"id": 42}))
        await tool_by_name("git_update_pull_request").handler(
            project="P", repository="r", pullRequestIndex=42, title="T", targetBranch="develop"
        )
        body = json.loads(captured[0].content)
        assert body == {"title": "T", "toRef": {"id": "refs/heads/develop"}}

    async def test_add_comment_passes_anchor_fields(self):
        captured = use_transport(json_response(201, {}))
        await tool_by_name("git_add_pull_request_comment").handler(
            project="P", repository="r", pullRequestIndex=1, body="LGTM", filePath="a/b.py", line=3, lineType="ADDED"
        )
        anchor = json.loads(captured[0].content)["anchor"]
        assert anchor["filepath"] == "a/b.py"
        assert anchor["line"] == 3

    async def test_list_files_root(self):
        captured = use_transport(json_response(200, []))
        await tool_by_name("git_list_files").handler(project="P", repository="r")
        assert str(captured[0].url).endswith("/contents")

    async def test_get_user(self):
        captured = use_transport(json_response(200, {"login": "jdoe"}))
        result = await tool_by_name("git_get_user").handler(username="jdoe")
        assert json.loads(result) == {"login": "jdoe"}
        assert str(captured[0].url).endswith("/users/jdoe")