tests/test_cli.py

"""Tests for the FastMCP CLI."""

import json
import sys
from pathlib import Path
from unittest.mock import call, patch

import pytest
from typer.testing import CliRunner

from fastmcp.cli.cli import _parse_env_var, _parse_file_path, app


@pytest.fixture
def mock_config(tmp_path):
    """Create a mock Claude config file."""
    config = {"mcpServers": {}}
    config_file = tmp_path / "claude_desktop_config.json"
    config_file.write_text(json.dumps(config))
    return config_file


@pytest.fixture
def server_file(tmp_path):
    """Create a server file."""
    server_file = tmp_path / "server.py"
    server_file.write_text(
        """from fastmcp import FastMCP
mcp = FastMCP("test")
"""
    )
    return server_file


@pytest.fixture
def mock_env_file(tmp_path):
    """Create a mock .env file."""
    env_file = tmp_path / ".env"
    env_file.write_text("FOO=bar\nBAZ=123")
    return env_file


def test_parse_env_var():
    """Test parsing environment variables."""
    assert _parse_env_var("FOO=bar") == ("FOO", "bar")
    assert _parse_env_var("FOO=") == ("FOO", "")
    assert _parse_env_var("FOO=bar baz") == ("FOO", "bar baz")
    assert _parse_env_var("FOO = bar ") == ("FOO", "bar")

    with pytest.raises(SystemExit):
        _parse_env_var("invalid")


@pytest.mark.parametrize(
    "args,expected_env",
    [
        # Basic env var
        (
            ["--env-var", "FOO=bar"],
            {"FOO": "bar"},
        ),
        # Multiple env vars
        (
            ["--env-var", "FOO=bar", "--env-var", "BAZ=123"],
            {"FOO": "bar", "BAZ": "123"},
        ),
        # Env var with spaces
        (
            ["--env-var", "FOO=bar baz"],
            {"FOO": "bar baz"},
        ),
    ],
)
def test_install_with_env_vars(mock_config, server_file, args, expected_env):
    """Test installing with environment variables."""
    runner = CliRunner()

    with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
        mock_config_path.return_value = mock_config.parent

        result = runner.invoke(
            app,
            ["install", str(server_file)] + args,
        )

        assert result.exit_code == 0

        # Read the config file and check env vars
        config = json.loads(mock_config.read_text())
        assert "mcpServers" in config
        assert len(config["mcpServers"]) == 1
        server = next(iter(config["mcpServers"].values()))
        assert server["env"] == expected_env


def test_parse_file_path_windows_drive():
    """Test parsing a Windows file path with a drive letter."""
    file_spec = r"C:\path\to\file.txt"
    with (
        patch("pathlib.Path.exists", return_value=True),
        patch("pathlib.Path.is_file", return_value=True),
    ):
        file_path, server_object = _parse_file_path(file_spec)
        assert file_path == Path(r"C:\path\to\file.txt").resolve()
        assert server_object is None


def test_parse_file_path_with_object():
    """Test parsing a file path with an object specification."""
    file_spec = "/path/to/file.txt:object"
    with patch("sys.exit") as mock_exit:
        _parse_file_path(file_spec)

        # Check that sys.exit was called twice with code 1
        assert mock_exit.call_count == 2
        mock_exit.assert_has_calls([call(1), call(1)])


def test_parse_file_path_windows_with_object():
    """Test parsing a Windows file path with an object specification."""
    file_spec = r"C:\path\to\file.txt:object"
    with (
        patch("pathlib.Path.exists", return_value=True),
        patch("pathlib.Path.is_file", return_value=True),
    ):
        file_path, server_object = _parse_file_path(file_spec)
        assert file_path == Path(r"C:\path\to\file.txt").resolve()
        assert server_object == "object"


def test_install_with_env_file(mock_config, server_file, mock_env_file):
    """Test installing with environment variables from a file."""
    runner = CliRunner()

    with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
        mock_config_path.return_value = mock_config.parent

        result = runner.invoke(
            app,
            ["install", str(server_file), "--env-file", str(mock_env_file)],
        )

        assert result.exit_code == 0

        # Read the config file and check env vars
        config = json.loads(mock_config.read_text())
        assert "mcpServers" in config
        assert len(config["mcpServers"]) == 1
        server = next(iter(config["mcpServers"].values()))
        assert server["env"] == {"FOO": "bar", "BAZ": "123"}


def test_install_preserves_existing_env_vars(mock_config, server_file):
    """Test that installing preserves existing environment variables."""
    # Set up initial config with env vars
    config = {
        "mcpServers": {
            "test": {
                "command": "uv",
                "args": [
                    "run",
                    "--with",
                    "fastmcp",
                    "fastmcp",
                    "run",
                    str(server_file),
                ],
                "env": {"FOO": "bar", "BAZ": "123"},
            }
        }
    }
    mock_config.write_text(json.dumps(config))

    runner = CliRunner()

    with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
        mock_config_path.return_value = mock_config.parent

        # Install with a new env var
        result = runner.invoke(
            app,
            ["install", str(server_file), "--env-var", "NEW=value"],
        )

        assert result.exit_code == 0

        # Read the config file and check env vars are preserved
        config = json.loads(mock_config.read_text())
        server = next(iter(config["mcpServers"].values()))
        assert server["env"] == {"FOO": "bar", "BAZ": "123", "NEW": "value"}


