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

tests/test_client.py
274 строки12 KB
pickling-21
Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP
29 июл 2026, 02:18
29 июл 2026, 02:18bc8bde4
"""Порт tests/client/sc-client.test.ts на httpx.MockTransport."""

from __future__ import annotations

import json

import httpx
import pytest

from mcp_sourcecontrol.client import ScApiError, ScNetworkError, SimpleScClient
from tests.conftest import json_response, make_mock_transport, text_response

BASE_URL = "https://api.sc-ci.sber.ru/sc/api/v3"
TENANT_ID = "test-tenant-id"
TOKEN = "test-token"


def make_client(*responses: httpx.Response) -> tuple[SimpleScClient, list[httpx.Request]]:
    transport, captured = make_mock_transport(*responses)
    return SimpleScClient(BASE_URL, TENANT_ID, TOKEN, transport=transport), captured


def body_of(request: httpx.Request) -> dict:
    return json.loads(request.content.decode("utf-8"))


class TestListBranches:
    async def test_get_branches_url(self):
        mock_response = {
            "values": [
                {"displayId": "main", "latestCommit": "abc123"},
                {"displayId": "develop", "latestCommit": "def456"},
            ],
            "isLastPage": True,
        }
        client, captured = make_client(json_response(200, mock_response))
        result = await client.list_branches("my-project", "my-repo", 1, 30)
        assert result == mock_response
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/branches?page=1&limit=30"
        assert captured[0].method == "GET"

    async def test_auth_and_agent_headers(self):
        client, captured = make_client(json_response(200, {}))
        await client.list_branches("p", "r")
        assert captured[0].headers["Authorization"] == f"Bearer {TOKEN}"
        assert captured[0].headers["User-Agent"] == "mcp-sourcecontrol/1.0"
        assert captured[0].headers["Content-Type"] == "application/json"

    async def test_no_auth_header_without_token(self):
        transport, captured = make_mock_transport(json_response(200, {}))
        client = SimpleScClient(BASE_URL, TENANT_ID, None, transport=transport)
        await client.list_branches("p", "r")
        assert "Authorization" not in captured[0].headers


class TestCreateBranch:
    async def test_post_branches(self):
        client, captured = make_client(json_response(201, {"displayId": "feature/new"}))
        result = await client.create_branch("my-project", "my-repo", "feature/new", "parent-commit-sha")
        assert result == {"displayId": "feature/new"}
        assert captured[0].method == "POST"
        assert body_of(captured[0]) == {"new_branch": "feature/new", "parent_commit_SHA": "parent-commit-sha"}


class TestGetBranch:
    async def test_get_branch_url(self):
        client, captured = make_client(json_response(200, {"displayId": "main", "isDefault": True}))
        result = await client.get_branch("my-project", "my-repo", "main")
        assert result == {"displayId": "main", "isDefault": True}
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/branches/main"

    async def test_encodes_branch_name(self):
        client, captured = make_client(json_response(200, {}))
        await client.get_branch("my-project", "my-repo", "feature/my-branch")
        assert "feature%2Fmy-branch" in str(captured[0].url)


class TestDeleteBranch:
    async def test_delete_branch(self):
        client, captured = make_client(json_response(200, {}))
        assert await client.delete_branch("my-project", "my-repo", "feature/old") is None
        assert captured[0].method == "DELETE"
        assert "/branches/feature%2Fold" in str(captured[0].url)

    async def test_handles_204_no_content(self):
        client, captured = make_client(httpx.Response(204))
        assert await client.delete_branch("p", "r", "old") is None


class TestListCommits:
    async def test_get_commits_with_filters(self):
        client, captured = make_client(json_response(200, {"values": [], "isLastPage": True}))
        await client.list_commits("my-project", "my-repo", "main", None, None, "src/index.ts", 1, 30)
        url = str(captured[0].url)
        assert "branch=main" in url
        assert "path=src%2Findex.ts" in url


