prefecthq--fastmcp/tests/cli/test_run.py

import asyncio
import inspect
import json
import subprocess
import sys
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
from urllib.parse import parse_qs, urlsplit

import pytest
from pydantic import ValidationError
from starlette.testclient import TestClient

from fastmcp.cli.apps_dev import _make_dev_app, _MessageLog
from fastmcp.cli.cli import apps, inspector, run
from fastmcp.cli.run import (
    create_mcp_config_server,
    is_url,
    run_module_command,
    run_with_reload,
)
from fastmcp.client.client import Client
from fastmcp.client.transports import FastMCPTransport
from fastmcp.mcp_config import MCPConfig, StdioMCPServer
from fastmcp.server.server import FastMCP
from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import FileSystemSource


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 TestFileSystemSource:
    """Test FileSystemSource path parsing functionality."""

    def test_parse_simple_path(self, tmp_path):
        """Test parsing simple file path without object."""
        test_file = tmp_path / "server.py"
        test_file.write_text("# test server")

        source = FileSystemSource(path=str(test_file))
        assert Path(source.path).resolve() == test_file.resolve()
        assert source.entrypoint is None

    def test_parse_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")

        source = FileSystemSource(path=f"{test_file}:app")
        assert Path(source.path).resolve() == test_file.resolve()
        assert source.entrypoint == "app"

    def test_parse_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 implementation splits on the last colon, so file:module:app
        # becomes file_path="file:module" and entrypoint="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")

        source = FileSystemSource(path=f"{complex_file}:app")
        assert Path(source.path).resolve() == complex_file.resolve()
        assert source.entrypoint == "app"

    async def test_load_server_nonexistent(self):
        """Test loading nonexistent file path exits."""
        source = FileSystemSource(path="nonexistent.py")
        with pytest.raises(SystemExit) as exc_info:
            await source.load_server()
        assert isinstance(exc_info.value, SystemExit)
        assert exc_info.value.code == 1

    async def test_load_server_directory(self, tmp_path):
        """Test loading directory path exits."""
        source = FileSystemSource(path=str(tmp_path))
        with pytest.raises(SystemExit) as exc_info:
            await source.load_server()
        assert isinstance(exc_info.value, SystemExit)
        assert exc_info.value.code == 1


class TestMCPConfig:
    """Test MCPConfig functionality."""

    @pytest.mark.timeout(15)
    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Stdio subprocess lifecycle is unreliable on Windows CI",
    )
    async def test_run_mcp_config(self, tmp_path: Path):
        """Test creating a server from an MCPConfig file."""
        server_script = inspect.cleandoc("""
            from fastmcp import FastMCP

            mcp = FastMCP()

            @mcp.tool
            def add(a: int, b: int) -> int:
                return a + b

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

        script_path: Path = tmp_path / "test.py"
        script_path.write_text(server_script)

        mcp_config_path = tmp_path / "mcp_config.json"

        mcp_config = MCPConfig(
            mcpServers={
                "test_server": StdioMCPServer(command="python", args=[str(script_path)])
            }
        )
        mcp_config.write_to_file(mcp_config_path)

        server: FastMCP[None] = create_mcp_config_server(mcp_config_path)

        client = Client[FastMCPTransport](server)

        async with client:
            tools = await client.list_tools()
            assert len(tools) == 1

    async def test_validate_mcp_config(self, tmp_path: Path):
        """Test creating a server from an MCPConfig file."""

        mcp_config_path = tmp_path / "mcp_config.json"

        mcp_config = {"mcpServers": {"test_server": dict(x=1, y=2)}}
        with mcp_config_path.open("w") as f:
            json.dump(mcp_config, f)

        with pytest.raises(ValidationError, match="validation errors for MCPConfig"):
            create_mcp_config_server(mcp_config_path)


class TestServerImport:
    """Test server import functionality using real files."""

    async def test_import_server_basic_mcp(self, tmp_path):
        """Test importing server with basic FastMCP server."""
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp

mcp = fastmcp.FastMCP("TestServer")

@mcp.tool
def greet(name: str) -> str:
    return f"Hello, {name}!"
""")

        source = FileSystemSource(path=str(test_file))
        server = await source.load_server()
        assert server.name == "TestServer"
        tools = await server.list_tools()
        assert any(t.name == "greet" for t in tools)

    async def test_import_server_with_main_block(self, tmp_path):
        """Test importing server with if __name__ == '__main__' block."""
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp

app = fastmcp.FastMCP("MainServer")

