prefecthq--fastmcp/tests/server/mount/test_resources.py

"""Tests for resource and template mounting."""

import json

from fastmcp import FastMCP


class TestResourcesAndTemplates:
    """Test mounting with resources and resource templates."""

    async def test_mount_with_resources(self):
        """Test mounting a server with resources."""
        main_app = FastMCP("MainApp")
        data_app = FastMCP("DataApp")

        @data_app.resource(uri="data://users")
        async def get_users() -> str:
            return "user1, user2"

        # Mount the data app
        main_app.mount(data_app, "data")

        # Resource should be accessible through main app
        resources = await main_app.list_resources()
        assert any(str(r.uri) == "data://data/users" for r in resources)

        # Check that resource can be accessed
        result = await main_app.read_resource("data://data/users")
        assert len(result.contents) == 1
        # Note: The function returns "user1, user2" which is not valid JSON
        # This test should be updated to return proper JSON or check the string directly
        assert result.contents[0].content == "user1, user2"

    async def test_mount_with_resource_templates(self):
        """Test mounting a server with resource templates."""
        main_app = FastMCP("MainApp")
        user_app = FastMCP("UserApp")

        @user_app.resource(uri="users://{user_id}/profile")
        def get_user_profile(user_id: str) -> str:
            return json.dumps({"id": user_id, "name": f"User {user_id}"})

        # Mount the user app
        main_app.mount(user_app, "api")

        # Template should be accessible through main app
        templates = await main_app.list_resource_templates()
        assert any(t.uri_template == "users://api/{user_id}/profile" for t in templates)

        # Check template instantiation
        result = await main_app.read_resource("users://api/123/profile")
        assert len(result.contents) == 1
        profile = json.loads(result.contents[0].content)
        assert profile["id"] == "123"
        assert profile["name"] == "User 123"

    async def test_mount_with_wildcard_resource_template(self):
        """Wildcard `{name*}` params must survive round-trip through a namespaced mount."""
        main_app = FastMCP("MainApp")
        sub_app = FastMCP("SubApp")

        @sub_app.resource("resource://multi/{extra*}")
        def multi(extra: str) -> str:
            return extra

        main_app.mount(sub_app, namespace="sub")

        result = await main_app.read_resource("resource://sub/multi/abc/def")
        assert result.contents[0].content == "abc/def"

        result = await main_app.read_resource("resource://sub/multi/abc")
        assert result.contents[0].content == "abc"

    async def test_adding_resource_after_mounting(self):
        """Test adding a resource after mounting."""
        main_app = FastMCP("MainApp")
        data_app = FastMCP("DataApp")

        # Mount the data app before adding resources
        main_app.mount(data_app, "data")

        # Add a resource after mounting
        @data_app.resource(uri="data://config")
        def get_config() -> str:
            return json.dumps({"version": "1.0"})

        # Resource should be accessible through main app
        resources = await main_app.list_resources()
        assert any(str(r.uri) == "data://data/config" for r in resources)

        # Check access to the resource
        result = await main_app.read_resource("data://data/config")
        assert len(result.contents) == 1
        config = json.loads(result.contents[0].content)
        assert config["version"] == "1.0"


class TestResourceUriPrefixing:
    """Test that resource and resource template URIs get prefixed when mounted (names are NOT prefixed)."""

    async def test_resource_uri_prefixing(self):
        """Test that resource URIs are prefixed when mounted (names are NOT prefixed)."""

        # Create a sub-app with a resource
        sub_app = FastMCP("SubApp")

        @sub_app.resource("resource://my_resource")
        def my_resource() -> str:
            return "Resource content"

        # Create main app and mount sub-app with prefix
        main_app = FastMCP("MainApp")
        main_app.mount(sub_app, "prefix")

        # Get resources from main app
        resources = await main_app.list_resources()

        # Should have prefixed key (using path format: resource://prefix/resource_name)
        assert any(str(r.uri) == "resource://prefix/my_resource" for r in resources)

        # The resource name should NOT be prefixed (only URI is prefixed)
        resource = next(
            r for r in resources if str(r.uri) == "resource://prefix/my_resource"
        )
        assert resource.name == "my_resource"

    async def test_resource_template_uri_prefixing(self):
        """Test that resource template URIs are prefixed when mounted (names are NOT prefixed)."""

        # Create a sub-app with a resource template
        sub_app = FastMCP("SubApp")

        @sub_app.resource("resource://user/{user_id}")
        def user_template(user_id: str) -> str:
            return f"User {user_id} data"

        # Create main app and mount sub-app with prefix
        main_app = FastMCP("MainApp")
        main_app.mount(sub_app, "prefix")

        # Get resource templates from main app
        templates = await main_app.list_resource_templates()

        # Should have prefixed key (using path format: resource://prefix/template_uri)
        assert any(
            t.uri_template == "resource://prefix/user/{user_id}" for t in templates
        )

        # The template name should NOT be prefixed (only URI template is prefixed)
        template = next(
            t for t in templates if t.uri_template == "resource://prefix/user/{user_id}"
        )
        assert template.name == "user_template"


class TestMountedResourceTemplateQueryParams:
    """Test that resource templates with query params work on mounted servers."""

    async def test_mounted_template_with_query_param(self):
        """Query params in resource templates should work through mount."""
        sub = FastMCP("Sub")

        @sub.resource("resource://greet{?name}")
        def greet(name: str = "World") -> str:
            return f"Hello, {name}!"

        main = FastMCP("Main")
        main.mount(sub, "sub")

        result = await main.read_resource("resource://sub/greet?name=Alice")
        assert result.contents[0].content == "Hello, Alice!"

    async def test_mounted_template_with_query_param_default(self):
        """Missing query params should use defaults through mount."""
        sub = FastMCP("Sub")

        @sub.resource("resource://greet{?name}")
        def greet(name: str = "World") -> str:
            return f"Hello, {name}!"

        main = FastMCP("Main")
        main.mount(sub, "sub")

        result = await main.read_resource("resource://sub/greet")
        assert result.contents[0].content == "Hello, World!"

    async def test_mounted_template_with_multiple_query_params(self):
        """Multiple query params should all pass through mount correctly."""
        sub = FastMCP("Sub")

        @sub.resource("resource://data/{id}{?format,verbose}")
        def get_data(id: str, format: str = "json", verbose: bool = False) -> str:
            return f"id={id} format={format} verbose={verbose}"

        main = FastMCP("Main")
        main.mount(sub, "api")

        result = await main.read_resource(
            "resource://api/data/42?format=xml&verbose=true"
        )
        assert result.contents[0].content == "id=42 format=xml verbose=True"

    async def test_mounted_template_with_partial_query_params(self):
        """Providing only some query params should use defaults for the rest."""
        sub = FastMCP("Sub")

        @sub.resource("resource://data/{id}{?format,limit}")
        def get_data(id: str, format: str = "json", limit: int = 10) -> str:
            return f"id={id} format={format} limit={limit}"

        main = FastMCP("Main")
        main.mount(sub, "api")

        result = await main.read_resource("resource://api/data/42?limit=5")
        assert result.contents[0].content == "id=42 format=json limit=5"