def test_install_updates_existing_env_vars(mock_config, server_file):
    """Test that installing updates existing environment variables."""
    # Set up initial config with env vars
    config = {
        "mcpServers": {
            "test": {
                "command": "uv",
                "args": [
                    "run",
                    "--with",
                    "fastmcp",
                    "fastmcp",
                    "run",
                    str(server_file),
                ],
                "env": {"FOO": "bar", "BAZ": "123"},
            }
        }
    }
    mock_config.write_text(json.dumps(config))

    runner = CliRunner()

    with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
        mock_config_path.return_value = mock_config.parent

        # Update an existing env var
        result = runner.invoke(
            app,
            ["install", str(server_file), "--env-var", "FOO=newvalue"],
        )

        assert result.exit_code == 0

        # Read the config file and check env var was updated
        config = json.loads(mock_config.read_text())
        server = next(iter(config["mcpServers"].values()))
        assert server["env"] == {"FOO": "newvalue", "BAZ": "123"}


def test_server_dependencies(mock_config, server_file):
    """Test that server dependencies are correctly handled."""
    # Create a server file with dependencies
    server_file = server_file.parent / "server_with_deps.py"
    server_file.write_text(
        """from fastmcp import FastMCP
mcp = FastMCP("test", dependencies=["pandas", "numpy"])
"""
    )

    runner = CliRunner()

    with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
        mock_config_path.return_value = mock_config.parent

        result = runner.invoke(app, ["install", str(server_file)])

        assert result.exit_code == 0

        # Read the config file and check dependencies were added as --with args
        config = json.loads(mock_config.read_text())
        server = next(iter(config["mcpServers"].values()))
        assert "--with" in server["args"]
        assert "pandas" in server["args"]
        assert "numpy" in server["args"]


def test_server_dependencies_empty(mock_config, server_file):
    """Test that server with no dependencies works correctly."""
    runner = CliRunner()

    with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
        mock_config_path.return_value = mock_config.parent

        result = runner.invoke(app, ["install", str(server_file)])

        assert result.exit_code == 0

        # Read the config file and check only fastmcp is in --with args
        config = json.loads(mock_config.read_text())
        server = next(iter(config["mcpServers"].values()))
        assert server["args"].count("--with") == 1
        assert "fastmcp" in server["args"]


def test_dev_with_dependencies(mock_config, server_file):
    """Test that dev command handles dependencies correctly."""
    server_file = server_file.parent / "server_with_deps.py"
    server_file.write_text(
        """from fastmcp import FastMCP
mcp = FastMCP("test", dependencies=["pandas", "numpy"])
"""
    )

    runner = CliRunner()

    with patch("subprocess.run") as mock_run:
        mock_run.return_value.returncode = 0
        result = runner.invoke(app, ["dev", str(server_file)])
        assert result.exit_code == 0

        if sys.platform == "win32":
            # On Windows, expect two calls
            assert mock_run.call_count == 2
            assert mock_run.call_args_list[0] == call(
                ["npx.cmd", "--version"], check=True, capture_output=True, shell=True
            )

            # get the actual command and expected command without dependencies
            actual_cmd = mock_run.call_args_list[1][0][0]
            expected_start = [
                "npx.cmd",
                "@modelcontextprotocol/inspector",
                "uv",
                "run",
                "--with",
                "fastmcp",
            ]
            expected_end = ["fastmcp", "run", str(server_file)]

            # verify start and end of command
            assert actual_cmd[: len(expected_start)] == expected_start
            assert actual_cmd[-len(expected_end) :] == expected_end

            # verify dependencies are present (order-independent)
            deps_section = actual_cmd[len(expected_start) : -len(expected_end)]
            assert all(
                x in deps_section for x in ["--with", "numpy", "--with", "pandas"]
            )

            # Verify subprocess call kwargs, allowing for environment variables
            call_kwargs = mock_run.call_args_list[1][1]
            assert call_kwargs["check"] is True
            assert call_kwargs["shell"] is True
            assert isinstance(call_kwargs["env"], dict)
        else:
            # same verification for unix, just with different command prefix
            actual_cmd = mock_run.call_args_list[0][0][0]
            expected_start = [
                "npx",
                "@modelcontextprotocol/inspector",
                "uv",
                "run",
                "--with",
                "fastmcp",
            ]
            expected_end = ["fastmcp", "run", str(server_file)]

            assert actual_cmd[: len(expected_start)] == expected_start
            assert actual_cmd[-len(expected_end) :] == expected_end

            deps_section = actual_cmd[len(expected_start) : -len(expected_end)]
            assert all(
                x in deps_section for x in ["--with", "numpy", "--with", "pandas"]
            )

            # Verify subprocess call kwargs, allowing for environment variables
            call_kwargs = mock_run.call_args_list[0][1]
            assert call_kwargs["check"] is True
            assert call_kwargs["shell"] is False
            assert isinstance(call_kwargs["env"], dict)


def test_run_with_dependencies(mock_config, server_file):
    """Test that run command does not handle dependencies."""
    # Create a server file with dependencies
    server_file = server_file.parent / "server_with_deps.py"
    server_file.write_text(
        """from fastmcp import FastMCP
mcp = FastMCP("test", dependencies=["pandas", "numpy"])

if __name__ == "__main__":
    mcp.run()
"""
    )

    runner = CliRunner()

    with patch("subprocess.run") as mock_run:
        result = runner.invoke(app, ["run", str(server_file)])
        assert result.exit_code == 0

        # Run command should not call subprocess.run
        mock_run.assert_not_called()