@app.tool
def calculate(x: int, y: int) -> int:
    return x + y

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

        source = FileSystemSource(path=str(test_file))
        server = await source.load_server()
        assert server.name == "MainServer"
        tools = await server.list_tools()
        assert any(t.name == "calculate" for t in tools)

    async def test_import_server_standard_names(self, tmp_path):
        """Test automatic detection of standard names (mcp, server, app)."""
        # Test with 'mcp' name
        mcp_file = tmp_path / "mcp_server.py"
        mcp_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("MCPServer")
""")

        source = FileSystemSource(path=str(mcp_file))
        server = await source.load_server()
        assert server.name == "MCPServer"

        # Test with 'server' name
        server_file = tmp_path / "server_server.py"
        server_file.write_text("""
import fastmcp
server = fastmcp.FastMCP("ServerServer")
""")

        source = FileSystemSource(path=str(server_file))
        server = await source.load_server()
        assert server.name == "ServerServer"

        # Test with 'app' name
        app_file = tmp_path / "app_server.py"
        app_file.write_text("""
import fastmcp
app = fastmcp.FastMCP("AppServer")
""")

        source = FileSystemSource(path=str(app_file))
        server = await source.load_server()
        assert server.name == "AppServer"

    async def test_import_server_nonstandard_name(self, tmp_path):
        """Test importing server with non-standard object name."""
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp

my_custom_server = fastmcp.FastMCP("CustomServer")

@my_custom_server.tool
def custom_tool() -> str:
    return "custom"
""")

        source = FileSystemSource(path=f"{test_file}:my_custom_server")
        server = await source.load_server()
        assert server.name == "CustomServer"
        tools = await server.list_tools()
        assert any(t.name == "custom_tool" for t in tools)

    async def test_import_server_no_standard_names_fails(self, tmp_path):
        """Test importing server when no standard names exist fails."""
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp

other_name = fastmcp.FastMCP("OtherServer")
""")

        source = FileSystemSource(path=str(test_file))
        with pytest.raises(SystemExit) as exc_info:
            await source.load_server()
        assert isinstance(exc_info.value, SystemExit)
        assert exc_info.value.code == 1

    async def test_import_server_nonexistent_object_fails(self, tmp_path):
        """Test importing nonexistent server object fails."""
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp

mcp = fastmcp.FastMCP("TestServer")
""")

        source = FileSystemSource(path=f"{test_file}:nonexistent")
        with pytest.raises(SystemExit) as exc_info:
            await source.load_server()

        assert isinstance(exc_info.value, SystemExit)
        assert exc_info.value.code == 1


