diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py new file mode 100644 index 000000000..3ce4d5af1 --- /dev/null +++ b/tests/cli/test_cli.py @@ -0,0 +1,422 @@ +"""Tests for the CLI module.""" + + +import subprocess +from pathlib import Path +from unittest.mock import MagicMock, Mock, patch + +import pytest +from typer.testing import CliRunner + +from fastmcp.cli import cli + + +# Set up test runner +runner = CliRunner() + + +@pytest.fixture +def mock_console(): + """Mock the rich console to test output.""" + with patch("fastmcp.cli.cli.console") as mock_console: + yield mock_console + + +@pytest.fixture +def mock_logger(): + """Mock the logger to test logging.""" + with patch("fastmcp.cli.cli.logger") as mock_logger: + yield mock_logger + + +@pytest.fixture +def mock_exit(): + """Mock sys.exit to prevent tests from exiting.""" + with patch("sys.exit") as mock_exit: + yield mock_exit + + +@pytest.fixture +def temp_python_file(tmp_path): + """Create a temporary Python file with a test server.""" + server_code = """ +from mcp import Server + +class TestServer(Server): + name = "test_server" + dependencies = ["package1", "package2"] + + def run(self, **kwargs): + print("Running server with", kwargs) + +mcp = TestServer() +server = TestServer() +app = TestServer() +custom_server = TestServer() +""" + file_path = tmp_path / "test_server.py" + file_path.write_text(server_code) + return file_path + + +@pytest.fixture +def temp_env_file(tmp_path): + """Create a temporary .env file.""" + env_content = """ +TEST_VAR1=value1 +TEST_VAR2=value2 +""" + env_path = tmp_path / ".env" + env_path.write_text(env_content) + return env_path + + +class TestHelperFunctions: + """Tests for helper functions in cli.py.""" + + def test_get_npx_command_unix(self): + """Test getting npx command on unix systems.""" + with patch("sys.platform", "linux"): + with patch("subprocess.run") as mock_run: + mock_run.return_value = Mock(returncode=0) + assert cli._get_npx_command() == "npx" + + def test_get_npx_command_windows(self): + """Test getting npx command on Windows.""" + with patch("sys.platform", "win32"): + with patch("subprocess.run") as mock_run: + # First try fails, second succeeds + mock_run.side_effect = [ + subprocess.CalledProcessError(1, "npx.cmd"), + Mock(returncode=0), + ] + assert cli._get_npx_command() == "npx.exe" + + def test_get_npx_command_not_found(self): + """Test when npx command is not found.""" + with patch("sys.platform", "win32"): + with patch("subprocess.run") as mock_run: + mock_run.side_effect = [ + subprocess.CalledProcessError(1, "npx.cmd"), + subprocess.CalledProcessError(1, "npx.exe"), + subprocess.CalledProcessError(1, "npx"), + ] + assert cli._get_npx_command() is None + + def test_parse_env_var_valid(self): + """Test parsing valid environment variables.""" + assert cli._parse_env_var("KEY=VALUE") == ("KEY", "VALUE") + assert cli._parse_env_var("KEY=") == ("KEY", "") + assert cli._parse_env_var("KEY=VALUE=WITH=EQUALS") == ("KEY", "VALUE=WITH=EQUALS") + assert cli._parse_env_var(" KEY = VALUE ") == ("KEY", "VALUE") + + def test_build_uv_command_basic(self): + """Test building basic uv command.""" + cmd = cli._build_uv_command("file.py") + assert cmd == ["uv", "run", "--with", "fastmcp", "fastmcp", "run", "file.py"] + + def test_build_uv_command_with_editable(self): + """Test building uv command with editable flag.""" + cmd = cli._build_uv_command("file.py", with_editable=Path("/path/to/project")) + assert cmd == [ + "uv", "run", "--with", "fastmcp", + "--with-editable", "/path/to/project", + "fastmcp", "run", "file.py" + ] + + def test_build_uv_command_with_packages(self): + """Test building uv command with additional packages.""" + cmd = cli._build_uv_command("file.py", with_packages=["pkg1", "pkg2"]) + assert cmd == [ + "uv", "run", "--with", "fastmcp", + "--with", "pkg1", "--with", "pkg2", + "fastmcp", "run", "file.py" + ] + + def test_build_uv_command_full(self): + """Test building full uv command with all options.""" + cmd = cli._build_uv_command( + "file.py:server", + with_editable=Path("/path/to/project"), + with_packages=["pkg1", "pkg2"], + ) + assert cmd == [ + "uv", "run", "--with", "fastmcp", + "--with-editable", "/path/to/project", + "--with", "pkg1", "--with", "pkg2", + "fastmcp", "run", "file.py:server" + ] + + def test_parse_file_path_simple(self): + """Test parsing simple file path.""" + with patch("pathlib.Path.exists") as mock_exists, \ + patch("pathlib.Path.is_file") as mock_is_file, \ + patch("pathlib.Path.expanduser") as mock_expanduser, \ + patch("pathlib.Path.resolve") as mock_resolve: + + mock_exists.return_value = True + mock_is_file.return_value = True + mock_expanduser.return_value = Path("file.py") + mock_resolve.return_value = Path("file.py") + + path, obj = cli._parse_file_path("file.py") + assert path == Path("file.py") + assert obj is None + + def test_parse_file_path_with_object(self): + """Test parsing file path with object.""" + with patch("pathlib.Path.exists") as mock_exists, \ + patch("pathlib.Path.is_file") as mock_is_file, \ + patch("pathlib.Path.expanduser") as mock_expanduser, \ + patch("pathlib.Path.resolve") as mock_resolve: + + mock_exists.return_value = True + mock_is_file.return_value = True + mock_expanduser.return_value = Path("file.py") + mock_resolve.return_value = Path("file.py") + + path, obj = cli._parse_file_path("file.py:server") + assert path == Path("file.py") + assert obj == "server" + + def test_parse_file_path_windows(self): + """Test parsing Windows file path.""" + with patch("pathlib.Path.exists") as mock_exists, \ + patch("pathlib.Path.is_file") as mock_is_file, \ + patch("pathlib.Path.expanduser") as mock_expanduser, \ + patch("pathlib.Path.resolve") as mock_resolve: + + mock_exists.return_value = True + mock_is_file.return_value = True + mock_expanduser.return_value = Path("C:/path/file.py") + mock_resolve.return_value = Path("C:/path/file.py") + + path, obj = cli._parse_file_path("C:/path/file.py:server") + assert path == Path("C:/path/file.py") + assert obj == "server" + + def test_parse_file_path_not_file(self, mock_exit, mock_logger): + """Test parsing path that is not a file.""" + with patch("pathlib.Path.exists") as mock_exists, \ + patch("pathlib.Path.is_file") as mock_is_file, \ + patch("pathlib.Path.expanduser") as mock_expanduser, \ + patch("pathlib.Path.resolve") as mock_resolve: + + mock_exists.return_value = True + mock_is_file.return_value = False + mock_expanduser.return_value = Path("directory") + mock_resolve.return_value = Path("directory") + + cli._parse_file_path("directory") + mock_logger.error.assert_called_once() + mock_exit.assert_called_once_with(1) + + +class TestVersionCommand: + """Tests for the version command.""" + + def test_version_early_exit_with_resilient_parsing(self): + """Test version command exits early with resilient parsing.""" + ctx = MagicMock() + ctx.resilient_parsing = True + result = cli.version(ctx) + assert result is None + + +class TestDevCommand: + """Tests for the dev command.""" + + def test_dev_command_success(self, temp_python_file, mock_logger): + """Test successful dev command execution.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import, \ + patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx, \ + patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv, \ + patch("subprocess.run") as mock_run: + + mock_parse.return_value = (temp_python_file, None) + mock_server = MagicMock() + mock_server.dependencies = ["extra_dep"] + mock_import.return_value = mock_server + mock_get_npx.return_value = "npx" + mock_build_uv.return_value = ["uv", "command"] + mock_run.return_value = MagicMock(returncode=0) + + result = runner.invoke(cli.app, ["dev", str(temp_python_file)]) + assert result.exit_code == 0 + mock_run.assert_called_once() + + # Check dependencies were passed correctly + mock_build_uv.assert_called_once_with( + str(temp_python_file), + None, + ["extra_dep"] + ) + + def test_dev_command_with_ui_port(self, temp_python_file): + """Test dev command with UI port.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import, \ + patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx, \ + patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv, \ + patch("subprocess.run") as mock_run: + + mock_parse.return_value = (temp_python_file, None) + mock_import.return_value = MagicMock(dependencies=[]) + mock_get_npx.return_value = "npx" + mock_build_uv.return_value = ["uv", "command"] + mock_run.return_value = MagicMock(returncode=0) + + result = runner.invoke(cli.app, ["dev", str(temp_python_file), "--ui-port", "3000"]) + assert result.exit_code == 0 + + # Check environment variables were set + env = mock_run.call_args[1]["env"] + assert "CLIENT_PORT" in env + assert env["CLIENT_PORT"] == "3000" + + def test_dev_command_with_server_port(self, temp_python_file): + """Test dev command with server port.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import, \ + patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx, \ + patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv, \ + patch("subprocess.run") as mock_run: + + mock_parse.return_value = (temp_python_file, None) + mock_import.return_value = MagicMock(dependencies=[]) + mock_get_npx.return_value = "npx" + mock_build_uv.return_value = ["uv", "command"] + mock_run.return_value = MagicMock(returncode=0) + + result = runner.invoke(cli.app, ["dev", str(temp_python_file), "--server-port", "8080"]) + assert result.exit_code == 0 + + # Check environment variables were set + env = mock_run.call_args[1]["env"] + assert "SERVER_PORT" in env + assert env["SERVER_PORT"] == "8080" + + def test_dev_command_inspector_version(self, temp_python_file): + """Test dev command with specific inspector version.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import, \ + patch("fastmcp.cli.cli._get_npx_command") as mock_get_npx, \ + patch("fastmcp.cli.cli._build_uv_command") as mock_build_uv, \ + patch("subprocess.run") as mock_run: + + mock_parse.return_value = (temp_python_file, None) + mock_import.return_value = MagicMock(dependencies=[]) + mock_get_npx.return_value = "npx" + mock_build_uv.return_value = ["uv", "command"] + mock_run.return_value = MagicMock(returncode=0) + + result = runner.invoke(cli.app, [ + "dev", str(temp_python_file), + "--inspector-version", "1.0.0" + ]) + assert result.exit_code == 0 + + # Check inspector version was used + inspector_cmd = mock_run.call_args[0][0][1] + assert inspector_cmd == "@modelcontextprotocol/inspector@1.0.0" + + +class TestRunCommand: + """Tests for the run command.""" + + def test_run_command_success(self, temp_python_file, mock_logger): + """Test successful run command execution.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import: + + mock_parse.return_value = (temp_python_file, None) + mock_server = MagicMock() + mock_server.name = "test_server" + mock_import.return_value = mock_server + + result = runner.invoke(cli.app, ["run", str(temp_python_file)]) + assert result.exit_code == 0 + mock_server.run.assert_called_once_with() + mock_logger.info.assert_called_with(f'Found server "test_server" in {temp_python_file}') + + def test_run_command_with_transport(self, temp_python_file): + """Test run command with transport option.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import: + + mock_parse.return_value = (temp_python_file, None) + mock_server = MagicMock() + mock_server.name = "test_server" + mock_import.return_value = mock_server + + result = runner.invoke(cli.app, ["run", str(temp_python_file), "--transport", "sse"]) + assert result.exit_code == 0 + mock_server.run.assert_called_once_with(transport="sse") + + def test_run_command_with_host(self, temp_python_file): + """Test run command with host option.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import: + + mock_parse.return_value = (temp_python_file, None) + mock_server = MagicMock() + mock_server.name = "test_server" + mock_import.return_value = mock_server + + result = runner.invoke(cli.app, ["run", str(temp_python_file), "--host", "0.0.0.0"]) + assert result.exit_code == 0 + mock_server.run.assert_called_once_with(host="0.0.0.0") + + def test_run_command_with_port(self, temp_python_file): + """Test run command with port option.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import: + + mock_parse.return_value = (temp_python_file, None) + mock_server = MagicMock() + mock_server.name = "test_server" + mock_import.return_value = mock_server + + result = runner.invoke(cli.app, ["run", str(temp_python_file), "--port", "8080"]) + assert result.exit_code == 0 + mock_server.run.assert_called_once_with(port=8080) + + def test_run_command_with_log_level(self, temp_python_file): + """Test run command with log level option.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import: + + mock_parse.return_value = (temp_python_file, None) + mock_server = MagicMock() + mock_server.name = "test_server" + mock_import.return_value = mock_server + + result = runner.invoke(cli.app, ["run", str(temp_python_file), "--log-level", "DEBUG"]) + assert result.exit_code == 0 + mock_server.run.assert_called_once_with(log_level="DEBUG") + + def test_run_command_with_multiple_options(self, temp_python_file): + """Test run command with multiple options.""" + with patch("fastmcp.cli.cli._parse_file_path") as mock_parse, \ + patch("fastmcp.cli.cli._import_server") as mock_import: + + mock_parse.return_value = (temp_python_file, None) + mock_server = MagicMock() + mock_server.name = "test_server" + mock_import.return_value = mock_server + + result = runner.invoke(cli.app, [ + "run", str(temp_python_file), + "--transport", "sse", + "--host", "0.0.0.0", + "--port", "8080", + "--log-level", "DEBUG" + ]) + assert result.exit_code == 0 + mock_server.run.assert_called_once_with( + transport="sse", + host="0.0.0.0", + port=8080, + log_level="DEBUG" + )