modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix flaky test test_88_random_error (#1171)

Browse source
This commit is contained in:
Inna Harper 2025-07-18 16:21:33 +01:00 • committed by GitHub
commit 0b1b52ba45
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 5 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

19
tests/issues/test_88_random_error.py
View file

@ -28,8 +28,7 @@ async def test_notification_validation_error(tmp_path: Path):
server = Server(name="test") server = Server(name="test")
request_count = 0 request_count = 0
slow_request_started = anyio.Event() slow_request_lock = anyio.Event()
slow_request_complete = anyio.Event()
@server.list_tools() @server.list_tools()
async def list_tools() -> list[types.Tool]: async def list_tools() -> list[types.Tool]:
@ -52,16 +51,9 @@ async def test_notification_validation_error(tmp_path: Path):
request_count += 1 request_count += 1
if name == "slow": if name == "slow":
# Signal that slow request has started await slow_request_lock.wait() # it should timeout here
slow_request_started.set()
# Long enough to ensure timeout
await anyio.sleep(0.2)
# Signal completion
slow_request_complete.set()
return [TextContent(type="text", text=f"slow {request_count}")] return [TextContent(type="text", text=f"slow {request_count}")]
elif name == "fast": elif name == "fast":
# Fast enough to complete before timeout
await anyio.sleep(0.01)
return [TextContent(type="text", text=f"fast {request_count}")] return [TextContent(type="text", text=f"fast {request_count}")]
return [TextContent(type="text", text=f"unknown {request_count}")] return [TextContent(type="text", text=f"unknown {request_count}")]
@ -90,16 +82,15 @@ async def test_notification_validation_error(tmp_path: Path):
# First call should work (fast operation) # First call should work (fast operation)
result = await session.call_tool("fast") result = await session.call_tool("fast")
assert result.content == [TextContent(type="text", text="fast 1")] assert result.content == [TextContent(type="text", text="fast 1")]
assert not slow_request_complete.is_set() assert not slow_request_lock.is_set()
# Second call should timeout (slow operation) # Second call should timeout (slow operation)
with pytest.raises(McpError) as exc_info: with pytest.raises(McpError) as exc_info:
await session.call_tool("slow") await session.call_tool("slow")
assert "Timed out while waiting" in str(exc_info.value) assert "Timed out while waiting" in str(exc_info.value)
# Wait for slow request to complete in the background # release the slow request not to have hanging process
with anyio.fail_after(1): # Timeout after 1 second slow_request_lock.set()
await slow_request_complete.wait()
# Third call should work (fast operation), # Third call should work (fast operation),
# proving server is still responsive # proving server is still responsive