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

tests/test_http_headers.py
114 строк5 KB
pickling-21
Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP
29 июл 2026, 02:18
29 июл 2026, 02:18bc8bde4
"""HTTP-режим: переопределение окружения заголовками X-SC-Url/X-SC-Token на каждом запросе."""

from __future__ import annotations

import json

import httpx

import mcp_sourcecontrol.runtime as runtime
from mcp_sourcecontrol.config import ResolvedConfig, resolve_tenant_id
from mcp_sourcecontrol.http_app import build_http_app
from mcp_sourcecontrol.server import create_server
from tests.conftest import json_response, make_mock_transport

DEFAULT_CONFIG = ResolvedConfig(
    api_url="https://api.sc-ci.sber.ru/sc/api/v3",
    tenant_id="default-tenant",
    token="default-token",
    log_level="error",
    tls_reject_unauthorized=False,
)

MCP_HEADERS = {
    "Content-Type": "application/json",
    "Accept": "application/json, text/event-stream",
}


def call_tool_request(name: str, arguments: dict) -> dict:
    return {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {"name": name, "arguments": arguments},
    }


def parse_mcp_response(response: httpx.Response) -> dict:
    """Достаёт JSON-RPC-ответ из JSON- или SSE-ответа Streamable HTTP."""
    content_type = response.headers.get("content-type", "")
    if content_type.startswith("application/json"):
        return response.json()
    for line in response.text.splitlines():
        if line.startswith("data:"):
            return json.loads(line[len("data:") :].strip())
    raise AssertionError(f"no data in response: {response.text!r}")


async def post_call_tool(app, extra_headers: dict | None = None) -> tuple[dict, list[httpx.Request]]:
    """Поднимает lifespan, шлёт tools/call и возвращает (jsonrpc-ответ, запросы к SC API)."""
    transport, captured = make_mock_transport(json_response(200, {"values": [], "isLastPage": True}))
    runtime.set_transport_override(transport)

    inner = app.app  # Starlette-приложение под header-middleware
    async with inner.router.lifespan_context(inner):
        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app), base_url="http://testserver"
        ) as http:
            response = await http.post(
                "/mcp",
                headers={**MCP_HEADERS, **(extra_headers or {})},
                json=call_tool_request("git_list_branches", {"project": "P", "repository": "r"}),
            )
    assert response.status_code == 200, response.text
    return parse_mcp_response(response), captured


class TestHttpHeaders:
    async def test_default_config_without_headers(self):
        server = create_server(DEFAULT_CONFIG)
        app = build_http_app(server, DEFAULT_CONFIG)
        payload, captured = await post_call_tool(app)

        assert payload["result"]["isError"] is False
        url = str(captured[0].url)
        assert url.startswith("https://api.sc-ci.sber.ru/sc/api/v3/repos/default-tenant/")
        assert captured[0].headers["Authorization"] == "Bearer default-token"

    async def test_headers_override_url_and_token(self):
        server = create_server(DEFAULT_CONFIG)
        app = build_http_app(server, DEFAULT_CONFIG)
        payload, captured = await post_call_tool(
            app,
            extra_headers={"X-SC-Url": "https://sc-cd.sber.ru", "X-SC-Token": "cd-token"},
        )

        assert payload["result"]["isError"] is False
        url = str(captured[0].url)
        expected_tenant = resolve_tenant_id("https://sc-cd.sber.ru")
        assert url.startswith(f"https://api.sc-cd.sber.ru/sccd/api/v3/repos/{expected_tenant}/")
        assert captured[0].headers["Authorization"] == "Bearer cd-token"

    async def test_token_only_override_keeps_default_url(self):
        server = create_server(DEFAULT_CONFIG)
        app = build_http_app(server, DEFAULT_CONFIG)
        _, captured = await post_call_tool(app, extra_headers={"X-SC-Token": "personal-token"})

        # URL остаётся дефолтным; tenant при этом пересчитывается из URL —
        # ровно как resolveSessionConfig() в TS-версии
        expected_tenant = resolve_tenant_id(DEFAULT_CONFIG.api_url)
        assert str(captured[0].url).startswith(f"https://api.sc-ci.sber.ru/sc/api/v3/repos/{expected_tenant}/")
        assert captured[0].headers["Authorization"] == "Bearer personal-token"

    async def test_health_endpoint(self):
        server = create_server(DEFAULT_CONFIG)
        app = build_http_app(server, DEFAULT_CONFIG)
        inner = app.app
        async with inner.router.lifespan_context(inner):
            async with httpx.AsyncClient(
                transport=httpx.ASGITransport(app=app), base_url="http://testserver"
            ) as http:
                response = await http.get("/health")
        assert response.status_code == 200
        assert response.json() == {"status": "ok", "mode": "http"}