class TestV1ServerAsync:
    """Test FastMCP 1.x server async support."""

    async def test_run_v1_server_stdio(self, tmp_path):
        """Test that v1 server uses async stdio method."""
        from unittest.mock import AsyncMock, patch

        from mcp.server.mcpserver import MCPServer as SDKServer

        from fastmcp.cli.run import run_command

        # Create a v1 FastMCP server file with both sync and async tools
        test_file = tmp_path / "v1_server.py"
        test_file.write_text("""
from mcp.server.mcpserver import MCPServer as FastMCP

mcp = FastMCP("V1Server")

@mcp.tool()
def sync_echo(text: str) -> str:
    '''Sync tool for testing'''
    return f"sync: {text}"

@mcp.tool()
async def async_echo(text: str) -> str:
    '''Async tool for testing'''
    return f"async: {text}"
""")

        # Mock the async run method
        with patch.object(
            SDKServer, "run_stdio_async", new_callable=AsyncMock
        ) as run_mock:
            await run_command(str(test_file), transport="stdio")
            run_mock.assert_called_once()

    async def test_run_v1_server_http(self, tmp_path):
        """Test that v1 server uses async http method."""
        from unittest.mock import AsyncMock, patch

        from mcp.server.mcpserver import MCPServer as SDKServer

        from fastmcp.cli.run import run_command

        # Create a v1 FastMCP server file with both sync and async tools
        test_file = tmp_path / "v1_server.py"
        test_file.write_text("""
from mcp.server.mcpserver import MCPServer as FastMCP

mcp = FastMCP("V1Server")

@mcp.tool()
def sync_echo(text: str) -> str:
    '''Sync tool for testing'''
    return f"sync: {text}"

@mcp.tool()
async def async_echo(text: str) -> str:
    '''Async tool for testing'''
    return f"async: {text}"
""")

        # Mock the async run method
        with patch.object(
            SDKServer, "run_streamable_http_async", new_callable=AsyncMock
        ) as run_mock:
            await run_command(str(test_file), transport="http")
            run_mock.assert_called_once()

    async def test_run_v1_server_streamable_http(self, tmp_path):
        """Test that v1 server uses async streamable-http method."""
        from unittest.mock import AsyncMock, patch

        from mcp.server.mcpserver import MCPServer as SDKServer

        from fastmcp.cli.run import run_command

        # Create a v1 FastMCP server file with both sync and async tools
        test_file = tmp_path / "v1_server.py"
        test_file.write_text("""
from mcp.server.mcpserver import MCPServer as FastMCP

mcp = FastMCP("V1Server")

@mcp.tool()
def sync_echo(text: str) -> str:
    '''Sync tool for testing'''
    return f"sync: {text}"

@mcp.tool()
async def async_echo(text: str) -> str:
    '''Async tool for testing'''
    return f"async: {text}"
""")

        # Mock the async run method
        with patch.object(
            SDKServer, "run_streamable_http_async", new_callable=AsyncMock
        ) as run_mock:
            await run_command(str(test_file), transport="streamable-http")
            run_mock.assert_called_once()

    async def test_run_v1_server_sse(self, tmp_path):
        """Test that v1 server uses async sse method."""
        from unittest.mock import AsyncMock, patch

        from mcp.server.mcpserver import MCPServer as SDKServer

        from fastmcp.cli.run import run_command

        # Create a v1 FastMCP server file with both sync and async tools
        test_file = tmp_path / "v1_server.py"
        test_file.write_text("""
from mcp.server.mcpserver import MCPServer as FastMCP

mcp = FastMCP("V1Server")

@mcp.tool()
def sync_echo(text: str) -> str:
    '''Sync tool for testing'''
    return f"sync: {text}"

@mcp.tool()
async def async_echo(text: str) -> str:
    '''Async tool for testing'''
    return f"async: {text}"
""")

        # Mock the async run method
        with patch.object(
            SDKServer, "run_sse_async", new_callable=AsyncMock
        ) as run_mock:
            await run_command(str(test_file), transport="sse")
            run_mock.assert_called_once()

    async def test_run_v1_server_default_transport(self, tmp_path):
        """Test that v1 server uses streamable-http by default."""
        from unittest.mock import AsyncMock, patch

        from mcp.server.mcpserver import MCPServer as SDKServer

        from fastmcp.cli.run import run_command

        # Create a v1 FastMCP server file with both sync and async tools
        test_file = tmp_path / "v1_server.py"
        test_file.write_text("""
from mcp.server.mcpserver import MCPServer as FastMCP

mcp = FastMCP("V1Server")

@mcp.tool()
def sync_echo(text: str) -> str:
    '''Sync tool for testing'''
    return f"sync: {text}"

@mcp.tool()
async def async_echo(text: str) -> str:
    '''Async tool for testing'''
    return f"async: {text}"
""")

        # Mock the async run method
        with patch.object(
            SDKServer, "run_streamable_http_async", new_callable=AsyncMock
        ) as run_mock:
            await run_command(str(test_file))
            run_mock.assert_called_once()

    async def test_run_v1_server_with_host_port(self, tmp_path):
        """Test that v1 server receives host/port settings."""
        from unittest.mock import AsyncMock, patch

        from mcp.server.mcpserver import MCPServer as SDKServer

        from fastmcp.cli.run import run_command

        # Create a v1 FastMCP server file with both sync and async tools
        test_file = tmp_path / "v1_server.py"
        test_file.write_text("""
from mcp.server.mcpserver import MCPServer as FastMCP

mcp = FastMCP("V1Server")

@mcp.tool()
def sync_echo(text: str) -> str:
    '''Sync tool for testing'''
    return f"sync: {text}"

@mcp.tool()
async def async_echo(text: str) -> str:
    '''Async tool for testing'''
    return f"async: {text}"
""")

        # Mock the async run method
        with patch.object(
            SDKServer, "run_streamable_http_async", new_callable=AsyncMock
        ) as run_mock:
            await run_command(
                str(test_file), transport="http", host="0.0.0.0", port=9000
            )
            run_mock.assert_called_once()


