copilot-sdk/python/test_mcp_prompts_generated.py at main · github/copilot-sdk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
# Copyright (c) Microsoft Corporation. All rights reserved.

"""Generated prompt RPC contracts; real MCP behavior is covered by Node and .NET E2Es."""

import json
from pathlib import Path
from unittest.mock import AsyncMock

import pytest

from copilot.rpc import MCPPromptsGetRequest, MCPPromptsListRequest, SessionRpc


@pytest.fixture
def prompt_fixtures():
    path = Path(__file__).parent.parent / "test" / "harness" / "mcp-prompt-fixtures.json"
    return json.loads(path.read_text(encoding="utf-8"))


@pytest.mark.asyncio
async def test_prompt_list_dispatch_and_optional_fields(prompt_fixtures):
    client = AsyncMock()
    api = SessionRpc(client, "bound-session")
    client.request.return_value = prompt_fixtures["firstPage"]

    first = await api.mcp.prompts.list(MCPPromptsListRequest(server_name="fixture"))

    client.request.assert_awaited_once_with(
        "session.mcp.prompts.list", {"sessionId": "bound-session", "serverName": "fixture"}
    )
    assert first.to_dict() == prompt_fixtures["firstPage"]
    assert [argument.required for argument in first.prompts[0].arguments] == [True, False, None]

    client.request.reset_mock()
    client.request.return_value = prompt_fixtures["secondPage"]
    second = await api.mcp.prompts.list(
        MCPPromptsListRequest(server_name="fixture", cursor=first.next_cursor)
    )
    client.request.assert_awaited_once_with(
        "session.mcp.prompts.list",
        {"sessionId": "bound-session", "serverName": "fixture", "cursor": first.next_cursor},
    )
    assert second.to_dict() == prompt_fixtures["secondPage"]


@pytest.mark.asyncio
@pytest.mark.parametrize("arguments", [None, {}, {"topic": "日本語", "style": ""}])
async def test_prompt_get_dispatch_and_opaque_json(prompt_fixtures, arguments):
    client = AsyncMock()
    client.request.return_value = prompt_fixtures["richPrompt"]
    api = SessionRpc(client, "bound-session")

    result = await api.mcp.prompts.get(
        MCPPromptsGetRequest(server_name="fixture", prompt_name="rich", arguments=arguments)
    )

    expected = {"sessionId": "bound-session", "serverName": "fixture", "promptName": "rich"}
    if arguments is not None:
        expected["arguments"] = arguments
    client.request.assert_awaited_once_with("session.mcp.prompts.get", expected)
    assert result.to_dict() == prompt_fixtures["richPrompt"]


@pytest.mark.asyncio
async def test_prompt_get_propagates_rpc_error():
    client = AsyncMock()
    client.request.side_effect = RuntimeError("Missing required argument: topic")

    with pytest.raises(RuntimeError, match="Missing required argument: topic"):
        await SessionRpc(client, "bound-session").mcp.prompts.get(
            MCPPromptsGetRequest(server_name="fixture", prompt_name="rich")
        )