class TestListPullRequests:
    async def test_get_pulls_with_state(self):
        client, captured = make_client(json_response(200, {"values": [], "isLastPage": True}))
        await client.list_pull_requests("my-project", "my-repo", "OPEN", None, None, None, None, None, 1, 30)
        assert "state=OPEN" in str(captured[0].url)
        assert str(captured[0].url).startswith(f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/pulls?")


class TestCreatePullRequest:
    async def test_post_pulls(self):
        client, captured = make_client(json_response(201, {"id": 42}))
        await client.create_pull_request("my-project", "my-repo", "feature", "main", "My PR", "Description")
        assert body_of(captured[0]) == {"title": "My PR", "head": "feature", "base": "main", "body": "Description"}

    async def test_post_pulls_without_body(self):
        client, captured = make_client(json_response(201, {"id": 43}))
        await client.create_pull_request("my-project", "my-repo", "feature", "main", "My PR")
        assert body_of(captured[0]) == {"title": "My PR", "head": "feature", "base": "main"}


class TestGetPullRequest:
    async def test_get_pull_url(self):
        client, captured = make_client(json_response(200, {"id": 42}))
        await client.get_pull_request("my-project", "my-repo", 42)
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/pulls/42"


class TestMergePullRequest:
    async def test_post_merge(self):
        client, captured = make_client(json_response(200, {"id": 42, "state": "MERGED"}))
        await client.merge_pull_request("my-project", "my-repo", 42, "squash")
        assert captured[0].method == "POST"
        assert str(captured[0].url).endswith("/pulls/42/merge")
        assert body_of(captured[0]) == {"merge_method": "squash"}


class TestDeclinePullRequest:
    async def test_post_decline(self):
        client, captured = make_client(json_response(200, {"id": 42, "state": "DECLINED"}))
        await client.decline_pull_request("my-project", "my-repo", 42)
        assert captured[0].method == "POST"
        assert str(captured[0].url).endswith("/pulls/42/decline")
        assert body_of(captured[0]) == {}


class TestUpdatePullRequest:
    async def test_patch_pull(self):
        client, captured = make_client(json_response(200, {"id": 42}))
        await client.update_pull_request(
            "my-project", "my-repo", 42, title="Updated PR", body="New desc", target_branch="develop"
        )
        assert captured[0].method == "PATCH"
        assert body_of(captured[0]) == {
            "title": "Updated PR",
            "body": "New desc",
            "toRef": {"id": "refs/heads/develop"},
        }


class TestGetPullRequestDiff:
    async def test_get_diff_as_text(self):
        client, captured = make_client(text_response(200, "diff --git a/x b/x"))
        result = await client.get_pull_request_diff("my-project", "my-repo", 42)
        assert result == "diff --git a/x b/x"
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/pulls/42/diff"
        assert captured[0].headers["Accept"] == "text/plain"

    async def test_get_diff_for_file_preserves_path_slashes(self):
        client, captured = make_client(text_response(200, ""))
        await client.get_pull_request_diff("my-project", "my-repo", 42, "src/index.ts")
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/pulls/42/diff/src/index.ts"


class TestPullRequestSubresources:
    async def test_list_files(self):
        client, captured = make_client(json_response(200, {"values": [], "isLastPage": True}))
        await client.list_pull_request_files("my-project", "my-repo", 42, 1, 30)
        url = str(captured[0].url)
        assert "/pulls/42/files" in url
        assert "page=1" in url and "limit=30" in url

    async def test_list_commits(self):
        client, captured = make_client(json_response(200, {"values": [], "isLastPage": True}))
        await client.list_pull_request_commits("my-project", "my-repo", 42)
        assert "/pulls/42/commits" in str(captured[0].url)

    async def test_list_comments(self):
        client, captured = make_client(json_response(200, {"values": [], "isLastPage": True}))
        await client.list_pull_request_comments("my-project", "my-repo", 7)
        assert "/pulls/7/comments" in str(captured[0].url)


class TestAddPullRequestComment:
    async def test_general_comment_without_anchor(self):
        client, captured = make_client(json_response(201, {"id": 1}))
        await client.add_pull_request_comment("my-project", "my-repo", 42, "Nice change!")
        assert body_of(captured[0]) == {"body": "Nice change!"}

    async def test_inline_comment_with_anchor(self):
        client, captured = make_client(json_response(201, {}))
        await client.add_pull_request_comment("my-project", "my-repo", 42, "LGTM", "src/index.ts", 10, "ADDED")
        body = body_of(captured[0])
        assert body["body"] == "LGTM"
        assert body["anchor"] == {
            "filepath": "src/index.ts",
            "file_variant": "TO",
            "diff_type": "EFFECTIVE",
            "line": 10,
            "line_type": "ADDED",
        }

    async def test_removed_line_uses_from_variant(self):
        client, captured = make_client(json_response(201, {}))
        await client.add_pull_request_comment("my-project", "my-repo", 42, "?", "src/index.ts", 5, "REMOVED")
        assert body_of(captured[0])["anchor"]["file_variant"] == "FROM"

    async def test_file_level_comment_without_line(self):
        client, captured = make_client(json_response(201, {}))
        await client.add_pull_request_comment("my-project", "my-repo", 42, "note", "src/index.ts")
        anchor = body_of(captured[0])["anchor"]
        assert "line" not in anchor and "line_type" not in anchor


class TestGetFileContent:
    async def test_get_raw_file(self):
        client, captured = make_client(text_response(200, 'console.log("hello");'))
        result = await client.get_file_content("my-project", "my-repo", "src/index.ts")
        assert result == 'console.log("hello");'
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/raw/src/index.ts"

    async def test_ref_parameter_is_encoded(self):
        client, captured = make_client(text_response(200, ""))
        await client.get_file_content("my-project", "my-repo", "src/index.ts", "refs/heads/main")
        assert "ref=refs%2Fheads%2Fmain" in str(captured[0].url)


class TestListFiles:
    async def test_contents_root(self):
        client, captured = make_client(json_response(200, []))
        await client.list_files("my-project", "my-repo", ".")
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/contents"

    async def test_contents_with_path(self):
        client, captured = make_client(json_response(200, []))
        await client.list_files("my-project", "my-repo", "src/utils")
        assert str(captured[0].url) == f"{BASE_URL}/repos/{TENANT_ID}/my-project/my-repo/contents/src/utils"

    async def test_contents_with_ref(self):
        client, captured = make_client(json_response(200, []))
        await client.list_files("my-project", "my-repo", None, "refs/heads/main")
        assert "ref=refs%2Fheads%2Fmain" in str(captured[0].url)


class TestGetUser:
    async def test_users_path_has_no_tenant_prefix(self):
        client, captured = make_client(json_response(200, {"login": "jdoe"}))
        result = await client.get_user("jdoe")
        assert result == {"login": "jdoe"}
        assert str(captured[0].url) == f"{BASE_URL}/users/jdoe"
        assert TENANT_ID not in str(captured[0].url)


class TestErrors:
    async def test_http_401_raises_api_error(self):
        client, _ = make_client(httpx.Response(401, text="Unauthorized"))
        with pytest.raises(ScApiError, match="HTTP 401"):
            await client.list_branches("p", "r")

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

        client = SimpleScClient(BASE_URL, TENANT_ID, TOKEN, transport=httpx.MockTransport(raise_connect_error))
        with pytest.raises(ScNetworkError, match="Network error") as exc_info:
            await client.list_branches("p", "r")
        assert exc_info.value.code == "ConnectError"