Rename FastMCP to MCPServer (#1951)

tests/server/mcpserver/resources/test_resource_manager.py

@@ -0,0 +1,176 @@
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+
+import pytest
+from pydantic import AnyUrl
+
+from mcp.server.mcpserver.resources import FileResource, FunctionResource, ResourceManager, ResourceTemplate
+
+
+@pytest.fixture
+def temp_file():
+    """Create a temporary file for testing.
+
+    File is automatically cleaned up after the test if it still exists.
+    """
+    content = "test content"
+    with NamedTemporaryFile(mode="w", delete=False) as f:
+        f.write(content)
+        path = Path(f.name).resolve()
+    yield path
+    try:  # pragma: lax no cover
+        path.unlink()
+    except FileNotFoundError:  # pragma: lax no cover
+        pass  # File was already deleted by the test
+
+
+class TestResourceManager:
+    """Test ResourceManager functionality."""
+
+    def test_add_resource(self, temp_file: Path):
+        """Test adding a resource."""
+        manager = ResourceManager()
+        resource = FileResource(
+            uri=f"file://{temp_file}",
+            name="test",
+            path=temp_file,
+        )
+        added = manager.add_resource(resource)
+        assert added == resource
+        assert manager.list_resources() == [resource]
+
+    def test_add_duplicate_resource(self, temp_file: Path):
+        """Test adding the same resource twice."""
+        manager = ResourceManager()
+        resource = FileResource(
+            uri=f"file://{temp_file}",
+            name="test",
+            path=temp_file,
+        )
+        first = manager.add_resource(resource)
+        second = manager.add_resource(resource)
+        assert first == second
+        assert manager.list_resources() == [resource]
+
+    def test_warn_on_duplicate_resources(self, temp_file: Path, caplog: pytest.LogCaptureFixture):
+        """Test warning on duplicate resources."""
+        manager = ResourceManager()
+        resource = FileResource(
+            uri=f"file://{temp_file}",
+            name="test",
+            path=temp_file,
+        )
+        manager.add_resource(resource)
+        manager.add_resource(resource)
+        assert "Resource already exists" in caplog.text
+
+    def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog: pytest.LogCaptureFixture):
+        """Test disabling warning on duplicate resources."""
+        manager = ResourceManager(warn_on_duplicate_resources=False)
+        resource = FileResource(
+            uri=f"file://{temp_file}",
+            name="test",
+            path=temp_file,
+        )
+        manager.add_resource(resource)
+        manager.add_resource(resource)
+        assert "Resource already exists" not in caplog.text
+
+    @pytest.mark.anyio
+    async def test_get_resource(self, temp_file: Path):
+        """Test getting a resource by URI."""
+        manager = ResourceManager()
+        resource = FileResource(
+            uri=f"file://{temp_file}",
+            name="test",
+            path=temp_file,
+        )
+        manager.add_resource(resource)
+        retrieved = await manager.get_resource(resource.uri)
+        assert retrieved == resource
+
+    @pytest.mark.anyio
+    async def test_get_resource_from_template(self):
+        """Test getting a resource through a template."""
+        manager = ResourceManager()
+
+        def greet(name: str) -> str:
+            return f"Hello, {name}!"
+
+        template = ResourceTemplate.from_function(
+            fn=greet,
+            uri_template="greet://{name}",
+            name="greeter",
+        )
+        manager._templates[template.uri_template] = template
+
+        resource = await manager.get_resource(AnyUrl("greet://world"))
+        assert isinstance(resource, FunctionResource)
+        content = await resource.read()
+        assert content == "Hello, world!"
+
+    @pytest.mark.anyio
+    async def test_get_unknown_resource(self):
+        """Test getting a non-existent resource."""
+        manager = ResourceManager()
+        with pytest.raises(ValueError, match="Unknown resource"):
+            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}",
+            name="test1",
+            path=temp_file,
+        )
+        resource2 = FileResource(
+            uri=f"file://{temp_file}2",
+            name="test2",
+            path=temp_file,
+        )
+        manager.add_resource(resource1)
+        manager.add_resource(resource2)
+        resources = manager.list_resources()
+        assert len(resources) == 2
+        assert resources == [resource1, resource2]
+
+
+class TestResourceManagerMetadata:
+    """Test ResourceManager Metadata"""
+
+    def test_add_template_with_metadata(self):
+        """Test that ResourceManager.add_template() accepts and passes meta parameter."""
+
+        manager = ResourceManager()
+
+        def get_item(id: str) -> str:  # pragma: no cover
+            return f"Item {id}"
+
+        metadata = {"source": "database", "cached": True}
+
+        template = manager.add_template(
+            fn=get_item,
+            uri_template="resource://items/{id}",
+            meta=metadata,
+        )
+
+        assert template.meta is not None
+        assert template.meta == metadata
+        assert template.meta["source"] == "database"
+        assert template.meta["cached"] is True
+
+    def test_add_template_without_metadata(self):
+        """Test that ResourceManager.add_template() works without meta parameter."""
+
+        manager = ResourceManager()
+
+        def get_item(id: str) -> str:  # pragma: no cover
+            return f"Item {id}"
+
+        template = manager.add_template(
+            fn=get_item,
+            uri_template="resource://items/{id}",
+        )
+
+        assert template.meta is None