ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Handle URL tests

Browse source
This commit is contained in:
Jeremiah Lowin 2024-12-03 12:54:25 -05:00
commit 83eb81ff7f
5 changed files with 104 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

17
tests/resources/test_file_resources.py
View file

@ -3,6 +3,7 @@ import os
import pytest
from pathlib import Path
from tempfile import NamedTemporaryFile
from pydantic import FileUrl
from fastmcp.resources import FileResource
@ -30,7 +31,7 @@ class TestFileResource:
def test_file_resource_creation(self, temp_file: Path):
"""Test creating a FileResource."""
resource = FileResource(
uri=temp_file.as_uri(),
uri=FileUrl(temp_file.as_uri()),
name="test",
description="test file",
path=temp_file,
@ -45,9 +46,9 @@ class TestFileResource:
def test_file_resource_str_path_conversion(self, temp_file: Path):
"""Test FileResource handles string paths."""
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=str(temp_file),
path=Path(str(temp_file)),
)
assert isinstance(resource.path, Path)
assert resource.path.is_absolute()
@ -55,7 +56,7 @@ class TestFileResource:
async def test_read_text_file(self, temp_file: Path):
"""Test reading a text file."""
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
@ -66,7 +67,7 @@ class TestFileResource:
async def test_read_binary_file(self, temp_file: Path):
"""Test reading a file as binary."""
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
is_binary=True,
@ -79,7 +80,7 @@ class TestFileResource:
"""Test error on relative path."""
with pytest.raises(ValueError, match="Path must be absolute"):
FileResource(
uri="file:///test.txt",
uri=FileUrl("file:///test.txt"),
name="test",
path=Path("test.txt"),
)
@ -89,7 +90,7 @@ class TestFileResource:
# Create path to non-existent file
missing = temp_file.parent / "missing.txt"
resource = FileResource(
uri="file:///missing.txt",
uri=FileUrl("file:///missing.txt"),
name="test",
path=missing,
)
@ -104,7 +105,7 @@ class TestFileResource:
temp_file.chmod(0o000) # Remove all permissions
try:
resource = FileResource(
uri=temp_file.as_uri(),
uri=FileUrl(temp_file.as_uri()),
name="test",
path=temp_file,
)

17
tests/resources/test_function_resources.py
View file

