Adding tests for CLI

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"
+            )