class TestSkipSource:
    """Test the --skip-source functionality."""

    async def test_run_command_calls_prepare_by_default(self, tmp_path):
        """Test that run_command calls source.prepare() by default."""
        from unittest.mock import AsyncMock, patch

        from fastmcp.cli.run import run_command

        # Create a test server file
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")

        # Create a test config file
        config_file = tmp_path / "fastmcp.json"
        config_data = {"source": {"path": str(test_file), "entrypoint": "mcp"}}
        config_file.write_text(json.dumps(config_data))

        # Mock the prepare method and server run
        with (
            patch.object(
                FileSystemSource, "prepare", new_callable=AsyncMock
            ) as prepare_mock,
            patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
        ):
            # Run the command
            await run_command(str(config_file))

            # Verify prepare was called
            prepare_mock.assert_called_once()

    async def test_run_command_skips_prepare_with_flag(self, tmp_path):
        """Test that run_command skips source.prepare() when skip_source=True."""
        from unittest.mock import AsyncMock, patch

        from fastmcp.cli.run import run_command

        # Create a test server file
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")

        # Create a test config file
        config_file = tmp_path / "fastmcp.json"
        config_data = {"source": {"path": str(test_file), "entrypoint": "mcp"}}
        config_file.write_text(json.dumps(config_data))

        # Mock the prepare method and server run
        with (
            patch.object(
                FileSystemSource, "prepare", new_callable=AsyncMock
            ) as prepare_mock,
            patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
        ):
            # Run the command with skip_source=True
            await run_command(str(config_file), skip_source=True)

            # Verify prepare was NOT called
            prepare_mock.assert_not_called()

    async def test_filesystem_source_prepare_by_default(self, tmp_path):
        """Test that FileSystemSource is prepared when using direct file spec."""
        from unittest.mock import AsyncMock, patch

        from fastmcp.cli.run import run_command
        from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import (
            FileSystemSource,
        )

        # Create a test server file
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")

        # Mock the prepare method and server run
        with (
            patch.object(
                FileSystemSource, "prepare", new_callable=AsyncMock
            ) as prepare_mock,
            patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
        ):
            # Run with direct file specification
            await run_command(str(test_file))

            # Verify prepare was called
            prepare_mock.assert_called_once()

    async def test_filesystem_source_skip_prepare_with_flag(self, tmp_path):
        """Test that FileSystemSource.prepare() is skipped with skip_source flag."""
        from unittest.mock import AsyncMock, patch

        from fastmcp.cli.run import run_command
        from fastmcp.utilities.mcp_server_config.v1.sources.filesystem import (
            FileSystemSource,
        )

        # Create a test server file
        test_file = tmp_path / "server.py"
        test_file.write_text("""
import fastmcp
mcp = fastmcp.FastMCP("TestServer")
""")

        # Mock the prepare method and server run
        with (
            patch.object(
                FileSystemSource, "prepare", new_callable=AsyncMock
            ) as prepare_mock,
            patch("fastmcp.server.server.FastMCP.run_async", new_callable=AsyncMock),
        ):
            # Run with direct file specification and skip_source=True
            await run_command(str(test_file), skip_source=True)

            # Verify prepare was NOT called
            prepare_mock.assert_not_called()


class TestReloadFunctionality:
    """Test reload functionality."""

    def test_watch_filter_accepts_watched_extensions(self):
        """Test that watch filter accepts common source file extensions."""
        from watchfiles import Change

        from fastmcp.cli.run import _watch_filter

        # Python
        assert _watch_filter(Change.modified, "/path/to/file.py") is True
        assert _watch_filter(Change.added, "server.py") is True
        # JavaScript/TypeScript
        assert _watch_filter(Change.modified, "/path/to/file.js") is True
        assert _watch_filter(Change.modified, "/path/to/file.ts") is True
        assert _watch_filter(Change.modified, "/path/to/file.jsx") is True
        assert _watch_filter(Change.modified, "/path/to/file.tsx") is True
        # Markup/Content
        assert _watch_filter(Change.modified, "/path/to/file.html") is True
        assert _watch_filter(Change.modified, "/path/to/file.md") is True
        assert _watch_filter(Change.modified, "/path/to/file.txt") is True
        # Styles
        assert _watch_filter(Change.modified, "/path/to/file.css") is True
        assert _watch_filter(Change.modified, "/path/to/file.scss") is True
        # Data/Config
        assert _watch_filter(Change.modified, "/path/to/file.json") is True
        assert _watch_filter(Change.modified, "/path/to/file.yaml") is True
        # Images
        assert _watch_filter(Change.modified, "/path/to/file.png") is True
        assert _watch_filter(Change.modified, "/path/to/file.svg") is True

    def test_watch_filter_rejects_unwatched_extensions(self):
        """Test that watch filter rejects files not in the watched set."""
        from watchfiles import Change

        from fastmcp.cli.run import _watch_filter

        assert _watch_filter(Change.modified, "/path/to/file.pyc") is False
        assert _watch_filter(Change.modified, "/path/to/file.pyo") is False
        assert _watch_filter(Change.modified, "Dockerfile") is False
        assert _watch_filter(Change.modified, "/path/to/file.lock") is False
        assert _watch_filter(Change.modified, "/path/to/.gitignore") is False

    def test_all_watched_extensions_are_accepted(self):
        """Test that every extension in WATCHED_EXTENSIONS is accepted."""
        from watchfiles import Change

        from fastmcp.cli.run import WATCHED_EXTENSIONS, _watch_filter

        for ext in WATCHED_EXTENSIONS:
            path = f"/path/to/file{ext}"
            assert _watch_filter(Change.modified, path) is True, (
                f"Expected {ext} to be watched"
            )

    def test_watched_extensions_includes_frontend_types(self):
        """Verify WATCHED_EXTENSIONS contains the expected frontend file types."""
        from fastmcp.cli.run import WATCHED_EXTENSIONS

        # Core frontend extensions that must be present
        expected = {
            # Python
            ".py",
            # JavaScript/TypeScript
            ".js",
            ".ts",
            ".jsx",
            ".tsx",
            # Markup
            ".html",
            ".md",
            ".mdx",
            ".xml",
            # Styles
            ".css",
            ".scss",
            ".sass",
            ".less",
            # Data/Config
            ".json",
            ".yaml",
            ".yml",
            ".toml",
            # Images
            ".png",
            ".jpg",
            ".svg",
            # Media
            ".mp4",
            ".mp3",
        }
        for ext in expected:
            assert ext in WATCHED_EXTENSIONS, f"Expected {ext} in WATCHED_EXTENSIONS"


