test_server.py (4435 bytes)
1 """Checks for server.py: a round trip, name validation, and one run over stdio.""" 2 3 import asyncio 4 import json 5 import os 6 import sys 7 from collections.abc import Iterator 8 from pathlib import Path 9 from typing import Any 10 11 import pytest 12 from hypothesis import example, given 13 from hypothesis import strategies as st 14 from mcp import ClientSession, StdioServerParameters 15 from mcp.client.stdio import stdio_client 16 from mcp.server.mcpserver.exceptions import ToolError 17 18 import server 19 20 SERVER = str(Path(__file__).with_name("server.py")) 21 UPPER = "import sys\nsys.stdout.buffer.write(sys.stdin.buffer.read().upper())\n" 22 23 24 # One interpreter per extension. Each copies its program to stdout. 25 ECHO = { 26 "awk": "{ print }\n", 27 "sh": "cat\n", 28 "lisp": "(loop for line = (read-line nil nil) while line do (write-line line))\n", 29 "scm": "(use-modules (ice-9 rdelim))\n" 30 "(let loop ((line (read-line)))\n" 31 " (unless (eof-object? line) (write-line line) (loop (read-line))))\n", 32 "hs": "main :: IO ()\nmain = interact id\n", 33 "clj": "(print (slurp *in*))\n(flush)\n", 34 "js": "process.stdin.pipe(process.stdout);\n", 35 "py": "import sys\nsys.stdout.write(sys.stdin.read())\n", 36 } 37 38 39 @pytest.fixture(scope="module", autouse=True) 40 def dsl_home(tmp_path_factory: pytest.TempPathFactory) -> Iterator[None]: 41 home = tmp_path_factory.mktemp("dsl") 42 # Keep the overrides of this machine, so the runtime test checks them. 43 mine = server.home() / "runtimes.json" 44 table = json.loads(mine.read_bytes()) if mine.exists() else {} 45 # "up" exists only in this override, so a run with it proves the override loads. 46 (home / "runtimes.json").write_text(json.dumps(table | {"up": [sys.executable]})) 47 with pytest.MonkeyPatch.context() as patch: 48 patch.setenv("DSL_HOME", str(home)) 49 yield 50 51 52 @given( 53 grammar=st.text(), 54 ext=st.sampled_from(sorted(server.RUNTIMES)), 55 source=st.text(), 56 ) 57 def test_put_then_get_returns_the_same_dsl(grammar: str, ext: str, source: str) -> None: 58 server.put("t", grammar, ext, source) 59 assert server.get("t") == {"grammar": grammar, "ext": ext, "source": source} 60 assert server.ls()["t"] == ext 61 62 63 @given(name=st.text().filter(lambda s: not server.NAME.fullmatch(s))) 64 @example("../evil") 65 @example("a/b") 66 @example("") 67 def test_put_rejects_a_name_that_is_not_a_slug(name: str) -> None: 68 with pytest.raises(ToolError): 69 server.put(name, "", "py", "") 70 71 72 def test_run_reports_a_timeout() -> None: 73 server.put("slow", "program <- .*", "py", "import time\ntime.sleep(30)\n") 74 with pytest.raises(ToolError, match="timed out"): 75 server.run("slow", "", timeout=0.5) 76 77 78 @pytest.mark.parametrize("ext", sorted(server.RUNTIMES)) 79 def test_each_installed_runtime_runs_an_interpreter(ext: str) -> None: 80 if not server.runtimes()[ext]: 81 pytest.skip(f"{server.table()[ext][0]} is not installed") 82 server.put("echo", "program <- .*", ext, ECHO[ext]) 83 result = server.run("echo", "a b\n", timeout=60) 84 assert (result["exit"], str(result["stdout"]).splitlines()) == (0, ["a b"]), result 85 86 87 def test_put_and_run_over_stdio() -> None: 88 async def call() -> Any: 89 params = StdioServerParameters( 90 command=sys.executable, args=[SERVER], env=dict(os.environ) 91 ) 92 async with ( 93 stdio_client(params) as (read, write), 94 ClientSession(read, write) as session, 95 ): 96 await session.initialize() 97 tools = {tool.name for tool in (await session.list_tools()).tools} 98 assert tools == {"runtimes", "ls", "get", "put", "run"} 99 # Regression: mcp 2.x hides the message of every error but ToolError, 100 # so the client saw only "Error executing tool get". 101 missing = await session.call_tool("get", {"name": "nope"}) 102 assert missing.is_error 103 assert "no DSL named nope" in getattr(missing.content[0], "text", "") 104 dsl = { 105 "name": "upper", 106 "grammar": "text <- .*", 107 "ext": "up", 108 "source": UPPER, 109 } 110 await session.call_tool("put", dsl) 111 result = await session.call_tool( 112 "run", {"name": "upper", "program": "a b\n"} 113 ) 114 assert not result.is_error 115 return result.structured_content 116 117 assert asyncio.run(call()) == {"stdout": "A B\n", "stderr": "", "exit": 0}