File size: 4,559 Bytes
b45adb7
 
4d9630f
 
b45adb7
 
 
 
 
 
 
 
4d9630f
 
 
b45adb7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4d88236
 
 
b45adb7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4d88236
 
 
b45adb7
 
09cabfa
b45adb7
 
 
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
"""Tests for legacy resource prefix behavior."""

import pytest

from fastmcp import Client, FastMCP
from fastmcp.server.server import (
    add_resource_prefix,
    has_resource_prefix,
    remove_resource_prefix,
)
from fastmcp.utilities.tests import temporary_settings

# reset deprecation warnings for this module
pytestmark = pytest.mark.filterwarnings("default::DeprecationWarning")


class TestLegacyResourcePrefixes:
    """Test the legacy resource prefix behavior."""

    def test_add_resource_prefix_legacy(self):
        """Test that add_resource_prefix uses the legacy format when resource_prefix_format is 'protocol'."""
        with temporary_settings(resource_prefix_format="protocol"):
            result = add_resource_prefix("resource://path/to/resource", "prefix")
            assert result == "prefix+resource://path/to/resource"

            # Empty prefix should return the original URI
            result = add_resource_prefix("resource://path/to/resource", "")
            assert result == "resource://path/to/resource"

    def test_remove_resource_prefix_legacy(self):
        """Test that remove_resource_prefix uses the legacy format when resource_prefix_format is 'protocol'."""
        with temporary_settings(resource_prefix_format="protocol"):
            result = remove_resource_prefix(
                "prefix+resource://path/to/resource", "prefix"
            )
            assert result == "resource://path/to/resource"

            # URI without the prefix should be returned as is
            result = remove_resource_prefix("resource://path/to/resource", "prefix")
            assert result == "resource://path/to/resource"

            # Empty prefix should return the original URI
            result = remove_resource_prefix("resource://path/to/resource", "")
            assert result == "resource://path/to/resource"

    def test_has_resource_prefix_legacy(self):
        """Test that has_resource_prefix uses the legacy format when resource_prefix_format is 'protocol'."""
        with temporary_settings(resource_prefix_format="protocol"):
            result = has_resource_prefix("prefix+resource://path/to/resource", "prefix")
            assert result is True

            result = has_resource_prefix("resource://path/to/resource", "prefix")
            assert result is False

            # Empty prefix should always return False
            result = has_resource_prefix("resource://path/to/resource", "")
            assert result is False


async def test_mount_with_legacy_prefixes():
    """Test mounting a server with legacy resource prefixes."""
    with temporary_settings(resource_prefix_format="protocol"):
        main_server = FastMCP("MainServer")
        sub_server = FastMCP("SubServer")

        @sub_server.resource("resource://test")
        def get_test():
            return "test content"

        # Mount the server with a prefix (using old argument order for this legacy test)
        with pytest.warns(DeprecationWarning, match="Mount prefixes are now optional"):
            main_server.mount("sub", sub_server)  # type: ignore[arg-type]

        # Check that the resource is prefixed using the legacy format
        resources = await main_server.get_resources()

        # In legacy format, the key would be "sub+resource://test"
        assert "sub+resource://test" in resources

        # Test accessing the resource through client
        async with Client(main_server) as client:
            result = await client.read_resource("sub+resource://test")
            # Different content types might be returned, but we just want to verify we got something
            assert len(result) > 0


async def test_import_server_with_legacy_prefixes():
    """Test importing a server with legacy resource prefixes."""
    with temporary_settings(resource_prefix_format="protocol"):
        main_server = FastMCP("MainServer")
        sub_server = FastMCP("SubServer")

        @sub_server.resource("resource://test")
        def get_test():
            return "test content"

        # Import the server with a prefix (using old argument order for this legacy test)
        with pytest.warns(DeprecationWarning, match="Import prefixes are now optional"):
            await main_server.import_server("sub", sub_server)  # type: ignore[arg-type]

        # Check that the resource is prefixed using the legacy format
        resources = await main_server.get_resources()

        # In legacy format, the key would be "sub+resource://test"
        assert "sub+resource://test" in resources