class TestRunModuleCommand:
    """Test run_module_command functionality."""

    def test_runs_python_m_module(self):
        """Test that run_module_command invokes python -m <module>."""
        mock_result = MagicMock()
        mock_result.returncode = 0

        with (
            patch(
                "fastmcp.cli.run.subprocess.run", return_value=mock_result
            ) as mock_run,
            pytest.raises(SystemExit) as exc_info,
        ):
            run_module_command("my_package")

        assert exc_info.value.code == 0
        call_args = mock_run.call_args
        cmd = call_args[0][0]
        assert "-m" in cmd
        assert "my_package" in cmd

    def test_forwards_extra_args(self):
        """Test that extra arguments are forwarded after the module name."""
        mock_result = MagicMock()
        mock_result.returncode = 0

        with (
            patch(
                "fastmcp.cli.run.subprocess.run", return_value=mock_result
            ) as mock_run,
            pytest.raises(SystemExit),
        ):
            run_module_command("my_package", extra_args=["--host", "0.0.0.0"])

        cmd = mock_run.call_args[0][0]
        assert "--host" in cmd
        assert "0.0.0.0" in cmd

    def test_uses_env_command_builder(self):
        """Test that env_command_builder wraps the command."""
        mock_result = MagicMock()
        mock_result.returncode = 0

        def fake_builder(cmd: list[str]) -> list[str]:
            return ["uv", "run", *cmd]

        with (
            patch(
                "fastmcp.cli.run.subprocess.run", return_value=mock_result
            ) as mock_run,
            pytest.raises(SystemExit),
        ):
            run_module_command("my_package", env_command_builder=fake_builder)

        cmd = mock_run.call_args[0][0]
        assert cmd[0] == "uv"
        assert cmd[1] == "run"
        # Should use bare "python" (not sys.executable) so uv resolves the interpreter
        assert cmd[2] == "python"
        assert "-m" in cmd
        assert "my_package" in cmd

    def test_exits_with_subprocess_error_code(self):
        """Test that non-zero exit codes from the module are propagated."""
        with (
            patch(
                "fastmcp.cli.run.subprocess.run",
                side_effect=subprocess.CalledProcessError(42, ["python", "-m", "bad"]),
            ),
            pytest.raises(SystemExit) as exc_info,
        ):
            run_module_command("bad")

        assert exc_info.value.code == 42

    def test_no_env_builder_runs_plain_python(self):
        """Test that without env_command_builder, plain python is used."""
        mock_result = MagicMock()
        mock_result.returncode = 0

        with (
            patch(
                "fastmcp.cli.run.subprocess.run", return_value=mock_result
            ) as mock_run,
            pytest.raises(SystemExit),
        ):
            run_module_command("my_module", env_command_builder=None)

        cmd = mock_run.call_args[0][0]
        assert cmd[0] == sys.executable
        assert cmd[1] == "-m"
        assert cmd[2] == "my_module"


