mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Merge branch 'main' into copy-flag
This commit is contained in:
Jeremiah Lowin
GitHub
commit
f6fe3f5785
7 changed files with 221 additions and 396 deletions
|
|
@ -1,5 +1,3 @@
|
|||
"""Tests for the main CLI functionality."""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
|
@ -52,7 +50,7 @@ class TestMainCLI:
|
|||
"--with",
|
||||
"fastmcp",
|
||||
"--with-editable",
|
||||
"/path/to/package",
|
||||
str(editable_path),
|
||||
"fastmcp",
|
||||
"run",
|
||||
"server.py",
|
||||
|
|
@ -96,17 +94,19 @@ class TestMainCLI:
|
|||
class TestVersionCommand:
|
||||
"""Test the version command."""
|
||||
|
||||
@patch("fastmcp.cli.cli.sys.exit")
|
||||
@patch("fastmcp.cli.cli.console.print")
|
||||
def test_version_command(self, mock_print, mock_exit):
|
||||
"""Test that version command prints info and exits."""
|
||||
# Parse and execute version command
|
||||
def test_version_command_parsing(self):
|
||||
"""Test that version command can be parsed."""
|
||||
command, bound, _ = app.parse_args(["version"])
|
||||
command()
|
||||
assert command is not None
|
||||
|
||||
# Verify it printed something and exited with 0
|
||||
mock_print.assert_called_once()
|
||||
mock_exit.assert_called_once_with(0)
|
||||
def test_version_command_execution(self):
|
||||
"""Test that version command executes and exits properly."""
|
||||
# The version command should exit with code 0 when executed
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
command, bound, _ = app.parse_args(["version"])
|
||||
command()
|
||||
|
||||
assert exc_info.value.code == 0
|
||||
|
||||
def test_version_command_parsing(self):
|
||||
"""Test that the version command parses arguments correctly."""
|
||||
|
|
@ -177,25 +177,21 @@ class TestDevCommand:
|
|||
class TestRunCommand:
|
||||
"""Test the run command."""
|
||||
|
||||
@patch("fastmcp.cli.cli.run_module.run_command")
|
||||
def test_run_command_basic(self, mock_run_command):
|
||||
"""Test basic run command."""
|
||||
def test_run_command_parsing_basic(self):
|
||||
"""Test basic run command parsing."""
|
||||
command, bound, _ = app.parse_args(["run", "server.py"])
|
||||
command(**bound.arguments)
|
||||
|
||||
mock_run_command.assert_called_once_with(
|
||||
server_spec="server.py",
|
||||
transport=None,
|
||||
host=None,
|
||||
port=None,
|
||||
log_level=None,
|
||||
server_args=[],
|
||||
show_banner=True,
|
||||
)
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
# Cyclopts only includes non-default values
|
||||
assert "transport" not in bound.arguments
|
||||
assert "host" not in bound.arguments
|
||||
assert "port" not in bound.arguments
|
||||
assert "log_level" not in bound.arguments
|
||||
assert "no_banner" not in bound.arguments
|
||||
|
||||
@patch("fastmcp.cli.cli.run_module.run_command")
|
||||
def test_run_command_with_options(self, mock_run_command):
|
||||
"""Test run command with various options."""
|
||||
def test_run_command_parsing_with_options(self):
|
||||
"""Test run command parsing with various options."""
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"run",
|
||||
|
|
@ -211,28 +207,35 @@ class TestRunCommand:
|
|||
"--no-banner",
|
||||
]
|
||||
)
|
||||
command(**bound.arguments)
|
||||
|
||||
mock_run_command.assert_called_once_with(
|
||||
server_spec="server.py",
|
||||
transport="http",
|
||||
host="localhost",
|
||||
port=8080,
|
||||
log_level="DEBUG",
|
||||
server_args=[],
|
||||
show_banner=False,
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
assert bound.arguments["transport"] == "http"
|
||||
assert bound.arguments["host"] == "localhost"
|
||||
assert bound.arguments["port"] == 8080
|
||||
assert bound.arguments["log_level"] == "DEBUG"
|
||||
assert bound.arguments["no_banner"] is True
|
||||
|
||||
def test_run_command_parsing_partial_options(self):
|
||||
"""Test run command parsing with only some options."""
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"run",
|
||||
"server.py",
|
||||
"--transport",
|
||||
"http",
|
||||
"--no-banner",
|
||||
]
|
||||
)
|
||||
|
||||
@patch("fastmcp.cli.cli.run_module.run_command")
|
||||
def test_run_command_failure(self, mock_run_command):
|
||||
"""Test run command handling failures."""
|
||||
mock_run_command.side_effect = Exception("Test error")
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
command, bound, _ = app.parse_args(["run", "server.py"])
|
||||
command(**bound.arguments)
|
||||
|
||||
assert exc_info.value.code == 1
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
assert bound.arguments["transport"] == "http"
|
||||
assert bound.arguments["no_banner"] is True
|
||||
# Other options should not be present
|
||||
assert "host" not in bound.arguments
|
||||
assert "port" not in bound.arguments
|
||||
assert "log_level" not in bound.arguments
|
||||
|
||||
|
||||
class TestWindowsSpecific:
|
||||
|
|
@ -317,89 +320,93 @@ class TestWindowsSpecific:
|
|||
assert result == "npx"
|
||||
mock_run.assert_not_called()
|
||||
|
||||
def test_windows_path_parsing_with_colon(self):
|
||||
def test_windows_path_parsing_with_colon(self, tmp_path):
|
||||
"""Test parsing Windows paths with drive letters and colons."""
|
||||
from fastmcp.cli.run import parse_file_path
|
||||
|
||||
# We can't test actual Windows paths on non-Windows systems,
|
||||
# but we can test the logic with mock paths
|
||||
with patch("pathlib.Path.exists") as mock_exists:
|
||||
with patch("pathlib.Path.is_file") as mock_is_file:
|
||||
mock_exists.return_value = True
|
||||
mock_is_file.return_value = True
|
||||
# Create a real test file to test the logic
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("# test server")
|
||||
|
||||
# Test that C:\path\file.py is parsed correctly
|
||||
with patch("pathlib.Path.resolve") as mock_resolve:
|
||||
mock_resolve.return_value = Path("C:/path/file.py")
|
||||
# Test normal file parsing (works on all platforms)
|
||||
file_path, obj = parse_file_path(str(test_file))
|
||||
assert obj is None
|
||||
|
||||
file_path, obj = parse_file_path("C:\\path\\file.py")
|
||||
assert obj is None
|
||||
# Test file:object parsing
|
||||
file_path, obj = parse_file_path(f"{test_file}:myapp")
|
||||
assert obj == "myapp"
|
||||
|
||||
# Test C:\path\file.py:object parsing
|
||||
with patch("pathlib.Path.resolve") as mock_resolve:
|
||||
mock_resolve.return_value = Path("C:/path/file.py")
|
||||
|
||||
file_path, obj = parse_file_path("C:\\path\\file.py:myapp")
|
||||
assert obj == "myapp"
|
||||
# Test that the file portion resolves correctly when object is specified
|
||||
assert file_path == test_file.resolve()
|
||||
|
||||
|
||||
class TestInspectCommand:
|
||||
"""Test the inspect command."""
|
||||
|
||||
@patch("fastmcp.cli.cli.run_module.parse_file_path")
|
||||
@patch("fastmcp.cli.cli.run_module.import_server")
|
||||
@patch("fastmcp.cli.cli.inspect_fastmcp")
|
||||
def test_inspect_command_basic(
|
||||
self, mock_inspect, mock_import_server, mock_parse_file_path, tmp_path
|
||||
):
|
||||
"""Test basic inspect command functionality."""
|
||||
# Setup mocks
|
||||
mock_parse_file_path.return_value = (Path("server.py"), None)
|
||||
mock_server = Mock()
|
||||
mock_import_server.return_value = mock_server
|
||||
def test_inspect_command_parsing_basic(self):
|
||||
"""Test basic inspect command parsing."""
|
||||
command, bound, _ = app.parse_args(["inspect", "server.py"])
|
||||
|
||||
mock_info = Mock()
|
||||
mock_info.name = "TestServer"
|
||||
mock_info.tools = []
|
||||
mock_info.prompts = []
|
||||
mock_info.resources = []
|
||||
mock_info.templates = []
|
||||
mock_inspect.return_value = mock_info
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
# Only explicitly set parameters are in bound.arguments
|
||||
assert "output" not in bound.arguments
|
||||
|
||||
# Mock TypeAdapter
|
||||
with patch("fastmcp.cli.cli.TypeAdapter") as mock_adapter:
|
||||
mock_adapter.return_value.dump_json.return_value = b'{"name": "TestServer"}'
|
||||
def test_inspect_command_parsing_with_output(self, tmp_path):
|
||||
"""Test inspect command parsing with output file."""
|
||||
output_file = tmp_path / "output.json"
|
||||
|
||||
output_file = tmp_path / "test-output.json"
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"inspect",
|
||||
"server.py",
|
||||
"--output",
|
||||
str(output_file),
|
||||
]
|
||||
)
|
||||
|
||||
# Parse and execute
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"inspect",
|
||||
"server.py",
|
||||
"--output",
|
||||
str(output_file),
|
||||
]
|
||||
)
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
# Output is parsed as a Path object
|
||||
assert bound.arguments["output"] == output_file
|
||||
|
||||
# This is an async command, so we need to run it
|
||||
import asyncio
|
||||
async def test_inspect_command_with_real_server(self, tmp_path):
|
||||
"""Test inspect command with a real server file."""
|
||||
# Create a real server file
|
||||
server_file = tmp_path / "test_server.py"
|
||||
server_file.write_text("""
|
||||
import fastmcp
|
||||
|
||||
asyncio.run(command(**bound.arguments))
|
||||
mcp = fastmcp.FastMCP("InspectTestServer")
|
||||
|
||||
# Verify the output file was created
|
||||
@mcp.tool
|
||||
def test_tool(x: int) -> int:
|
||||
return x * 2
|
||||
|
||||
@mcp.prompt
|
||||
def test_prompt(name: str) -> str:
|
||||
return f"Hello, {name}!"
|
||||
""")
|
||||
|
||||
output_file = tmp_path / "inspect_output.json"
|
||||
|
||||
# Parse and execute the command
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"inspect",
|
||||
str(server_file),
|
||||
"--output",
|
||||
str(output_file),
|
||||
]
|
||||
)
|
||||
|
||||
await command(**bound.arguments)
|
||||
|
||||
# Verify the output file was created and contains expected content
|
||||
assert output_file.exists()
|
||||
assert output_file.read_text() == '{"name": "TestServer"}'
|
||||
content = output_file.read_text()
|
||||
|
||||
@patch("fastmcp.cli.cli.run_module.import_server")
|
||||
def test_inspect_command_failure(self, mock_import_server):
|
||||
"""Test inspect command handling failures."""
|
||||
mock_import_server.side_effect = Exception("Import failed")
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
command, bound, _ = app.parse_args(["inspect", "server.py"])
|
||||
import asyncio
|
||||
|
||||
asyncio.run(command(**bound.arguments))
|
||||
|
||||
assert exc_info.value.code == 1
|
||||
# Basic checks that the inspection worked
|
||||
assert "InspectTestServer" in content
|
||||
assert "test_tool" in content
|
||||
assert "test_prompt" in content
|
||||
|
|
|
|||
|
|
@ -1,5 +1,3 @@
|
|||
"""Tests for Cursor integration functionality."""
|
||||
|
||||
import base64
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
|
@ -257,7 +255,9 @@ class TestInstallCursor:
|
|||
config_data = json.loads(decoded)
|
||||
|
||||
assert "--with-editable" in config_data["args"]
|
||||
assert "/local/package" in config_data["args"]
|
||||
# Check for the editable path in a platform-agnostic way
|
||||
editable_path_str = str(Path("/local/package"))
|
||||
assert editable_path_str in config_data["args"]
|
||||
assert "server.py:custom_app" in " ".join(config_data["args"])
|
||||
|
||||
@patch("fastmcp.cli.install.cursor.open_deeplink")
|
||||
|
|
|
|||
|
|
@ -1,5 +1,3 @@
|
|||
"""Tests for the install subcommands."""
|
||||
|
||||
from fastmcp.cli.install import install_app
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,17 +1,9 @@
|
|||
"""Tests for the run module functionality."""
|
||||
|
||||
import sys
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from fastmcp.cli.run import (
|
||||
create_client_server,
|
||||
import_server,
|
||||
import_server_with_args,
|
||||
is_url,
|
||||
parse_file_path,
|
||||
run_command,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -87,275 +79,121 @@ class TestFilePathParsing:
|
|||
parse_file_path(str(tmp_path))
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific test")
|
||||
def test_parse_file_path_windows_drive(self, tmp_path):
|
||||
"""Test parsing Windows path with drive letter."""
|
||||
# This test would only work on Windows with actual drive letters
|
||||
# For now, just test the logic doesn't break with colons
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("# test server")
|
||||
|
||||
# Should handle paths that might look like Windows drives
|
||||
file_path, server_object = parse_file_path(str(test_file))
|
||||
assert file_path == test_file.resolve()
|
||||
assert server_object is None
|
||||
|
||||
|
||||
class TestServerImport:
|
||||
"""Test server import functionality."""
|
||||
"""Test server import functionality using real files."""
|
||||
|
||||
def test_import_server_with_standard_name(self, tmp_path):
|
||||
"""Test importing server with standard object name."""
|
||||
async def test_import_server_basic_mcp(self, tmp_path):
|
||||
"""Test importing server with basic FastMCP server."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("""
|
||||
import fastmcp
|
||||
|
||||
mcp = fastmcp.FastMCP("TestServer")
|
||||
|
||||
@mcp.tool
|
||||
def greet(name: str) -> str:
|
||||
return f"Hello, {name}!"
|
||||
""")
|
||||
|
||||
server = import_server(test_file)
|
||||
assert server.name == "TestServer"
|
||||
tools = await server.get_tools()
|
||||
assert "greet" in tools
|
||||
|
||||
async def test_import_server_with_main_block(self, tmp_path):
|
||||
"""Test importing server with if __name__ == '__main__' block."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("""
|
||||
import fastmcp
|
||||
|
||||
app = fastmcp.FastMCP("MainServer")
|
||||
|
||||
@app.tool
|
||||
def calculate(x: int, y: int) -> int:
|
||||
return x + y
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run()
|
||||
""")
|
||||
|
||||
server = import_server(test_file)
|
||||
assert server.name == "MainServer"
|
||||
tools = await server.get_tools()
|
||||
assert "calculate" in tools
|
||||
|
||||
def test_import_server_standard_names(self, tmp_path):
|
||||
"""Test automatic detection of standard names (mcp, server, app)."""
|
||||
# Test with 'mcp' name
|
||||
mcp_file = tmp_path / "mcp_server.py"
|
||||
mcp_file.write_text("""
|
||||
import fastmcp
|
||||
mcp = fastmcp.FastMCP("MCPServer")
|
||||
""")
|
||||
|
||||
server = import_server(mcp_file)
|
||||
assert server.name == "MCPServer"
|
||||
|
||||
# Test with 'server' name
|
||||
server_file = tmp_path / "server_server.py"
|
||||
server_file.write_text("""
|
||||
import fastmcp
|
||||
server = fastmcp.FastMCP("ServerServer")
|
||||
""")
|
||||
|
||||
server = import_server(server_file)
|
||||
assert server.name == "ServerServer"
|
||||
|
||||
# Test with 'app' name
|
||||
app_file = tmp_path / "app_server.py"
|
||||
app_file.write_text("""
|
||||
import fastmcp
|
||||
app = fastmcp.FastMCP("AppServer")
|
||||
""")
|
||||
|
||||
server = import_server(app_file)
|
||||
assert server.name == "AppServer"
|
||||
|
||||
async def test_import_server_nonstandard_name(self, tmp_path):
|
||||
"""Test importing server with non-standard object name."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("""
|
||||
import fastmcp
|
||||
|
||||
my_custom_server = fastmcp.FastMCP("CustomServer")
|
||||
|
||||
@my_custom_server.tool
|
||||
def custom_tool() -> str:
|
||||
return "custom"
|
||||
""")
|
||||
|
||||
server = import_server(test_file, "my_custom_server")
|
||||
assert server.name == "CustomServer"
|
||||
tools = await server.get_tools()
|
||||
assert "custom_tool" in tools
|
||||
|
||||
def test_import_server_no_standard_names_fails(self, tmp_path):
|
||||
"""Test importing server when no standard names exist fails."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("""
|
||||
import fastmcp
|
||||
|
||||
other_name = fastmcp.FastMCP("OtherServer")
|
||||
""")
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
import_server(test_file)
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
def test_import_server_nonexistent_object_fails(self, tmp_path):
|
||||
"""Test importing nonexistent server object fails."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("""
|
||||
import fastmcp
|
||||
|
||||
mcp = fastmcp.FastMCP("TestServer")
|
||||
""")
|
||||
|
||||
with patch("fastmcp.cli.run.sys.path") as mock_path:
|
||||
mock_path.__contains__ = Mock(return_value=False)
|
||||
mock_path.insert = Mock()
|
||||
|
||||
# Mock the actual import process
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.spec_from_file_location"
|
||||
) as mock_spec_from_file:
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.module_from_spec"
|
||||
) as mock_module_from_spec:
|
||||
# Setup mock module
|
||||
mock_module = Mock()
|
||||
mock_module.mcp = Mock()
|
||||
mock_module_from_spec.return_value = mock_module
|
||||
|
||||
# Setup mock spec
|
||||
mock_spec = Mock()
|
||||
mock_spec.loader = Mock()
|
||||
mock_spec_from_file.return_value = mock_spec
|
||||
|
||||
server = import_server(test_file)
|
||||
assert server == mock_module.mcp
|
||||
|
||||
def test_import_server_with_custom_object(self, tmp_path):
|
||||
"""Test importing server with custom object name."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("""
|
||||
import fastmcp
|
||||
my_app = fastmcp.FastMCP("TestServer")
|
||||
""")
|
||||
|
||||
with patch("fastmcp.cli.run.sys.path") as mock_path:
|
||||
mock_path.__contains__ = Mock(return_value=False)
|
||||
mock_path.insert = Mock()
|
||||
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.spec_from_file_location"
|
||||
) as mock_spec_from_file:
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.module_from_spec"
|
||||
) as mock_module_from_spec:
|
||||
mock_module = Mock()
|
||||
mock_module.my_app = Mock()
|
||||
mock_module_from_spec.return_value = mock_module
|
||||
|
||||
mock_spec = Mock()
|
||||
mock_spec.loader = Mock()
|
||||
mock_spec_from_file.return_value = mock_spec
|
||||
|
||||
server = import_server(test_file, "my_app")
|
||||
assert server == mock_module.my_app
|
||||
|
||||
def test_import_server_no_standard_names(self, tmp_path):
|
||||
"""Test importing server when no standard names exist."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("# No server objects")
|
||||
|
||||
with patch("fastmcp.cli.run.sys.path"):
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.spec_from_file_location"
|
||||
) as mock_spec_from_file:
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.module_from_spec"
|
||||
) as mock_module_from_spec:
|
||||
mock_module = Mock()
|
||||
|
||||
# Mock hasattr behavior for standard names
|
||||
def mock_hasattr(obj, name):
|
||||
return name not in ["mcp", "server", "app"]
|
||||
|
||||
with patch("builtins.hasattr", side_effect=mock_hasattr):
|
||||
mock_module_from_spec.return_value = mock_module
|
||||
|
||||
mock_spec = Mock()
|
||||
mock_spec.loader = Mock()
|
||||
mock_spec_from_file.return_value = mock_spec
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
import_server(test_file)
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
def test_import_server_nonexistent_object(self, tmp_path):
|
||||
"""Test importing nonexistent server object."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("# No server objects")
|
||||
|
||||
with patch("fastmcp.cli.run.sys.path"):
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.spec_from_file_location"
|
||||
) as mock_spec_from_file:
|
||||
with patch(
|
||||
"fastmcp.cli.run.importlib.util.module_from_spec"
|
||||
) as mock_module_from_spec:
|
||||
mock_module = Mock()
|
||||
mock_module.nonexistent = None
|
||||
mock_module_from_spec.return_value = mock_module
|
||||
|
||||
mock_spec = Mock()
|
||||
mock_spec.loader = Mock()
|
||||
mock_spec_from_file.return_value = mock_spec
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
import_server(test_file, "nonexistent")
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
class TestServerImportWithArgs:
|
||||
"""Test server import with command line arguments."""
|
||||
|
||||
@patch("fastmcp.cli.run.import_server")
|
||||
def test_import_server_with_args(self, mock_import_server, tmp_path):
|
||||
"""Test importing server with command line arguments."""
|
||||
test_file = tmp_path / "server.py"
|
||||
mock_server = Mock()
|
||||
mock_import_server.return_value = mock_server
|
||||
|
||||
original_argv = sys.argv[:]
|
||||
try:
|
||||
result = import_server_with_args(
|
||||
test_file, "app", ["--config", "test.json", "--debug"]
|
||||
)
|
||||
|
||||
assert result == mock_server
|
||||
mock_import_server.assert_called_once_with(test_file, "app")
|
||||
|
||||
finally:
|
||||
sys.argv = original_argv
|
||||
|
||||
@patch("fastmcp.cli.run.import_server")
|
||||
def test_import_server_no_args(self, mock_import_server, tmp_path):
|
||||
"""Test importing server without command line arguments."""
|
||||
test_file = tmp_path / "server.py"
|
||||
mock_server = Mock()
|
||||
mock_import_server.return_value = mock_server
|
||||
|
||||
result = import_server_with_args(test_file, "app")
|
||||
|
||||
assert result == mock_server
|
||||
mock_import_server.assert_called_once_with(test_file, "app")
|
||||
|
||||
|
||||
class TestClientServer:
|
||||
"""Test client server creation."""
|
||||
|
||||
def test_create_client_server(self):
|
||||
"""Test creating server from client URL."""
|
||||
# Patch the import at the builtins level since it's a local import
|
||||
with patch("builtins.__import__") as mock_import:
|
||||
mock_fastmcp = Mock()
|
||||
mock_import.return_value = mock_fastmcp
|
||||
|
||||
mock_client = Mock()
|
||||
mock_server = Mock()
|
||||
mock_fastmcp.Client.return_value = mock_client
|
||||
mock_fastmcp.FastMCP.from_client.return_value = mock_server
|
||||
|
||||
result = create_client_server("http://example.com")
|
||||
|
||||
assert result == mock_server
|
||||
mock_fastmcp.Client.assert_called_once_with("http://example.com")
|
||||
mock_fastmcp.FastMCP.from_client.assert_called_once_with(mock_client)
|
||||
|
||||
def test_create_client_server_failure(self):
|
||||
"""Test client server creation failure."""
|
||||
with patch("builtins.__import__") as mock_import:
|
||||
mock_fastmcp = Mock()
|
||||
mock_import.return_value = mock_fastmcp
|
||||
mock_fastmcp.Client.side_effect = Exception("Connection failed")
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
create_client_server("http://example.com")
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
class TestRunCommand:
|
||||
"""Test the main run command functionality."""
|
||||
|
||||
@patch("fastmcp.cli.run.create_client_server")
|
||||
def test_run_command_url(self, mock_create_client_server):
|
||||
"""Test running command with URL."""
|
||||
mock_server = Mock()
|
||||
mock_create_client_server.return_value = mock_server
|
||||
|
||||
run_command("http://example.com")
|
||||
|
||||
mock_create_client_server.assert_called_once_with("http://example.com")
|
||||
mock_server.run.assert_called_once()
|
||||
|
||||
@patch("fastmcp.cli.run.import_server_with_args")
|
||||
@patch("fastmcp.cli.run.parse_file_path")
|
||||
def test_run_command_file(self, mock_parse_file_path, mock_import_server):
|
||||
"""Test running command with file path."""
|
||||
mock_file = Mock()
|
||||
mock_parse_file_path.return_value = (mock_file, "app")
|
||||
mock_server = Mock()
|
||||
mock_server.name = "TestServer"
|
||||
mock_import_server.return_value = mock_server
|
||||
|
||||
run_command("server.py:app")
|
||||
|
||||
mock_parse_file_path.assert_called_once_with("server.py:app")
|
||||
mock_import_server.assert_called_once_with(mock_file, "app", None)
|
||||
mock_server.run.assert_called_once()
|
||||
|
||||
@patch("fastmcp.cli.run.import_server_with_args")
|
||||
@patch("fastmcp.cli.run.parse_file_path")
|
||||
def test_run_command_with_options(self, mock_parse_file_path, mock_import_server):
|
||||
"""Test running command with various options."""
|
||||
mock_file = Mock()
|
||||
mock_parse_file_path.return_value = (mock_file, None)
|
||||
mock_server = Mock()
|
||||
mock_server.name = "TestServer"
|
||||
mock_import_server.return_value = mock_server
|
||||
|
||||
run_command(
|
||||
"server.py",
|
||||
transport="http",
|
||||
host="localhost",
|
||||
port=8080,
|
||||
log_level="DEBUG",
|
||||
server_args=["--config", "test.json"],
|
||||
show_banner=False,
|
||||
)
|
||||
|
||||
mock_server.run.assert_called_once_with(
|
||||
transport="http",
|
||||
host="localhost",
|
||||
port=8080,
|
||||
log_level="DEBUG",
|
||||
show_banner=False,
|
||||
)
|
||||
|
||||
@patch("fastmcp.cli.run.import_server_with_args")
|
||||
@patch("fastmcp.cli.run.parse_file_path")
|
||||
def test_run_command_server_failure(self, mock_parse_file_path, mock_import_server):
|
||||
"""Test run command when server run fails."""
|
||||
mock_file = Mock()
|
||||
mock_parse_file_path.return_value = (mock_file, None)
|
||||
mock_server = Mock()
|
||||
mock_server.name = "TestServer"
|
||||
mock_server.run.side_effect = Exception("Server failed")
|
||||
mock_import_server.return_value = mock_server
|
||||
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
run_command("server.py")
|
||||
import_server(test_file, "nonexistent")
|
||||
assert exc_info.value.code == 1
|
||||
|
|
|
|||
|
|
@ -1,5 +1,3 @@
|
|||
"""Tests for shared CLI functionality."""
|
||||
|
||||
from fastmcp.cli.cli import _parse_env_var
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -566,7 +566,6 @@ class TestComponentManagerWithPath:
|
|||
def client_with_path(self, mcp_with_path):
|
||||
return TestClient(mcp_with_path.http_app())
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_enable_tool_route_with_path(self, client_with_path, mcp_with_path):
|
||||
tool = await mcp_with_path._tool_manager.get_tool("test_tool")
|
||||
tool.enabled = False
|
||||
|
|
@ -576,7 +575,6 @@ class TestComponentManagerWithPath:
|
|||
tool = await mcp_with_path._tool_manager.get_tool("test_tool")
|
||||
assert tool.enabled is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_disable_resource_route_with_path(
|
||||
self, client_with_path, mcp_with_path
|
||||
):
|
||||
|
|
@ -592,7 +590,6 @@ class TestComponentManagerWithPath:
|
|||
)
|
||||
assert resource.enabled is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_enable_prompt_route_with_path(self, client_with_path, mcp_with_path):
|
||||
prompt = await mcp_with_path._prompt_manager.get_prompt("test_prompt")
|
||||
prompt.enabled = False
|
||||
|
|
@ -646,7 +643,6 @@ class TestComponentManagerWithPathAuth:
|
|||
|
||||
self.client = TestClient(self.mcp.http_app())
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unauthorized_enable_tool(self):
|
||||
tool = await self.mcp._tool_manager.get_tool("test_tool")
|
||||
tool.enabled = False
|
||||
|
|
@ -654,7 +650,6 @@ class TestComponentManagerWithPathAuth:
|
|||
assert response.status_code == 401
|
||||
assert tool.enabled is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forbidden_enable_tool(self):
|
||||
tool = await self.mcp._tool_manager.get_tool("test_tool")
|
||||
tool.enabled = False
|
||||
|
|
@ -665,7 +660,6 @@ class TestComponentManagerWithPathAuth:
|
|||
assert response.status_code == 403
|
||||
assert tool.enabled is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_authorized_enable_tool(self):
|
||||
tool = await self.mcp._tool_manager.get_tool("test_tool")
|
||||
tool.enabled = False
|
||||
|
|
@ -678,7 +672,6 @@ class TestComponentManagerWithPathAuth:
|
|||
tool = await self.mcp._tool_manager.get_tool("test_tool")
|
||||
assert tool.enabled is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unauthorized_disable_resource(self):
|
||||
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
|
||||
resource.enabled = True
|
||||
|
|
@ -686,7 +679,6 @@ class TestComponentManagerWithPathAuth:
|
|||
assert response.status_code == 401
|
||||
assert resource.enabled is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forbidden_disable_resource(self):
|
||||
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
|
||||
resource.enabled = True
|
||||
|
|
@ -697,7 +689,6 @@ class TestComponentManagerWithPathAuth:
|
|||
assert response.status_code == 403
|
||||
assert resource.enabled is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_authorized_disable_resource(self):
|
||||
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
|
||||
resource.enabled = True
|
||||
|
|
@ -710,7 +701,6 @@ class TestComponentManagerWithPathAuth:
|
|||
resource = await self.mcp._resource_manager.get_resource("data://test_resource")
|
||||
assert resource.enabled is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unauthorized_enable_prompt(self):
|
||||
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
|
||||
prompt.enabled = False
|
||||
|
|
@ -718,7 +708,6 @@ class TestComponentManagerWithPathAuth:
|
|||
assert response.status_code == 401
|
||||
assert prompt.enabled is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forbidden_enable_prompt(self):
|
||||
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
|
||||
prompt.enabled = False
|
||||
|
|
@ -729,7 +718,6 @@ class TestComponentManagerWithPathAuth:
|
|||
assert response.status_code == 403
|
||||
assert prompt.enabled is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_authorized_enable_prompt(self):
|
||||
prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
|
||||
prompt.enabled = False
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ specifications and properly applied during HTTP request serialization.
|
|||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from fastmcp.server.openapi import OpenAPITool
|
||||
from fastmcp.utilities.openapi import parse_openapi_to_http_routes
|
||||
|
|
@ -130,7 +129,6 @@ class TestExplodeIntegration:
|
|||
f"Expected explode=None, got {parameter.explode}"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_explode_false_request_serialization(self):
|
||||
"""Test that explode=false results in comma-separated query parameters in HTTP requests.
|
||||
|
||||
|
|
@ -201,7 +199,6 @@ class TestExplodeIntegration:
|
|||
f"Expected 'red,blue,green', got '{tags_value}'"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_explode_true_request_serialization(self):
|
||||
"""Test that explode=true results in separate query parameters in HTTP requests."""
|
||||
openapi_spec = {
|
||||
|
|
@ -262,7 +259,6 @@ class TestExplodeIntegration:
|
|||
f"Expected ['red', 'blue', 'green'], got {tags_value}"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_explode_default_request_serialization(self):
|
||||
"""Test that default behavior (no explode) uses explode=true for query parameters."""
|
||||
openapi_spec = {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue