Marcelo Trylesinski
392bb35d35
Consolidate the separate README.v2.md and docs/ pytest-examples tests
into a single unified test using `find_examples("README.v2.md", "docs/")`.
Add skip tags to pre-existing code blocks that aren't self-contained:
- migration.md: all blocks (before/after migration patterns)
- experimental/*.md: all blocks (experimental API patterns)
- testing.md: test blocks (reference local server module)
- README.v2.md: 4 context-dependent snippets
144 lines
5.1 KiB
Python
144 lines
5.1 KiB
Python
"""Tests for example servers"""
|
|
# TODO(Marcelo): The `examples` directory needs to be importable as a package.
|
|
# pyright: reportMissingImports=false
|
|
# pyright: reportUnknownVariableType=false
|
|
# pyright: reportUnknownArgumentType=false
|
|
# pyright: reportUnknownMemberType=false
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from inline_snapshot import snapshot
|
|
from pytest_examples import CodeExample, EvalExample, find_examples
|
|
|
|
from mcp import Client
|
|
from mcp.types import CallToolResult, TextContent, TextResourceContents
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_simple_echo():
|
|
"""Test the simple echo server"""
|
|
from examples.mcpserver.simple_echo import mcp
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("echo", {"text": "hello"})
|
|
assert result == snapshot(
|
|
CallToolResult(content=[TextContent(text="hello")], structured_content={"result": "hello"})
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_complex_inputs():
|
|
"""Test the complex inputs server"""
|
|
from examples.mcpserver.complex_inputs import mcp
|
|
|
|
async with Client(mcp) as client:
|
|
tank = {"shrimp": [{"name": "bob"}, {"name": "alice"}]}
|
|
result = await client.call_tool("name_shrimp", {"tank": tank, "extra_names": ["charlie"]})
|
|
assert result == snapshot(
|
|
CallToolResult(
|
|
content=[
|
|
TextContent(text="bob"),
|
|
TextContent(text="alice"),
|
|
TextContent(text="charlie"),
|
|
],
|
|
structured_content={"result": ["bob", "alice", "charlie"]},
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_direct_call_tool_result_return():
|
|
"""Test the CallToolResult echo server"""
|
|
from examples.mcpserver.direct_call_tool_result_return import mcp
|
|
|
|
async with Client(mcp) as client:
|
|
result = await client.call_tool("echo", {"text": "hello"})
|
|
assert result == snapshot(
|
|
CallToolResult(
|
|
meta={"some": "metadata"}, # type: ignore[reportUnknownMemberType]
|
|
content=[TextContent(text="hello")],
|
|
structured_content={"text": "hello"},
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_desktop(monkeypatch: pytest.MonkeyPatch):
|
|
"""Test the desktop server"""
|
|
# Mock desktop directory listing
|
|
mock_files = [Path("/fake/path/file1.txt"), Path("/fake/path/file2.txt")]
|
|
monkeypatch.setattr(Path, "iterdir", lambda self: mock_files) # type: ignore[reportUnknownArgumentType]
|
|
monkeypatch.setattr(Path, "home", lambda: Path("/fake/home"))
|
|
|
|
from examples.mcpserver.desktop import mcp
|
|
|
|
async with Client(mcp) as client:
|
|
# Test the sum function
|
|
result = await client.call_tool("sum", {"a": 1, "b": 2})
|
|
assert result == snapshot(CallToolResult(content=[TextContent(text="3")], structured_content={"result": 3}))
|
|
|
|
# Test the desktop resource
|
|
result = await client.read_resource("dir://desktop")
|
|
assert len(result.contents) == 1
|
|
content = result.contents[0]
|
|
assert isinstance(content, TextResourceContents)
|
|
assert isinstance(content.text, str)
|
|
if sys.platform == "win32": # pragma: no cover
|
|
file_1 = "/fake/path/file1.txt".replace("/", "\\\\") # might be a bug
|
|
file_2 = "/fake/path/file2.txt".replace("/", "\\\\") # might be a bug
|
|
assert file_1 in content.text
|
|
assert file_2 in content.text
|
|
# might be a bug, but the test is passing
|
|
else: # pragma: lax no cover
|
|
assert "/fake/path/file1.txt" in content.text
|
|
assert "/fake/path/file2.txt" in content.text
|
|
|
|
|
|
SKIP_RUN_TAGS = ["skip", "skip-run"]
|
|
SKIP_LINT_TAGS = ["skip", "skip-lint"]
|
|
|
|
# TODO(v2): Change "README.v2.md" back to "README.md" when v2 is released
|
|
_ALL_EXAMPLES = list(find_examples("README.v2.md", "docs/"))
|
|
|
|
|
|
def _set_eval_config(eval_example: EvalExample) -> None:
|
|
eval_example.set_config(
|
|
ruff_ignore=["F841", "I001", "F821"],
|
|
target_version="py310",
|
|
line_length=120,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"example",
|
|
[ex for ex in _ALL_EXAMPLES if not any(ex.prefix_settings().get(key) == "true" for key in SKIP_LINT_TAGS)],
|
|
ids=str,
|
|
)
|
|
def test_docs_examples(example: CodeExample, eval_example: EvalExample):
|
|
_set_eval_config(eval_example)
|
|
|
|
if eval_example.update_examples: # pragma: no cover
|
|
eval_example.format_ruff(example)
|
|
else:
|
|
eval_example.lint_ruff(example)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"example",
|
|
[ex for ex in _ALL_EXAMPLES if not any(ex.prefix_settings().get(key) == "true" for key in SKIP_RUN_TAGS)],
|
|
ids=str,
|
|
)
|
|
def test_docs_examples_run(example: CodeExample, eval_example: EvalExample):
|
|
_set_eval_config(eval_example)
|
|
|
|
# Prevent `if __name__ == "__main__"` blocks from starting servers
|
|
module_globals: dict[str, Any] = {"__name__": "__docs_test__"}
|
|
|
|
if eval_example.update_examples: # pragma: no cover
|
|
eval_example.run_print_update(example, module_globals=module_globals)
|
|
else:
|
|
eval_example.run_print_check(example, module_globals=module_globals)
|