Spaces:
Running
Running
File size: 3,556 Bytes
48f7143 09438a8 48f7143 0feb19f 48f7143 0feb19f 48f7143 0feb19f 48f7143 81e8514 48f7143 81e8514 48f7143 0feb19f 48f7143 | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | import json
from pathlib import Path
import pytest
from fastmcp import FastMCP
@pytest.fixture()
def test_dir(tmp_path_factory) -> Path:
"""Create a temporary directory with test files."""
tmp = tmp_path_factory.mktemp("test_files")
# Create test files
(tmp / "example.py").write_text("print('hello world')")
(tmp / "readme.md").write_text("# Test Directory\nThis is a test.")
(tmp / "config.json").write_text('{"test": true}')
return tmp
@pytest.fixture
def mcp() -> FastMCP:
mcp = FastMCP()
return mcp
@pytest.fixture(autouse=True)
def resources(mcp: FastMCP, test_dir: Path) -> FastMCP:
@mcp.resource("dir://test_dir")
def list_test_dir() -> list[str]:
"""List the files in the test directory"""
return [str(f) for f in test_dir.iterdir()]
@mcp.resource("file://test_dir/example.py")
def read_example_py() -> str:
"""Read the example.py file"""
try:
return (test_dir / "example.py").read_text()
except FileNotFoundError:
return "File not found"
@mcp.resource("file://test_dir/readme.md")
def read_readme_md() -> str:
"""Read the readme.md file"""
try:
return (test_dir / "readme.md").read_text()
except FileNotFoundError:
return "File not found"
@mcp.resource("file://test_dir/config.json")
def read_config_json() -> str:
"""Read the config.json file"""
try:
return (test_dir / "config.json").read_text()
except FileNotFoundError:
return "File not found"
return mcp
@pytest.fixture(autouse=True)
def tools(mcp: FastMCP, test_dir: Path) -> FastMCP:
@mcp.tool
def delete_file(path: str) -> bool:
# ensure path is in test_dir
if Path(path).resolve().parent != test_dir:
raise ValueError(f"Path must be in test_dir: {path}")
Path(path).unlink()
return True
return mcp
async def test_list_resources(mcp: FastMCP):
resources = await mcp._mcp_list_resources()
assert len(resources) == 4
assert [str(r.uri) for r in resources] == [
"dir://test_dir",
"file://test_dir/example.py",
"file://test_dir/readme.md",
"file://test_dir/config.json",
]
async def test_read_resource_dir(mcp: FastMCP):
res_iter = await mcp._mcp_read_resource("dir://test_dir")
res_list = list(res_iter)
assert len(res_list) == 1
res = res_list[0]
assert res.mime_type == "text/plain"
files = json.loads(res.content)
assert sorted([Path(f).name for f in files]) == [
"config.json",
"example.py",
"readme.md",
]
async def test_read_resource_file(mcp: FastMCP):
res_iter = await mcp._mcp_read_resource("file://test_dir/example.py")
res_list = list(res_iter)
assert len(res_list) == 1
res = res_list[0]
assert res.content == "print('hello world')"
async def test_delete_file(mcp: FastMCP, test_dir: Path):
await mcp._mcp_call_tool(
"delete_file", arguments=dict(path=str(test_dir / "example.py"))
)
assert not (test_dir / "example.py").exists()
async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path):
await mcp._mcp_call_tool(
"delete_file", arguments=dict(path=str(test_dir / "example.py"))
)
res_iter = await mcp._mcp_read_resource("file://test_dir/example.py")
res_list = list(res_iter)
assert len(res_list) == 1
res = res_list[0]
assert res.content == "File not found"
|