mcp-sonarqube-proxy/tests/test_proxy.py

"""Tests for the proxy core: discovery, schema preservation, call forwarding."""

from __future__ import annotations

from contextlib import asynccontextmanager
from typing import Any
from unittest.mock import AsyncMock

import pytest
from mcp import types

from mcp_sonarqube_proxy.proxy import (
    DEFAULT_UPSTREAM_URL,
    TokenMissingError,
    build_proxy_server,
    call_tool_via,
    list_tools_via,
    sonarqube_token,
    upstream_session,
    upstream_url,
)


def _full_tool() -> types.Tool:
    """A Tool with every passthrough field populated."""
    return types.Tool(
        name="rich_tool",
        title="Rich Tool",
        description="A tool with full metadata.",
        inputSchema={
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "search query"},
                "limit": {"type": "integer", "minimum": 1, "default": 10},
            },
            "required": ["query"],
            "additionalProperties": False,
        },
        outputSchema={
            "type": "object",
            "properties": {"hits": {"type": "array", "items": {"type": "string"}}},
        },
        annotations=types.ToolAnnotations(
            title="Rich Tool",
            readOnlyHint=True,
            idempotentHint=True,
            openWorldHint=False,
        ),
    )


def _fake_upstream(*, list_tools_return: Any = None, call_tool_return: Any = None) -> AsyncMock:
    upstream = AsyncMock()
    if list_tools_return is not None:
        upstream.list_tools.return_value = list_tools_return
    if call_tool_return is not None:
        upstream.call_tool.return_value = call_tool_return
    return upstream


def test_upstream_url_default(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("SONARQUBE_MCP_URL", raising=False)
    assert upstream_url() == DEFAULT_UPSTREAM_URL


def test_upstream_url_env_override(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("SONARQUBE_MCP_URL", "http://example.test:9000/mcp")
    assert upstream_url() == "http://example.test:9000/mcp"


async def test_list_tools_via_returns_full_tool_objects() -> None:
    tool = _full_tool()
    upstream = _fake_upstream(list_tools_return=types.ListToolsResult(tools=[tool]))

    tools = await list_tools_via(upstream)

    assert len(tools) == 1
    # Round-trip dump must match exactly: schemas, annotations, title all survive.
    assert tools[0].model_dump(by_alias=True) == tool.model_dump(by_alias=True)
    upstream.list_tools.assert_awaited_once()


async def test_list_tools_via_returns_empty_list() -> None:
    upstream = _fake_upstream(list_tools_return=types.ListToolsResult(tools=[]))

    tools = await list_tools_via(upstream)

    assert tools == []


async def test_list_tools_via_propagates_errors() -> None:
    upstream = AsyncMock()
    upstream.list_tools.side_effect = RuntimeError("upstream gone")

    with pytest.raises(RuntimeError, match="upstream gone"):
        await list_tools_via(upstream)


async def test_call_tool_via_forwards_full_result() -> None:
    expected = types.CallToolResult(
        content=[types.TextContent(type="text", text="ok")],
        structuredContent={"score": 42},
        isError=False,
    )
    upstream = _fake_upstream(call_tool_return=expected)

    result = await call_tool_via(upstream, "do_thing", {"x": 1})

    assert result is expected
    upstream.call_tool.assert_awaited_once_with("do_thing", {"x": 1})


async def test_call_tool_via_preserves_is_error_results() -> None:
    expected = types.CallToolResult(
        content=[types.TextContent(type="text", text="boom")],
        isError=True,
    )
    upstream = _fake_upstream(call_tool_return=expected)

    result = await call_tool_via(upstream, "broken", {})

    assert result.isError is True
    assert result.content[0].text == "boom"


async def test_call_tool_via_propagates_errors() -> None:
    upstream = AsyncMock()
    upstream.call_tool.side_effect = ConnectionError("link down")

    with pytest.raises(ConnectionError, match="link down"):
        await call_tool_via(upstream, "anything", {})


def test_build_proxy_server_registers_handlers() -> None:
    upstream = AsyncMock()
    server = build_proxy_server(upstream)

    assert types.ListToolsRequest in server.request_handlers
    assert types.CallToolRequest in server.request_handlers


async def test_build_proxy_server_handler_forwards_tools_through_dispatch() -> None:
    """End-to-end: dispatch a ListToolsRequest through the registered handler."""
    tool = _full_tool()
    upstream = _fake_upstream(list_tools_return=types.ListToolsResult(tools=[tool]))
    server = build_proxy_server(upstream)

    handler = server.request_handlers[types.ListToolsRequest]
    server_result = await handler(types.ListToolsRequest(method="tools/list"))

    inner = server_result.root  # ServerResult wraps a ListToolsResult
    assert isinstance(inner, types.ListToolsResult)
    assert len(inner.tools) == 1
    assert inner.tools[0].model_dump(by_alias=True) == tool.model_dump(by_alias=True)


# --- SONARQUBE_TOKEN handling ------------------------------------------------


def test_sonarqube_token_returns_value(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("SONARQUBE_TOKEN", "my-token-xyz")
    assert sonarqube_token() == "my-token-xyz"


def test_sonarqube_token_strips_whitespace(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("SONARQUBE_TOKEN", "  trimmed-token  ")
    assert sonarqube_token() == "trimmed-token"


def test_sonarqube_token_raises_when_missing(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("SONARQUBE_TOKEN", raising=False)
    with pytest.raises(TokenMissingError):
        sonarqube_token()


def test_sonarqube_token_raises_when_blank(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("SONARQUBE_TOKEN", "   ")
    with pytest.raises(TokenMissingError):
        sonarqube_token()


def test_token_missing_error_message_carries_no_token() -> None:
    msg = str(TokenMissingError())
    assert "SONARQUBE_TOKEN" in msg
    assert "Security" in msg
    # No token material is ever in this message — there is no token to leak.


async def test_upstream_session_attaches_bearer_header(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The token must be forwarded as Authorization: Bearer <token>."""
    monkeypatch.setenv("SONARQUBE_TOKEN", "header-test-token")
    captured: dict[str, Any] = {}
    fake_session = AsyncMock()

    @asynccontextmanager
    async def fake_streamable(url: str, headers: dict[str, str] | None = None, **_: Any):
        captured["url"] = url
        captured["headers"] = headers
        yield ("READ", "WRITE", lambda: None)

    class FakeClientSession:
        def __init__(self, read: Any, write: Any) -> None:
            self.read = read
            self.write = write

        async def __aenter__(self) -> AsyncMock:
            return fake_session

        async def __aexit__(self, *exc: Any) -> None:
            return None

    monkeypatch.setattr("mcp_sonarqube_proxy.proxy.streamablehttp_client", fake_streamable)
    monkeypatch.setattr("mcp_sonarqube_proxy.proxy.ClientSession", FakeClientSession)

    async with upstream_session(url="http://x.example/mcp") as session:
        assert session is fake_session

    assert captured["url"] == "http://x.example/mcp"
    assert captured["headers"] == {"Authorization": "Bearer header-test-token"}
    fake_session.initialize.assert_awaited_once()


async def test_upstream_session_aborts_when_token_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("SONARQUBE_TOKEN", raising=False)

    with pytest.raises(TokenMissingError):
        async with upstream_session(url="http://x.example/mcp"):
            pass