From 6694d9a19add35bfc7d42cccb8bee7977a5c3ee2 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 6 Jul 2025 20:50:31 -0400 Subject: [PATCH] Add tests --- tests/cli/__init__.py | 1 + tests/cli/test_cli.py | 259 +++++++++++++++++++++++++++ tests/cli/test_install.py | 167 ++++++++++++++++++ tests/cli/test_run.py | 361 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 788 insertions(+) create mode 100644 tests/cli/__init__.py create mode 100644 tests/cli/test_cli.py create mode 100644 tests/cli/test_install.py create mode 100644 tests/cli/test_run.py diff --git a/tests/cli/__init__.py b/tests/cli/__init__.py new file mode 100644 index 000000000..6bfb427aa --- /dev/null +++ b/tests/cli/__init__.py @@ -0,0 +1 @@ +"""CLI test package.""" diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py new file mode 100644 index 000000000..cf72ef64e --- /dev/null +++ b/tests/cli/test_cli.py @@ -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 diff --git a/tests/cli/test_install.py b/tests/cli/test_install.py new file mode 100644 index 000000000..ece867c3a --- /dev/null +++ b/tests/cli/test_install.py @@ -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" diff --git a/tests/cli/test_run.py b/tests/cli/test_run.py new file mode 100644 index 000000000..4ca1c2e25 --- /dev/null +++ b/tests/cli/test_run.py @@ -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