wehub-resource-sync
60e0ffc959
Upgrade checks / Notify on failure (push) Has been cancelled
Upgrade checks / Close issue on success (push) Has been cancelled
Schema Crash Test / Real-world schema crash test (232K schemas) (push) Has been cancelled
Run static analysis / static_analysis (push) Has been cancelled
Tests / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Tests / Tests with lowest-direct dependencies (push) Has been cancelled
Tests / MCP conformance tests (push) Has been cancelled
Tests / Integration tests (push) Has been cancelled
Tests / Package install smoke (push) Has been cancelled
Upgrade checks / Static analysis (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Upgrade checks / Integration tests (push) Has been cancelled
Update MCPServerConfig Schema / update-config-schema (push) Has been cancelled
Update SDK Documentation / update-sdk-docs (push) Has been cancelled
377 lines
11 KiB
Python
377 lines
11 KiB
Python
"""Tests for version fallback when the highest version is disabled via visibility.
|
|
|
|
Regression tests for https://github.com/jlowin/fastmcp/issues/3421:
|
|
When the latest version of a component is disabled, get_* methods should
|
|
fall back to the next-highest enabled version instead of returning None.
|
|
"""
|
|
# ruff: noqa: F811 # Intentional function redefinition for version testing
|
|
|
|
from __future__ import annotations
|
|
|
|
from mcp.server.auth.middleware.auth_context import auth_context_var
|
|
from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
|
|
from mcp_types import TextContent
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import AccessToken, require_scopes
|
|
from fastmcp.utilities.versions import VersionSpec
|
|
|
|
|
|
def _make_token(scopes: list[str] | None = None) -> AccessToken:
|
|
"""Create a test access token."""
|
|
return AccessToken(
|
|
token="test-token",
|
|
client_id="test-client",
|
|
scopes=scopes or [],
|
|
expires_at=None,
|
|
claims={},
|
|
)
|
|
|
|
|
|
def _set_token(token: AccessToken | None):
|
|
"""Set the access token in the auth context var."""
|
|
if token is None:
|
|
return auth_context_var.set(None)
|
|
return auth_context_var.set(AuthenticatedUser(token))
|
|
|
|
|
|
class TestToolVersionFallback:
|
|
"""Test that disabling the latest tool version falls back correctly."""
|
|
|
|
async def test_list_tools_shows_v1_when_v2_disabled(self):
|
|
"""list_tools should show v1 when v2 is disabled."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
tools = await mcp.list_tools()
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "calc"
|
|
assert tools[0].version == "1.0"
|
|
|
|
async def test_get_tool_returns_v1_when_v2_disabled(self):
|
|
"""get_tool should return v1 when v2 is disabled (core bug)."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
tool = await mcp.get_tool("calc")
|
|
assert tool is not None
|
|
assert tool.version == "1.0"
|
|
|
|
async def test_call_tool_uses_v1_when_v2_disabled(self):
|
|
"""call_tool should invoke v1 when v2 is disabled."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
result = await mcp.call_tool("calc", {})
|
|
first = result.content[0]
|
|
assert isinstance(first, TextContent)
|
|
assert first.text == "1"
|
|
|
|
async def test_get_tool_explicit_disabled_version_returns_none(self):
|
|
"""Requesting a specific disabled version should return None."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
tool = await mcp.get_tool("calc", VersionSpec(eq="2.0"))
|
|
assert tool is None
|
|
|
|
async def test_get_tool_all_versions_disabled_returns_none(self):
|
|
"""When all versions are disabled, get_tool returns None."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
mcp.disable(names={"calc"})
|
|
|
|
tool = await mcp.get_tool("calc")
|
|
assert tool is None
|
|
|
|
async def test_get_tool_middle_version_fallback(self):
|
|
"""Disabling v3 should fall back to v2, not v1."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
@mcp.tool(version="3.0")
|
|
def calc() -> int:
|
|
return 3
|
|
|
|
mcp.disable(version=VersionSpec(eq="3.0"))
|
|
|
|
tool = await mcp.get_tool("calc")
|
|
assert tool is not None
|
|
assert tool.version == "2.0"
|
|
|
|
|
|
class TestResourceVersionFallback:
|
|
"""Test that disabling the latest resource version falls back correctly."""
|
|
|
|
async def test_get_resource_returns_v1_when_v2_disabled(self):
|
|
"""get_resource should return v1 when v2 is disabled."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://info", version="1.0")
|
|
def info() -> str:
|
|
return "v1"
|
|
|
|
@mcp.resource("data://info", version="2.0")
|
|
def info() -> str:
|
|
return "v2"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
resource = await mcp.get_resource("data://info")
|
|
assert resource is not None
|
|
assert resource.version == "1.0"
|
|
|
|
async def test_get_resource_explicit_disabled_version_returns_none(self):
|
|
"""Requesting a specific disabled resource version should return None."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://info", version="1.0")
|
|
def info() -> str:
|
|
return "v1"
|
|
|
|
@mcp.resource("data://info", version="2.0")
|
|
def info() -> str:
|
|
return "v2"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
resource = await mcp.get_resource("data://info", VersionSpec(eq="2.0"))
|
|
assert resource is None
|
|
|
|
|
|
class TestResourceTemplateVersionFallback:
|
|
"""Test that disabling the latest template version falls back correctly."""
|
|
|
|
async def test_get_resource_template_returns_v1_when_v2_disabled(self):
|
|
"""get_resource_template should return v1 when v2 is disabled."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://items/{id}", version="1.0")
|
|
def item(id: str) -> str:
|
|
return f"v1-{id}"
|
|
|
|
@mcp.resource("data://items/{id}", version="2.0")
|
|
def item(id: str) -> str:
|
|
return f"v2-{id}"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
template = await mcp.get_resource_template("data://items/{id}")
|
|
assert template is not None
|
|
assert template.version == "1.0"
|
|
|
|
|
|
class TestPromptVersionFallback:
|
|
"""Test that disabling the latest prompt version falls back correctly."""
|
|
|
|
async def test_get_prompt_returns_v1_when_v2_disabled(self):
|
|
"""get_prompt should return v1 when v2 is disabled."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet() -> str:
|
|
return "hello v1"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet() -> str:
|
|
return "hello v2"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
prompt = await mcp.get_prompt("greet")
|
|
assert prompt is not None
|
|
assert prompt.version == "1.0"
|
|
|
|
async def test_get_prompt_explicit_disabled_version_returns_none(self):
|
|
"""Requesting a specific disabled prompt version should return None."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0")
|
|
def greet() -> str:
|
|
return "hello v1"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet() -> str:
|
|
return "hello v2"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
prompt = await mcp.get_prompt("greet", VersionSpec(eq="2.0"))
|
|
assert prompt is None
|
|
|
|
|
|
class TestFallbackRespectsAuth:
|
|
"""Fallback to older versions must enforce auth checks.
|
|
|
|
When the highest version is disabled and the code falls back to older
|
|
versions, those candidates must go through auth filtering. Otherwise
|
|
a protected v1 could be exposed to unauthenticated users when a
|
|
public v2 is disabled.
|
|
"""
|
|
|
|
async def test_tool_fallback_respects_auth(self):
|
|
"""Disabling v2 should not expose auth-protected v1 to unauthorized users."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0", auth=require_scopes("admin"))
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
# Without an admin token, v1 should NOT be returned
|
|
tool = await mcp.get_tool("calc")
|
|
assert tool is None
|
|
|
|
async def test_tool_fallback_allows_authorized_user(self):
|
|
"""Fallback should return auth-protected v1 to authorized users."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0", auth=require_scopes("admin"))
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0")
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
token = _make_token(scopes=["admin"])
|
|
tok = _set_token(token)
|
|
try:
|
|
tool = await mcp.get_tool("calc")
|
|
assert tool is not None
|
|
assert tool.version == "1.0"
|
|
finally:
|
|
auth_context_var.reset(tok)
|
|
|
|
async def test_resource_fallback_respects_auth(self):
|
|
"""Disabling v2 should not expose auth-protected v1 resource."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://info", version="1.0", auth=require_scopes("admin"))
|
|
def info() -> str:
|
|
return "v1"
|
|
|
|
@mcp.resource("data://info", version="2.0")
|
|
def info() -> str:
|
|
return "v2"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
resource = await mcp.get_resource("data://info")
|
|
assert resource is None
|
|
|
|
async def test_resource_template_fallback_respects_auth(self):
|
|
"""Disabling v2 should not expose auth-protected v1 template."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.resource("data://items/{id}", version="1.0", auth=require_scopes("admin"))
|
|
def item(id: str) -> str:
|
|
return f"v1-{id}"
|
|
|
|
@mcp.resource("data://items/{id}", version="2.0")
|
|
def item(id: str) -> str:
|
|
return f"v2-{id}"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
template = await mcp.get_resource_template("data://items/{id}")
|
|
assert template is None
|
|
|
|
async def test_prompt_fallback_respects_auth(self):
|
|
"""Disabling v2 should not expose auth-protected v1 prompt."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.prompt(version="1.0", auth=require_scopes("admin"))
|
|
def greet() -> str:
|
|
return "hello v1"
|
|
|
|
@mcp.prompt(version="2.0")
|
|
def greet() -> str:
|
|
return "hello v2"
|
|
|
|
mcp.disable(version=VersionSpec(eq="2.0"))
|
|
|
|
prompt = await mcp.get_prompt("greet")
|
|
assert prompt is None
|
|
|
|
async def test_fallback_skips_unauthorized_picks_next(self):
|
|
"""When multiple fallback candidates exist, skip unauthorized ones."""
|
|
mcp = FastMCP()
|
|
|
|
@mcp.tool(version="1.0")
|
|
def calc() -> int:
|
|
return 1
|
|
|
|
@mcp.tool(version="2.0", auth=require_scopes("admin"))
|
|
def calc() -> int:
|
|
return 2
|
|
|
|
@mcp.tool(version="3.0")
|
|
def calc() -> int:
|
|
return 3
|
|
|
|
mcp.disable(version=VersionSpec(eq="3.0"))
|
|
|
|
# v2 requires admin, so unauthorized user should get v1
|
|
tool = await mcp.get_tool("calc")
|
|
assert tool is not None
|
|
assert tool.version == "1.0"
|