File size: 3,885 Bytes
c7708b2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Tests for deprecated FastMCPProxy client parameter."""

import warnings

import pytest

from fastmcp import Client, FastMCP
from fastmcp.server.proxy import FastMCPProxy, ProxyClient


@pytest.fixture
def simple_server():
    """Create a simple FastMCP server for testing."""
    server = FastMCP("TestServer")

    @server.tool
    def simple_tool() -> str:
        return "test_result"

    return server


class TestDeprecatedClientParameter:
    """Test the deprecated client parameter in FastMCPProxy."""

    def test_client_parameter_deprecation_warning(self, simple_server):
        """Test that using the client parameter raises a deprecation warning."""
        client = Client(simple_server)

        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter("always")  # Ensure all warnings are captured

            FastMCPProxy(client=client)

            # Verify a deprecation warning was raised
            assert len(w) == 1
            assert issubclass(w[0].category, DeprecationWarning)
            assert "client' to FastMCPProxy is deprecated" in str(w[0].message)
            assert "client_factory" in str(w[0].message)

    def test_client_parameter_still_works(self, simple_server):
        """Test that the deprecated client parameter still functions."""
        client = ProxyClient(simple_server)

        with warnings.catch_warnings():
            warnings.simplefilter("ignore")  # Suppress warnings for functionality test

            proxy = FastMCPProxy(client=client)

            # Verify the proxy was created successfully
            assert proxy is not None
            assert hasattr(proxy, "client_factory")
            assert callable(proxy.client_factory)

            # Verify the factory returns a new client instance (session isolation for backwards compatibility)
            returned_client = proxy.client_factory()
            assert returned_client is not client
            assert isinstance(returned_client, type(client))

    def test_cannot_specify_both_client_and_factory(self, simple_server):
        """Test that specifying both client and client_factory raises an error."""
        client = Client(simple_server)

        def factory():
            return Client(simple_server)

        with pytest.raises(
            ValueError, match="Cannot specify both 'client' and 'client_factory'"
        ):
            FastMCPProxy(client=client, client_factory=factory)

    def test_must_specify_client_factory_when_no_client(self):
        """Test that client_factory is required when client is not provided."""
        with pytest.raises(ValueError, match="Must specify 'client_factory'"):
            FastMCPProxy()

    def test_client_factory_preferred_over_deprecated_client(self, simple_server):
        """Test that the recommended client_factory approach works without warnings."""

        def factory():
            return ProxyClient(simple_server)

        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter("always")

            proxy = FastMCPProxy(client_factory=factory)

            # Verify no warnings were raised
            assert len(w) == 0

            # Verify the proxy works correctly
            assert proxy is not None
            assert proxy.client_factory is factory

    async def test_deprecated_client_functional_test(self, simple_server):
        """End-to-end test that deprecated client parameter still works functionally."""
        client = ProxyClient(simple_server)

        with warnings.catch_warnings():
            warnings.simplefilter("ignore")

            proxy = FastMCPProxy(client=client)

        # Test that the proxy can actually handle requests
        async with Client(proxy) as proxy_client:
            result = await proxy_client.call_tool("simple_tool", {})
            assert result.data == "test_result"