|
5 | 5 | from dataclasses import dataclass |
6 | 6 | from pathlib import Path |
7 | 7 | from types import SimpleNamespace |
8 | | -from typing import Annotated, Any |
| 8 | +from typing import Annotated, Any, Literal |
9 | 9 | from unittest.mock import AsyncMock, MagicMock, patch |
10 | 10 |
|
11 | 11 | import anyio |
@@ -111,10 +111,24 @@ def test_dependencies(self): |
111 | 111 | mcp_no_deps = MCPServer("test") |
112 | 112 | assert mcp_no_deps.dependencies == [] |
113 | 113 |
|
114 | | - def test_stdio_keyboard_interrupt_exits_cleanly(self): |
| 114 | + @pytest.mark.parametrize("transport", ["stdio", "sse", "streamable-http"]) |
| 115 | + def test_keyboard_interrupt_exits_cleanly(self, transport: Literal["stdio", "sse", "streamable-http"]): |
115 | 116 | mcp = MCPServer("test") |
116 | 117 |
|
117 | 118 | with patch("mcp.server.mcpserver.server.anyio.run", side_effect=KeyboardInterrupt) as run: |
| 119 | + mcp.run(transport) |
| 120 | + |
| 121 | + assert run.call_count == 1 |
| 122 | + if transport == "stdio": |
| 123 | + run.assert_called_once_with(mcp.run_stdio_async) |
| 124 | + |
| 125 | + def test_run_propagates_non_interrupt_errors(self): |
| 126 | + mcp = MCPServer("test") |
| 127 | + |
| 128 | + with ( |
| 129 | + patch("mcp.server.mcpserver.server.anyio.run", side_effect=RuntimeError("boom")) as run, |
| 130 | + pytest.raises(RuntimeError, match="boom"), |
| 131 | + ): |
118 | 132 | mcp.run("stdio") |
119 | 133 |
|
120 | 134 | run.assert_called_once_with(mcp.run_stdio_async) |
|
0 commit comments