mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Add tests
This commit is contained in:
parent
919d7e35ef
commit
6694d9a19a
4 changed files with 788 additions and 0 deletions
1
tests/cli/__init__.py
Normal file
1
tests/cli/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""CLI test package."""
|
||||
259
tests/cli/test_cli.py
Normal file
259
tests/cli/test_cli.py
Normal file
|
|
@ -0,0 +1,259 @@
|
|||
"""Tests for the main CLI functionality."""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from fastmcp.cli.cli import _build_uv_command, _parse_env_var, app
|
||||
|
||||
|
||||
class TestMainCLI:
|
||||
"""Test the main CLI application."""
|
||||
|
||||
def test_app_exists(self):
|
||||
"""Test that the main app is properly configured."""
|
||||
# app.name is a tuple in cyclopts
|
||||
assert "fastmcp" in app.name
|
||||
assert "FastMCP 2.0" in app.help
|
||||
# Just check that version exists, not the specific value
|
||||
assert hasattr(app, "version")
|
||||
|
||||
def test_parse_env_var_valid(self):
|
||||
"""Test parsing valid environment variables."""
|
||||
key, value = _parse_env_var("KEY=value")
|
||||
assert key == "KEY"
|
||||
assert value == "value"
|
||||
|
||||
key, value = _parse_env_var("COMPLEX_KEY=complex=value=with=equals")
|
||||
assert key == "COMPLEX_KEY"
|
||||
assert value == "complex=value=with=equals"
|
||||
|
||||
def test_parse_env_var_invalid(self):
|
||||
"""Test parsing invalid environment variables exits."""
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_parse_env_var("INVALID_FORMAT")
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
def test_build_uv_command_basic(self):
|
||||
"""Test building basic uv command."""
|
||||
cmd = _build_uv_command("server.py")
|
||||
expected = ["uv", "run", "--with", "fastmcp", "fastmcp", "run", "server.py"]
|
||||
assert cmd == expected
|
||||
|
||||
def test_build_uv_command_with_editable(self):
|
||||
"""Test building uv command with editable package."""
|
||||
editable_path = Path("/path/to/package")
|
||||
cmd = _build_uv_command("server.py", with_editable=editable_path)
|
||||
expected = [
|
||||
"uv",
|
||||
"run",
|
||||
"--with",
|
||||
"fastmcp",
|
||||
"--with-editable",
|
||||
"/path/to/package",
|
||||
"fastmcp",
|
||||
"run",
|
||||
"server.py",
|
||||
]
|
||||
assert cmd == expected
|
||||
|
||||
def test_build_uv_command_with_packages(self):
|
||||
"""Test building uv command with additional packages."""
|
||||
cmd = _build_uv_command("server.py", with_packages=["pkg1", "pkg2"])
|
||||
expected = [
|
||||
"uv",
|
||||
"run",
|
||||
"--with",
|
||||
"fastmcp",
|
||||
"--with",
|
||||
"pkg1",
|
||||
"--with",
|
||||
"pkg2",
|
||||
"fastmcp",
|
||||
"run",
|
||||
"server.py",
|
||||
]
|
||||
assert cmd == expected
|
||||
|
||||
def test_build_uv_command_no_banner(self):
|
||||
"""Test building uv command with no banner flag."""
|
||||
cmd = _build_uv_command("server.py", no_banner=True)
|
||||
expected = [
|
||||
"uv",
|
||||
"run",
|
||||
"--with",
|
||||
"fastmcp",
|
||||
"fastmcp",
|
||||
"run",
|
||||
"server.py",
|
||||
"--no-banner",
|
||||
]
|
||||
assert cmd == expected
|
||||
|
||||
|
||||
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
|
||||
command, bound, _ = app.parse_args(["version"])
|
||||
command()
|
||||
|
||||
# Verify it printed something and exited with 0
|
||||
mock_print.assert_called_once()
|
||||
mock_exit.assert_called_once_with(0)
|
||||
|
||||
|
||||
class TestDevCommand:
|
||||
"""Test the dev command."""
|
||||
|
||||
def test_dev_command_parsing(self):
|
||||
"""Test that dev command can be parsed with various options."""
|
||||
# Test basic parsing
|
||||
command, bound, _ = app.parse_args(["dev", "server.py"])
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
|
||||
# Test with options
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"dev",
|
||||
"server.py",
|
||||
"--with",
|
||||
"package1",
|
||||
"--inspector-version",
|
||||
"1.0.0",
|
||||
"--ui-port",
|
||||
"3000",
|
||||
]
|
||||
)
|
||||
assert bound.arguments["with_packages"] == ["package1"]
|
||||
assert bound.arguments["inspector_version"] == "1.0.0"
|
||||
assert bound.arguments["ui_port"] == 3000
|
||||
|
||||
|
||||
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."""
|
||||
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,
|
||||
)
|
||||
|
||||
@patch("fastmcp.cli.cli.run_module.run_command")
|
||||
def test_run_command_with_options(self, mock_run_command):
|
||||
"""Test run command with various options."""
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"run",
|
||||
"server.py",
|
||||
"--transport",
|
||||
"http",
|
||||
"--host",
|
||||
"localhost",
|
||||
"--port",
|
||||
"8080",
|
||||
"--log-level",
|
||||
"DEBUG",
|
||||
"--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,
|
||||
)
|
||||
|
||||
@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
|
||||
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
|
||||
# Mock TypeAdapter
|
||||
with patch("fastmcp.cli.cli.TypeAdapter") as mock_adapter:
|
||||
mock_adapter.return_value.dump_json.return_value = b'{"name": "TestServer"}'
|
||||
|
||||
output_file = tmp_path / "test-output.json"
|
||||
|
||||
# Parse and execute
|
||||
command, bound, _ = app.parse_args(
|
||||
[
|
||||
"inspect",
|
||||
"server.py",
|
||||
"--output",
|
||||
str(output_file),
|
||||
]
|
||||
)
|
||||
|
||||
# This is an async command, so we need to run it
|
||||
import asyncio
|
||||
|
||||
asyncio.run(command(**bound.arguments))
|
||||
|
||||
# Verify the output file was created
|
||||
assert output_file.exists()
|
||||
assert output_file.read_text() == '{"name": "TestServer"}'
|
||||
|
||||
@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
|
||||
167
tests/cli/test_install.py
Normal file
167
tests/cli/test_install.py
Normal file
|
|
@ -0,0 +1,167 @@
|
|||
"""Tests for the install subcommands."""
|
||||
|
||||
from fastmcp.cli.install import install_app
|
||||
|
||||
|
||||
class TestInstallApp:
|
||||
"""Test the install subapp."""
|
||||
|
||||
def test_install_app_exists(self):
|
||||
"""Test that the install app is properly configured."""
|
||||
# install_app.name is a tuple in cyclopts
|
||||
assert "install" in install_app.name
|
||||
assert "Install MCP servers" in install_app.help
|
||||
|
||||
def test_install_commands_registered(self):
|
||||
"""Test that all install commands are registered."""
|
||||
# Check that the app has the expected help text and structure
|
||||
# This is a simpler check that doesn't rely on internal methods
|
||||
assert hasattr(install_app, "help")
|
||||
assert "Install MCP servers" in install_app.help
|
||||
|
||||
# We can test that the commands parse without errors
|
||||
try:
|
||||
install_app.parse_args(["claude-code", "--help"])
|
||||
install_app.parse_args(["claude-desktop", "--help"])
|
||||
install_app.parse_args(["cursor", "--help"])
|
||||
install_app.parse_args(["mcp-json", "--help"])
|
||||
except SystemExit:
|
||||
# Help commands exit with 0, that's expected
|
||||
pass
|
||||
|
||||
|
||||
class TestClaudeCodeInstall:
|
||||
"""Test claude-code install command."""
|
||||
|
||||
def test_claude_code_basic(self):
|
||||
"""Test basic claude-code install command parsing."""
|
||||
# Parse command with correct parameter names
|
||||
command, bound, _ = install_app.parse_args(
|
||||
["claude-code", "server.py", "--server-name", "test-server"]
|
||||
)
|
||||
|
||||
# Verify parsing was successful
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
assert bound.arguments["server_name"] == "test-server"
|
||||
|
||||
def test_claude_code_with_options(self):
|
||||
"""Test claude-code install with various options."""
|
||||
command, bound, _ = install_app.parse_args(
|
||||
[
|
||||
"claude-code",
|
||||
"server.py",
|
||||
"--server-name",
|
||||
"test-server",
|
||||
"--with",
|
||||
"package1",
|
||||
"--with",
|
||||
"package2",
|
||||
"--env",
|
||||
"VAR1=value1",
|
||||
]
|
||||
)
|
||||
|
||||
assert bound.arguments["with_packages"] == ["package1", "package2"]
|
||||
assert bound.arguments["env_vars"] == ["VAR1=value1"]
|
||||
|
||||
|
||||
class TestClaudeDesktopInstall:
|
||||
"""Test claude-desktop install command."""
|
||||
|
||||
def test_claude_desktop_basic(self):
|
||||
"""Test basic claude-desktop install command parsing."""
|
||||
command, bound, _ = install_app.parse_args(
|
||||
["claude-desktop", "server.py", "--server-name", "test-server"]
|
||||
)
|
||||
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
assert bound.arguments["server_name"] == "test-server"
|
||||
|
||||
def test_claude_desktop_with_env_vars(self):
|
||||
"""Test claude-desktop install with environment variables."""
|
||||
command, bound, _ = install_app.parse_args(
|
||||
[
|
||||
"claude-desktop",
|
||||
"server.py",
|
||||
"--server-name",
|
||||
"test-server",
|
||||
"--env",
|
||||
"VAR1=value1",
|
||||
"--env",
|
||||
"VAR2=value2",
|
||||
]
|
||||
)
|
||||
|
||||
assert bound.arguments["env_vars"] == ["VAR1=value1", "VAR2=value2"]
|
||||
|
||||
|
||||
class TestCursorInstall:
|
||||
"""Test cursor install command."""
|
||||
|
||||
def test_cursor_basic(self):
|
||||
"""Test basic cursor install command parsing."""
|
||||
command, bound, _ = install_app.parse_args(
|
||||
["cursor", "server.py", "--server-name", "test-server"]
|
||||
)
|
||||
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
assert bound.arguments["server_name"] == "test-server"
|
||||
|
||||
def test_cursor_with_options(self):
|
||||
"""Test cursor install with options."""
|
||||
command, bound, _ = install_app.parse_args(
|
||||
["cursor", "server.py", "--server-name", "test-server"]
|
||||
)
|
||||
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
assert bound.arguments["server_name"] == "test-server"
|
||||
|
||||
|
||||
class TestMcpJsonInstall:
|
||||
"""Test mcp-json install command."""
|
||||
|
||||
def test_mcp_json_basic(self):
|
||||
"""Test basic mcp-json install command parsing."""
|
||||
command, bound, _ = install_app.parse_args(
|
||||
["mcp-json", "server.py", "--server-name", "test-server"]
|
||||
)
|
||||
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
assert bound.arguments["server_name"] == "test-server"
|
||||
|
||||
def test_mcp_json_with_copy(self):
|
||||
"""Test mcp-json install with copy to clipboard option."""
|
||||
command, bound, _ = install_app.parse_args(
|
||||
["mcp-json", "server.py", "--server-name", "test-server", "--copy"]
|
||||
)
|
||||
|
||||
assert bound.arguments["copy"] is True
|
||||
|
||||
|
||||
class TestInstallCommandParsing:
|
||||
"""Test command parsing and error handling."""
|
||||
|
||||
def test_install_minimal_args(self):
|
||||
"""Test install commands with minimal required arguments."""
|
||||
# Each command should work with just a server spec
|
||||
commands_to_test = [
|
||||
["claude-code", "server.py"],
|
||||
["claude-desktop", "server.py"],
|
||||
["cursor", "server.py"],
|
||||
]
|
||||
|
||||
for cmd_args in commands_to_test:
|
||||
command, bound, _ = install_app.parse_args(cmd_args)
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
|
||||
def test_mcp_json_minimal(self):
|
||||
"""Test that mcp-json works with minimal arguments."""
|
||||
# Should work with just server spec
|
||||
command, bound, _ = install_app.parse_args(["mcp-json", "server.py"])
|
||||
assert command is not None
|
||||
assert bound.arguments["server_spec"] == "server.py"
|
||||
361
tests/cli/test_run.py
Normal file
361
tests/cli/test_run.py
Normal file
|
|
@ -0,0 +1,361 @@
|
|||
"""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,
|
||||
)
|
||||
|
||||
|
||||
class TestUrlDetection:
|
||||
"""Test URL detection functionality."""
|
||||
|
||||
def test_is_url_valid_http(self):
|
||||
"""Test detection of valid HTTP URLs."""
|
||||
assert is_url("http://example.com")
|
||||
assert is_url("http://localhost:8080")
|
||||
assert is_url("http://127.0.0.1:3000/path")
|
||||
|
||||
def test_is_url_valid_https(self):
|
||||
"""Test detection of valid HTTPS URLs."""
|
||||
assert is_url("https://example.com")
|
||||
assert is_url("https://api.example.com/mcp")
|
||||
assert is_url("https://localhost:8443")
|
||||
|
||||
def test_is_url_invalid(self):
|
||||
"""Test detection of non-URLs."""
|
||||
assert not is_url("server.py")
|
||||
assert not is_url("/path/to/server.py")
|
||||
assert not is_url("server.py:app")
|
||||
assert not is_url("ftp://example.com") # Not http/https
|
||||
assert not is_url("file:///path/to/file")
|
||||
|
||||
|
||||
class TestFilePathParsing:
|
||||
"""Test file path parsing functionality."""
|
||||
|
||||
def test_parse_file_path_simple(self, tmp_path):
|
||||
"""Test parsing simple file path without object."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("# test server")
|
||||
|
||||
file_path, server_object = parse_file_path(str(test_file))
|
||||
assert file_path == test_file.resolve()
|
||||
assert server_object is None
|
||||
|
||||
def test_parse_file_path_with_object(self, tmp_path):
|
||||
"""Test parsing file path with object specification."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("# test server")
|
||||
|
||||
file_path, server_object = parse_file_path(f"{test_file}:app")
|
||||
assert file_path == test_file.resolve()
|
||||
assert server_object == "app"
|
||||
|
||||
def test_parse_file_path_complex_object(self, tmp_path):
|
||||
"""Test parsing file path with complex object specification."""
|
||||
test_file = tmp_path / "server.py"
|
||||
test_file.write_text("# test server")
|
||||
|
||||
# The current implementation splits on the last colon, so file:module:app
|
||||
# becomes file_path="file:module" and server_object="app"
|
||||
# We need to create a file with a colon in the name for this test
|
||||
complex_file = tmp_path / "server:module.py"
|
||||
complex_file.write_text("# test server")
|
||||
|
||||
file_path, server_object = parse_file_path(f"{complex_file}:app")
|
||||
assert file_path == complex_file.resolve()
|
||||
assert server_object == "app"
|
||||
|
||||
def test_parse_file_path_nonexistent(self):
|
||||
"""Test parsing nonexistent file path exits."""
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
parse_file_path("nonexistent.py")
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
def test_parse_file_path_directory(self, tmp_path):
|
||||
"""Test parsing directory path exits."""
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
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."""
|
||||
|
||||
def test_import_server_with_standard_name(self, tmp_path):
|
||||
"""Test importing server with standard object name."""
|
||||
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")
|
||||
assert exc_info.value.code == 1
|
||||
Loading…
Add table
Add a link
Reference in a new issue