forked from modelcontextprotocol/python-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_server_pagination.py
More file actions
111 lines (89 loc) · 4.02 KB
/
Copy pathtest_server_pagination.py
File metadata and controls
111 lines (89 loc) · 4.02 KB
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import pytest
from mcp.server import Server
from mcp.types import (
ListPromptsRequest,
ListPromptsResult,
ListResourcesRequest,
ListResourcesResult,
ListToolsRequest,
ListToolsResult,
PaginatedRequestParams,
ServerResult,
)
@pytest.mark.anyio
async def test_list_prompts_pagination() -> None:
server = Server("test")
test_cursor = "test-cursor-123"
# Track what request was received
received_request: ListPromptsRequest | None = None
@server.list_prompts()
async def handle_list_prompts(request: ListPromptsRequest) -> ListPromptsResult:
nonlocal received_request
received_request = request
return ListPromptsResult(prompts=[], nextCursor="next")
handler = server.request_handlers[ListPromptsRequest]
# Test: No cursor provided -> handler receives request with None params
request = ListPromptsRequest(method="prompts/list", params=None)
result = await handler(request)
assert received_request is not None
assert received_request.params is None
assert isinstance(result, ServerResult)
# Test: Cursor provided -> handler receives request with cursor in params
request_with_cursor = ListPromptsRequest(method="prompts/list", params=PaginatedRequestParams(cursor=test_cursor))
result2 = await handler(request_with_cursor)
assert received_request is not None
assert received_request.params is not None
assert received_request.params.cursor == test_cursor
assert isinstance(result2, ServerResult)
@pytest.mark.anyio
async def test_list_resources_pagination() -> None:
server = Server("test")
test_cursor = "resource-cursor-456"
# Track what request was received
received_request: ListResourcesRequest | None = None
@server.list_resources()
async def handle_list_resources(request: ListResourcesRequest) -> ListResourcesResult:
nonlocal received_request
received_request = request
return ListResourcesResult(resources=[], nextCursor="next")
handler = server.request_handlers[ListResourcesRequest]
# Test: No cursor provided -> handler receives request with None params
request = ListResourcesRequest(method="resources/list", params=None)
result = await handler(request)
assert received_request is not None
assert received_request.params is None
assert isinstance(result, ServerResult)
# Test: Cursor provided -> handler receives request with cursor in params
request_with_cursor = ListResourcesRequest(
method="resources/list", params=PaginatedRequestParams(cursor=test_cursor)
)
result2 = await handler(request_with_cursor)
assert received_request is not None
assert received_request.params is not None
assert received_request.params.cursor == test_cursor
assert isinstance(result2, ServerResult)
@pytest.mark.anyio
async def test_list_tools_pagination() -> None:
server = Server("test")
test_cursor = "tools-cursor-789"
# Track what request was received
received_request: ListToolsRequest | None = None
@server.list_tools()
async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult:
nonlocal received_request
received_request = request
return ListToolsResult(tools=[], nextCursor="next")
handler = server.request_handlers[ListToolsRequest]
# Test: No cursor provided -> handler receives request with None params
request = ListToolsRequest(method="tools/list", params=None)
result = await handler(request)
assert received_request is not None
assert received_request.params is None
assert isinstance(result, ServerResult)
# Test: Cursor provided -> handler receives request with cursor in params
request_with_cursor = ListToolsRequest(method="tools/list", params=PaginatedRequestParams(cursor=test_cursor))
result2 = await handler(request_with_cursor)
assert received_request is not None
assert received_request.params is not None
assert received_request.params.cursor == test_cursor
assert isinstance(result2, ServerResult)