class TestRunModuleMode:
    """Test the run command's module-mode branch."""

    async def test_run_module_mode_requires_server_spec(self):
        """Test that module mode exits with error when server_spec is None."""
        with pytest.raises(SystemExit) as exc_info:
            await run(None, module=True)

        assert exc_info.value.code == 1

    async def test_run_module_mode_warns_ignored_options(self, caplog):
        """Test that ignored options produce a warning in module mode."""
        mock_result = MagicMock()
        mock_result.returncode = 0

        with (
            patch("fastmcp.cli.run.subprocess.run", return_value=mock_result),
            pytest.raises(SystemExit),
            caplog.at_level("WARNING"),
        ):
            await run(
                "my_module",
                module=True,
                transport="sse",
                host="0.0.0.0",
                port=8080,
            )

        assert any("ignored in module mode" in r.message for r in caplog.records)

    async def test_run_module_mode_delegates_to_run_module_command(self):
        """Test that module mode calls run_module_command with correct args."""
        mock_result = MagicMock()
        mock_result.returncode = 0

        with (
            patch(
                "fastmcp.cli.run.subprocess.run", return_value=mock_result
            ) as mock_subprocess,
            pytest.raises(SystemExit),
        ):
            await run("my_module", module=True)

        cmd = mock_subprocess.call_args[0][0]
        assert "-m" in cmd
        assert "my_module" in cmd

    async def test_run_module_mode_with_reload(self):
        """Test that --reload in module mode delegates to run_with_reload."""
        with patch(
            "fastmcp.cli.run.run_with_reload", new_callable=AsyncMock
        ) as mock_reload:
            await run("my_module", module=True, reload=True, skip_env=True)

        mock_reload.assert_called_once()
        cmd = mock_reload.call_args[0][0]
        assert "fastmcp" in cmd
        assert "--module" in cmd
        assert "--no-reload" in cmd
        assert "my_module" in cmd


class TestRunWithReloadWithServerArgs:
    """Test the run command with reload(run_with_reload) with server args."""

    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        "reload_cmd",
        [
            [
                "fastmcp",
                "run",
                "my_module",
                "--module",
                "--no-reload",
                "--no-banner",
                "--",
                "--debug",
            ],
            [
                "fastmcp",
                "run",
                "my_module",
                "--no-banner",
                "--no-reload",
                "--stateless",
                "--",
                "--debug",
            ],
            ["fastmcp", "run", "my_module", "--module", "--no-reload", "--", "--debug"],
            [
                "fastmcp",
                "run",
                "my_module",
                "--no-reload",
                "--stateless",
                "--",
                "--debug",
            ],
        ],
    )
    async def test_run_with_reload_does_not_mutate_command(self, reload_cmd, caplog):
        """
        Test for issue 4081:
        Verify that run_with_reload does NOT mutate the command arguments.
        The command used for restart must be identical to the original command
        to ensure nothing is incorrectly appended or reordered.
        """
        reload_dirs = [Path(".")]
        shutdown_event = asyncio.Event()
        call_count = 0

        async def mock_create_subprocess(*args, **kwargs):
            nonlocal call_count
            call_count += 1

            proc = AsyncMock()
            proc.kill = MagicMock()
            proc.terminate = MagicMock()
            proc.wait.return_value = 0
            proc.pid = 99999
            proc.returncode = None

            if call_count >= 2:
                actual_args = list(args)
                try:
                    # VALIDATION: reload_cmd remains UNCHANGED
                    assert actual_args == reload_cmd, (
                        f"Regression: Command was mutated during reload.\n"
                        f"Expected: {reload_cmd}\n"
                        f"Actual:   {actual_args}"
                    )
                finally:
                    shutdown_event.set()

            return proc

        mock_watch = MagicMock()
        mock_watch.__aiter__.return_value = iter([[("Change.modified", "server.py")]])

        with (
            patch(
                "fastmcp.cli.run.asyncio.create_subprocess_exec",
                side_effect=mock_create_subprocess,
            ),
            patch("fastmcp.cli.run.awatch", return_value=mock_watch),
            patch("fastmcp.cli.run.asyncio.Event", return_value=shutdown_event),
            caplog.at_level("INFO"),
        ):
            await run_with_reload(reload_cmd, reload_dirs=reload_dirs)

        assert any("Detected changes" in record.message for record in caplog.records)
        assert call_count == 2, f"Restart logic was not triggered for {reload_cmd}"

    async def test_run_with_needs_uv_forwards_stateless_flag(self):
        """`--stateless` should survive the uv-wrapped subprocess path."""
        mock_config = MagicMock()
        mock_config.deployment.transport = None
        mock_config.deployment.host = None
        mock_config.deployment.port = None
        mock_config.deployment.path = None
        mock_config.deployment.log_level = None
        mock_config.deployment.args = ()
        mock_config.environment.build_command = lambda cmd: ["uv", "run", *cmd]

        mock_result = MagicMock()
        mock_result.returncode = 0

        with (
            patch(
                "fastmcp.cli.cli.load_and_merge_config",
                return_value=(mock_config, "server.py"),
            ),
            patch(
                "fastmcp.cli.cli.subprocess.run", return_value=mock_result
            ) as mock_run,
            pytest.raises(SystemExit) as exc_info,
        ):
            await run("server.py", stateless=True)

        assert exc_info.value.code == 0
        cmd = mock_run.call_args.args[0]
        assert "--stateless" in cmd
        assert cmd[cmd.index("--stateless") - 3 : cmd.index("--stateless") + 1] == [
            "fastmcp",
            "run",
            "server.py",
            "--stateless",
        ]


