PMFlow/backend/tests/test_mcp_runtime_service.py

from __future__ import annotations

import json
from types import SimpleNamespace

import httpx
import pytest

from app.services.mcp_config.runtime import DuckDuckGoSearchRuntime, MicrosoftLearnMCPRuntime


def _sse(payload: dict) -> httpx.Response:
    body = f"event: message\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
    return httpx.Response(
        200,
        headers={"content-type": "text/event-stream", "mcp-session-id": "test-session"},
        text=body,
    )


@pytest.mark.asyncio
async def test_microsoft_learn_runtime_probe_initializes_and_lists_tools() -> None:
    calls: list[dict] = []

    def _handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content.decode("utf-8"))
        calls.append(payload)
        if payload["method"] == "initialize":
            return _sse(
                {
                    "jsonrpc": "2.0",
                    "id": payload["id"],
                    "result": {
                        "serverInfo": {"name": "Microsoft Learn MCP Server", "version": "1.0.0"},
                    },
                }
            )
        if payload["method"] == "tools/list":
            assert request.headers["mcp-session-id"] == "test-session"
            return _sse(
                {
                    "jsonrpc": "2.0",
                    "id": payload["id"],
                    "result": {
                        "tools": [
                            {"name": "microsoft_docs_search"},
                            {"name": "microsoft_docs_fetch"},
                        ]
                    },
                }
            )
        raise AssertionError(f"unexpected method: {payload['method']}")

    runtime = MicrosoftLearnMCPRuntime(
        endpoint_url="https://learn.microsoft.com/api/mcp",
        client_factory=lambda: httpx.AsyncClient(transport=httpx.MockTransport(_handler)),
    )

    probe = await runtime.probe()

    assert probe["status"] == "ready"
    assert probe["tool_names"] == ["microsoft_docs_search", "microsoft_docs_fetch"]
    assert [item["method"] for item in calls] == ["initialize", "tools/list"]


@pytest.mark.asyncio
async def test_microsoft_learn_runtime_search_returns_external_evidence() -> None:
    def _handler(request: httpx.Request) -> httpx.Response:
        payload = json.loads(request.content.decode("utf-8"))
        if payload["method"] == "initialize":
            return _sse(
                {
                    "jsonrpc": "2.0",
                    "id": payload["id"],
                    "result": {
                        "serverInfo": {"name": "Microsoft Learn MCP Server", "version": "1.0.0"},
                    },
                }
            )
        if payload["method"] == "tools/call":
            assert payload["params"]["name"] == "microsoft_docs_search"
            assert payload["params"]["arguments"]["query"] == "azure openai quickstart"
            return _sse(
                {
                    "jsonrpc": "2.0",
                    "id": payload["id"],
                    "result": {
                        "content": [
                            {
                                "type": "text",
                                "text": json.dumps(
                                    {
                                        "results": [
                                            {
                                                "title": "Quickstart: Azure OpenAI",
                                                "content": "Use Azure OpenAI with official quickstart guidance.",
                                                "contentUrl": "https://learn.microsoft.com/azure/ai-foundry/openai/chatgpt-quickstart",
                                            }
                                        ]
                                    },
                                    ensure_ascii=False,
                                ),
                            }
                        ]
                    },
                }
            )
        raise AssertionError(f"unexpected method: {payload['method']}")

    runtime = MicrosoftLearnMCPRuntime(
        endpoint_url="https://learn.microsoft.com/api/mcp",
        client_factory=lambda: httpx.AsyncClient(transport=httpx.MockTransport(_handler)),
    )

    evidence = await runtime.search(query="azure openai quickstart", result_limit=3)

    assert len(evidence) == 1
    assert evidence[0].source_type == "mcp_search"
    assert evidence[0].title == "Quickstart: Azure OpenAI"
    assert evidence[0].target_url == "https://learn.microsoft.com/azure/ai-foundry/openai/chatgpt-quickstart"
    assert evidence[0].authority_score >= 0.9


@pytest.mark.asyncio
async def test_duckduckgo_runtime_probe_reports_text_and_news_tools() -> None:
    runtime = DuckDuckGoSearchRuntime(ddgs_factory=lambda: SimpleNamespace())

    probe = await runtime.probe()

    assert probe["status"] == "ready"
    assert probe["tool_names"] == ["ddg-text-search", "ddg-news-search"]


@pytest.mark.asyncio
async def test_duckduckgo_runtime_search_returns_news_evidence() -> None:
    class _FakeDDGS:
        def news(self, *, keywords: str, region: str, safesearch: str, timelimit: str | None, max_results: int):  # noqa: ANN001
            assert keywords == "OpenAI latest news"
            assert region == "wt-wt"
            assert safesearch == "moderate"
            assert timelimit == "d"
            assert max_results == 3
            return [
                {
                    "title": "OpenAI ships new release",
                    "url": "https://news.example/openai-release",
                    "body": "Latest OpenAI release details.",
                    "date": "2026-04-16T10:14:00+00:00",
                    "source": "Example News",
                }
            ]

    runtime = DuckDuckGoSearchRuntime(ddgs_factory=lambda: _FakeDDGS())

    evidence = await runtime.search(query="OpenAI latest news", result_limit=3, query_kind="news")

    assert len(evidence) == 1
    assert evidence[0].source_type == "mcp_search"
    assert evidence[0].source_label == "DuckDuckGo Search"
    assert evidence[0].title == "OpenAI ships new release"
    assert evidence[0].target_url == "https://news.example/openai-release"
    assert evidence[0].raw_metadata["provider"] == "duckduckgo_backend_direct"


@pytest.mark.asyncio
async def test_duckduckgo_runtime_search_falls_back_to_text_when_news_is_rate_limited() -> None:
    class _FakeDDGS:
        def news(self, **kwargs):  # noqa: ANN003
            _ = kwargs
            raise RuntimeError("403 Ratelimit")

        def text(self, *, keywords: str, region: str, safesearch: str, timelimit: str | None, max_results: int):  # noqa: ANN001
            assert keywords == "OpenAI latest news"
            assert region == "wt-wt"
            assert safesearch == "moderate"
            assert timelimit == "w"
            assert max_results == 2
            return [
                {
                    "title": "OpenAI text fallback result",
                    "href": "https://example.com/openai-fallback",
                    "body": "Fallback web result after news rate limit.",
                }
            ]

    runtime = DuckDuckGoSearchRuntime(ddgs_factory=lambda: _FakeDDGS())

    evidence = await runtime.search(query="OpenAI latest news", result_limit=2, query_kind="news")

    assert len(evidence) == 1
    assert evidence[0].title == "OpenAI text fallback result"
    assert evidence[0].target_url == "https://example.com/openai-fallback"
    assert evidence[0].grounded_fields == ["ddg-text-search"]