@ -1,4 +1,4 @@
from pydantic import BaseModel
from pydantic import BaseModel, AnyUrl
import pytest
from fastmcp.resources import FunctionResource
@ -13,7 +13,7 @@ class TestFunctionResource:
return "test content"
resource = FunctionResource(
uri="fn://test",
uri=AnyUrl("fn://test"),
name="test",
description="test function",
fn=my_func,
@ -31,7 +31,7 @@ class TestFunctionResource:
return "Hello, world!"
resource = FunctionResource(
uri="function://test",
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)
@ -46,7 +46,7 @@ class TestFunctionResource:
return b"Hello, world!"
resource = FunctionResource(
uri="function://test",
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)
@ -60,11 +60,12 @@ class TestFunctionResource:
return {"key": "value"}
resource = FunctionResource(
uri="function://test",
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)
content = await resource.read()
assert isinstance(content, str)
assert '"key": "value"' in content
async def test_error_handling(self):
@ -74,7 +75,7 @@ class TestFunctionResource:
raise ValueError("Test error")
resource = FunctionResource(
uri="function://test",
uri=AnyUrl("function://test"),
name="test",
fn=failing_func,
)
@ -88,7 +89,7 @@ class TestFunctionResource:
name: str
resource = FunctionResource(
uri="function://test",
uri=AnyUrl("function://test"),
name="test",
fn=lambda: MyModel(name="test"),
)
@ -106,7 +107,7 @@ class TestFunctionResource:
return CustomData()
resource = FunctionResource(
uri="function://test",
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)

19
tests/resources/test_resource_manager.py
View file

@ -1,6 +1,7 @@
import pytest
from pathlib import Path
from tempfile import NamedTemporaryFile
from pydantic import AnyUrl, FileUrl
from fastmcp.resources import (
FileResource,
@ -34,7 +35,7 @@ class TestResourceManager:
"""Test adding a resource."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
@ -46,7 +47,7 @@ class TestResourceManager:
"""Test adding the same resource twice."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
@ -59,7 +60,7 @@ class TestResourceManager:
"""Test warning on duplicate resources."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
@ -71,7 +72,7 @@ class TestResourceManager:
"""Test disabling warning on duplicate resources."""
manager = ResourceManager(warn_on_duplicate_resources=False)
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
@ -83,7 +84,7 @@ class TestResourceManager:
"""Test getting a resource by URI."""
manager = ResourceManager()
resource = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
@ -105,7 +106,7 @@ class TestResourceManager:
)
manager._templates[template.uri_template] = template
resource = await manager.get_resource("greet://world")
resource = await manager.get_resource(AnyUrl("greet://world"))
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "Hello, world!"
@ -114,18 +115,18 @@ class TestResourceManager:
"""Test getting a non-existent resource."""
manager = ResourceManager()
with pytest.raises(ValueError, match="Unknown resource"):
await manager.get_resource("unknown://test")
await manager.get_resource(AnyUrl("unknown://test"))
def test_list_resources(self, temp_file: Path):
"""Test listing all resources."""
manager = ResourceManager()
resource1 = FileResource(
uri=f"file://{temp_file}",
uri=FileUrl(f"file://{temp_file}"),
name="test1",
path=temp_file,
)
resource2 = FileResource(
uri=f"file://{temp_file}2",
uri=FileUrl(f"file://{temp_file}2"),
name="test2",
path=temp_file,
)

196
tests/resources/test_resource_template.py
View file

@ -1,121 +1,71 @@
import json
import pytest
from fastmcp.resources import ResourceTemplate, FunctionResource
from pydantic import BaseModel
from fastmcp.resources import FunctionResource, ResourceTemplate
class TestResourceTemplate:
"""Test ResourceTemplate functionality."""
def test_template_from_function(self):
def test_template_creation(self):
"""Test creating a template from a function."""
def weather(city: str, units: str = "metric") -> str:
return f"Weather in {city} ({units})"
def my_func(key: str, value: int) -> dict:
return {"key": key, "value": value}
template = ResourceTemplate.from_function(
fn=weather,
uri_template="weather://{city}/current",
name="weather",
description="Get current weather",
fn=my_func,
uri_template="test://{key}/{value}",
name="test",
)
assert template.name == "weather"
assert template.uri_template == "weather://{city}/current"
assert template.mime_type == "text/plain"
assert "city" in template.parameters["properties"]
def test_template_from_lambda_error(self):
"""Test error when creating template from lambda without name."""
with pytest.raises(
ValueError, match="You must provide a name for lambda functions"
):
ResourceTemplate.from_function(
fn=lambda x: x,
uri_template="test://{x}",
)
assert template.uri_template == "test://{key}/{value}"
assert template.name == "test"
assert template.mime_type == "text/plain" # default
assert template.fn == my_func
def test_template_matches(self):
"""Test URI matching against template."""
"""Test matching URIs against a template."""
def dummy(x: str) -> str:
return x
def my_func(key: str, value: int) -> dict:
return {"key": key, "value": value}
template = ResourceTemplate.from_function(
fn=dummy,
uri_template="test://{x}/value",
fn=my_func,
uri_template="test://{key}/{value}",
name="test",
)
# Test matching URI
params = template.matches("test://hello/value")
assert params == {"x": "hello"}
# Valid match
params = template.matches("test://foo/123")
assert params == {"key": "foo", "value": "123"}
# Test non-matching URI
params = template.matches("test://hello/wrong")
assert params is None
# No match
assert template.matches("test://foo") is None
assert template.matches("other://foo/123") is None
async def test_create_text_resource(self):
"""Test creating a text resource from template."""
async def test_create_resource(self):
"""Test creating a resource from a template."""
def greet(name: str) -> str:
return f"Hello, {name}!"
def my_func(key: str, value: int) -> dict:
return {"key": key, "value": value}
template = ResourceTemplate.from_function(
fn=greet,
uri_template="greet://{name}",
name="greeter",
fn=my_func,
uri_template="test://{key}/{value}",
name="test",
)
resource = await template.create_resource(
"greet://world",
{"name": "world"},
"test://foo/123",
{"key": "foo", "value": 123},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "Hello, world!"
async def test_create_binary_resource(self):
"""Test creating a binary resource from template."""
def get_bytes(value: str) -> bytes:
return value.encode()
template = ResourceTemplate.from_function(
fn=get_bytes,
uri_template="bytes://{value}",
name="bytes",
)
resource = await template.create_resource(
"bytes://test",
{"value": "test"},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == b"test"
async def test_json_conversion(self):
"""Test automatic JSON conversion of non-string/bytes results."""
def get_data(key: str) -> dict:
return {"key": key, "value": 123}
template = ResourceTemplate.from_function(
fn=get_data,
uri_template="data://{key}",
name="data",
)
resource = await template.create_resource(
"data://test",
{"key": "test"},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert '"key": "test"' in content
assert '"value": 123' in content
assert isinstance(content, str)
data = json.loads(content)
assert data == {"key": "foo", "value": 123}
async def test_template_error(self):
"""Test error handling in template resource creation."""
@ -174,65 +124,57 @@ class TestResourceTemplate:
content = await resource.read()
assert content == b"test"
async def test_async_json_conversion(self):
"""Test automatic JSON conversion of async results."""
async def test_basemodel_conversion(self):
"""Test handling of BaseModel types."""
async def get_data(key: str) -> dict:
return {"key": key, "value": 123}
class MyModel(BaseModel):
key: str
value: int
def get_data(key: str, value: int) -> MyModel:
return MyModel(key=key, value=value)
template = ResourceTemplate.from_function(
fn=get_data,
uri_template="data://{key}",
name="data",
uri_template="test://{key}/{value}",
name="test",
)
resource = await template.create_resource(
"data://test",
{"key": "test"},
"test://foo/123",
{"key": "foo", "value": 123},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert '"key": "test"' in content
assert '"value": 123' in content
assert isinstance(content, str)
data = json.loads(content)
assert data == {"key": "foo", "value": 123}
async def test_async_error(self):
"""Test error handling in async template."""
async def test_custom_type_conversion(self):
"""Test handling of custom types."""
async def failing_func(x: str) -> str:
raise ValueError("Test error")
class CustomData:
def __init__(self, value: str):
self.value = value
def __str__(self) -> str:
return self.value
def get_data(value: str) -> CustomData:
return CustomData(value)
template = ResourceTemplate.from_function(
fn=failing_func,
uri_template="fail://{x}",
name="fail",
)
with pytest.raises(
ValueError, match="Error creating resource from template: Test error"
):
await template.create_resource("fail://test", {"x": "test"})
async def test_sync_returning_coroutine(self):
"""Test sync function that returns a coroutine."""
async def async_helper(name: str) -> str:
return f"Hello, {name}!"
def get_greeting(name: str) -> str:
return async_helper(name) # Returns coroutine
template = ResourceTemplate.from_function(
fn=get_greeting,
uri_template="greet://{name}",
name="greeter",
fn=get_data,
uri_template="test://{value}",
name="test",
)
resource = await template.create_resource(
"greet://world",
{"name": "world"},
"test://hello",
{"value": "hello"},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "Hello, world!"
assert content == "hello"

17
tests/resources/test_resources.py
View file

@ -1,4 +1,5 @@
import pytest
from pydantic import AnyUrl
from fastmcp.resources import FunctionResource, Resource
@ -14,7 +15,7 @@ class TestResourceValidation:
# Valid URI
resource = FunctionResource(
uri="http://example.com/data",
uri=AnyUrl("http://example.com/data"),
name="test",
fn=dummy_func,
)
@ -23,7 +24,7 @@ class TestResourceValidation:
# Missing protocol
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri="invalid",
uri=AnyUrl("invalid"),
name="test",
fn=dummy_func,
)
@ -31,7 +32,7 @@ class TestResourceValidation:
# Missing host
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri="http://",
uri=AnyUrl("http://"),
name="test",
fn=dummy_func,
)
@ -43,7 +44,7 @@ class TestResourceValidation:
return "data"
resource = FunctionResource(
uri="resource://my-resource",
uri=AnyUrl("resource://my-resource"),
fn=dummy_func,
)
assert resource.name == "resource://my-resource"
@ -62,7 +63,7 @@ class TestResourceValidation:
# Explicit name takes precedence over URI
resource = FunctionResource(
uri="resource://uri-name",
uri=AnyUrl("resource://uri-name"),
name="explicit-name",
fn=dummy_func,
)
@ -76,14 +77,14 @@ class TestResourceValidation:
# Default mime type
resource = FunctionResource(
uri="resource://test",
uri=AnyUrl("resource://test"),
fn=dummy_func,
)
assert resource.mime_type == "text/plain"
# Custom mime type
resource = FunctionResource(
uri="resource://test",
uri=AnyUrl("resource://test"),
fn=dummy_func,
mime_type="application/json",
)
@ -96,4 +97,4 @@ class TestResourceValidation:
pass
with pytest.raises(TypeError, match="abstract method"):
ConcreteResource(uri="test://test", name="test") # type: ignore
ConcreteResource(uri=AnyUrl("test://test"), name="test") # type: ignore