class TestInspectorModuleMode:
    """Test the inspector command's module-mode handling."""

    async def test_inspector_module_mode_skips_load_server(self):
        """Test that inspector with module=True skips load_server() and forwards --module."""
        mock_config = MagicMock()
        mock_config.deployment.port = 8080
        mock_config.environment.build_command = lambda cmd: cmd
        mock_config.source.load_server = AsyncMock()

        mock_process = MagicMock()
        mock_process.returncode = 0

        with (
            patch(
                "fastmcp.cli.cli.load_and_merge_config",
                return_value=(mock_config, "my_module"),
            ),
            patch("fastmcp.cli.cli._get_npx_command", return_value="npx"),
            patch(
                "fastmcp.cli.cli.subprocess.run", return_value=mock_process
            ) as mock_subprocess,
            pytest.raises(SystemExit),
        ):
            await inspector("my_module", module=True)

        # load_server should NOT have been called in module mode
        mock_config.source.load_server.assert_not_called()

        # --module should be in the subprocess command
        cmd = mock_subprocess.call_args[0][0]
        assert "--module" in cmd


class TestRunDevApps:
    """Test running dev apps with the run command."""

    def test_launch_escapes_tool_name_in_html_contexts(self):
        """Test /launch escapes the tool query parameter in HTML sinks."""
        starlette_app = _make_dev_app(
            mcp_url="http://127.0.0.1:8000/mcp",
            app_bridge_js="// js",
            import_map_tag="",
            message_log=_MessageLog(),
            log_panel=False,
        )
        client = TestClient(starlette_app, raise_server_exceptions=False)

        payload = "</title><script>alert(1)</script><img src=x onerror=alert(2)>"
        response = client.get("/launch", params={"tool": payload, "args": "{}"})

        assert response.status_code == 200
        assert payload not in response.text
        assert (
            "&lt;/title&gt;&lt;script&gt;alert(1)&lt;/script&gt;"
            "&lt;img src=x onerror=alert(2)&gt;"
        ) in response.text
        assert "\\u003c/script\\u003e" in response.text

    def test_launch_serializes_args_safely_inside_script(self):
        """Test /launch escapes argument values embedded in the script element."""
        starlette_app = _make_dev_app(
            mcp_url="http://127.0.0.1:8000/mcp",
            app_bridge_js="// js",
            import_map_tag="",
            message_log=_MessageLog(),
            log_panel=False,
        )
        client = TestClient(starlette_app, raise_server_exceptions=False)

        payload = {"name": "</script><script>alert(1)</script>&"}
        response = client.get(
            "/launch",
            params={"tool": "safe_tool", "args": json.dumps(payload)},
        )

        assert response.status_code == 200
        assert json.dumps(payload) not in response.text
        assert (
            '"\\u003c/script\\u003e\\u003cscript\\u003ealert(1)'
            '\\u003c/script\\u003e\\u0026"'
        ) in response.text

    def test_api_launch_encodes_generated_launch_url(self):
        """Test /api/launch encodes query parameters in the returned URL."""
        starlette_app = _make_dev_app(
            mcp_url="http://127.0.0.1:8000/mcp",
            app_bridge_js="// js",
            import_map_tag="",
            message_log=_MessageLog(),
            log_panel=False,
        )
        client = TestClient(starlette_app, raise_server_exceptions=False)

        response = client.post(
            "/api/launch",
            json={
                "tool": "tool&name=<script>",
                "__json_args__": '{"value": "</script>"}',
            },
        )

        assert response.status_code == 200
        url = response.json()
        query = parse_qs(urlsplit(url).query)
        assert query["tool"] == ["tool&name=<script>"]
        assert json.loads(query["args"][0]) == {"value": "</script>"}

    @pytest.mark.parametrize(
        "host, expected_host",
        [
            ("0.0.0.0", "0.0.0.0"),
            ("127.0.0.1", "127.0.0.1"),
        ],
    )
    async def test_run_dev_apps_with_host(self, host, expected_host):
        """Test run command can run a dev app with below new options for issue 4121.
        - host option to support binding specific address other than localhost.
        """
        mock_proc = MagicMock()
        mock_proc.returncode = 0

        mock_server = AsyncMock()
        mock_server.install_signal_handlers = MagicMock()
        mock_server.serve = AsyncMock()

        mock_uvicorn = MagicMock()
        mock_uvicorn.Config = MagicMock()
        mock_uvicorn.Server.return_value = mock_server

        mock_make_dev_app = MagicMock(return_value=MagicMock())
        mock_webbrowser_open = MagicMock()

        with (
            patch(
                "fastmcp.cli.apps_dev._start_user_server",
                new_callable=AsyncMock,
                return_value=mock_proc,
            ),
            patch(
                "fastmcp.cli.apps_dev._fetch_app_bridge_bundle",
                new_callable=AsyncMock,
                return_value=("js_content", "{}"),
            ),
            patch(
                "fastmcp.cli.apps_dev._wait_for_server",
                new_callable=AsyncMock,
                return_value=True,
            ),
            patch("fastmcp.cli.apps_dev._make_dev_app", mock_make_dev_app),
            patch("fastmcp.cli.apps_dev.uvicorn", mock_uvicorn),
            patch("fastmcp.cli.apps_dev.webbrowser.open", mock_webbrowser_open),
            patch("fastmcp.cli.apps_dev.asyncio.sleep", new_callable=AsyncMock),
            patch("socket.socket"),
        ):
            await apps("server.py", host=host)

        make_dev_app_first_arg = mock_make_dev_app.call_args[0][0]
        assert expected_host in make_dev_app_first_arg

        webbrowser_open_first_arg = mock_webbrowser_open.call_args[0][0]
        assert expected_host in webbrowser_open_first_arg

    @pytest.mark.parametrize(
        "log_panel, expected_log_panel",
        [
            (True, True),
            (False, False),
        ],
    )
    async def test_run_dev_apps_log_panel_propagation(
        self, log_panel, expected_log_panel
    ):
        """Test run command can run a dev app with below new options for issue 4121.
        - toggle log-panel option to hide log/debug message in normal cases.

        The test divided into two parts:
        - first, verify if log_panel is correctly propagated to _make_dev_app.
        - second, verify if _make_dev_app calls _inject_log_panel only when log_panel=True

        This test function implements the first part, and the second part is implemented in test_make_dev_app_log_panel_controls_inject
        """

        mock_proc = MagicMock()
        mock_proc.returncode = 0

        mock_server = AsyncMock()
        mock_server.install_signal_handlers = MagicMock()
        mock_server.serve = AsyncMock()

        mock_uvicorn = MagicMock()
        mock_uvicorn.Config = MagicMock()
        mock_uvicorn.Server.return_value = mock_server

        mock_make_dev_app = MagicMock(return_value=MagicMock())
        mock_webbrowser_open = MagicMock()

        with (
            patch(
                "fastmcp.cli.apps_dev._start_user_server",
                new_callable=AsyncMock,
                return_value=mock_proc,
            ),
            patch(
                "fastmcp.cli.apps_dev._fetch_app_bridge_bundle",
                new_callable=AsyncMock,
                return_value=("js_content", "{}"),
            ),
            patch(
                "fastmcp.cli.apps_dev._wait_for_server",
                new_callable=AsyncMock,
                return_value=True,
            ),
            patch("fastmcp.cli.apps_dev._make_dev_app", mock_make_dev_app),
            patch("fastmcp.cli.apps_dev.uvicorn", mock_uvicorn),
            patch("fastmcp.cli.apps_dev.webbrowser.open", mock_webbrowser_open),
            patch("fastmcp.cli.apps_dev.asyncio.sleep", new_callable=AsyncMock),
            patch("socket.socket"),
        ):
            await apps("server.py", log_panel=log_panel)

        # log_panel is the 5th positional argument to _make_dev_app
        actual_log_panel = mock_make_dev_app.call_args[0][4]
        assert actual_log_panel is expected_log_panel

    @pytest.mark.parametrize("log_panel", [True, False])
    def test_make_dev_app_log_panel_controls_inject(self, log_panel):
        """Test if _make_dev_app calls _inject_log_panel only when log_panel=True, regarding issue 4121:
        - toggle log-panel option to hide log/debug message in normal cases.

        The test divided into two parts:
        - first, verify if log_panel is correctly propagated to _make_dev_app.
        - second, verify if _make_dev_app calls _inject_log_panel only when log_panel=True

        This test function implements the second part, and the first part is implemented in test_run_dev_apps_log_panel_propagation
        """

        mock_message_log = _MessageLog()

        with patch(
            "fastmcp.cli.apps_dev._inject_log_panel",
            return_value="<html>injected</html>",
        ) as mock_inject:
            starlette_app = _make_dev_app(
                mcp_url="http://127.0.0.1:8000/mcp",
                app_bridge_js="// js",
                import_map_tag="",
                message_log=mock_message_log,
                log_panel=log_panel,
            )
            client = TestClient(starlette_app, raise_server_exceptions=False)
            client.get("/")

        if log_panel:
            mock_inject.assert_called_once()
        else:
            mock_inject.assert_not_called()