fastmcp/tests/server/test_logging.py

import asyncio
import logging
from contextlib import nullcontext
from functools import partial
from unittest.mock import AsyncMock, MagicMock, Mock, patch

import anyio
import pytest

from fastmcp.server.server import FastMCP


class CustomLogFormatterForTest(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        return f"TEST_FORMAT::{record.levelname}::{record.name}::{record.getMessage()}"


@pytest.fixture
def mcp_server() -> FastMCP:
    return FastMCP(name="TestLogServer")


async def _wait_for_server_loop(mock_server_instance: Mock) -> None:
    while mock_server_instance._serve.await_count == 0:
        await asyncio.sleep(0)


async def _wait_for_event(event: anyio.Event, **_kwargs: object) -> None:
    await event.wait()


@patch("fastmcp.server.mixins.transport.uvicorn.Server")
@patch("fastmcp.server.mixins.transport.uvicorn.Config")
async def test_uvicorn_logging_default_level(
    mock_uvicorn_config_constructor: Mock,
    mock_uvicorn_server_constructor: Mock,
    mcp_server: FastMCP,
):
    """Tests that FastMCP passes log_level to uvicorn.Config if no log_config is given."""
    mock_server_instance = MagicMock()
    mock_server_instance._serve = AsyncMock()
    mock_server_instance.shutdown = AsyncMock()
    mock_server_instance.capture_signals.return_value = nullcontext()
    mock_uvicorn_server_constructor.return_value = mock_server_instance
    serve_finished_event = anyio.Event()
    mock_server_instance._serve.side_effect = partial(
        _wait_for_event, serve_finished_event
    )

    test_log_level = "warning"

    server_task = asyncio.create_task(
        mcp_server.run_http_async(log_level=test_log_level, port=8003)
    )
    await _wait_for_server_loop(mock_server_instance)

    mock_uvicorn_config_constructor.assert_called_once()
    _, kwargs_config = mock_uvicorn_config_constructor.call_args

    assert kwargs_config.get("log_level") == test_log_level.lower()
    assert "log_config" not in kwargs_config

    mock_uvicorn_server_constructor.assert_called_once_with(
        mock_uvicorn_config_constructor.return_value
    )
    mock_server_instance._serve.assert_awaited_once_with(sockets=None)

    # Signal the mock to finish and wait for run_http_async to return.
    serve_finished_event.set()
    await asyncio.wait_for(server_task, timeout=2.0)


@patch("fastmcp.server.mixins.transport.uvicorn.Server")
@patch("fastmcp.server.mixins.transport.uvicorn.Config")
async def test_uvicorn_logging_with_custom_log_config(
    mock_uvicorn_config_constructor: Mock,
    mock_uvicorn_server_constructor: Mock,
    mcp_server: FastMCP,
):
    """Tests that FastMCP passes log_config to uvicorn.Config and not log_level."""
    mock_server_instance = MagicMock()
    mock_server_instance._serve = AsyncMock()
    mock_server_instance.shutdown = AsyncMock()
    mock_server_instance.capture_signals.return_value = nullcontext()
    mock_uvicorn_server_constructor.return_value = mock_server_instance
    serve_finished_event = anyio.Event()
    mock_server_instance._serve.side_effect = partial(
        _wait_for_event, serve_finished_event
    )

    sample_log_config = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "test_formatter": {
                "()": "tests.server.test_logging.CustomLogFormatterForTest"
            }
        },
        "handlers": {
            "test_handler": {
                "formatter": "test_formatter",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
            }
        },
        "loggers": {
            "uvicorn.error": {
                "handlers": ["test_handler"],
                "level": "INFO",
                "propagate": False,
            }
        },
    }

    server_task = asyncio.create_task(
        mcp_server.run_http_async(
            uvicorn_config={"log_config": sample_log_config}, port=8004
        )
    )
    await _wait_for_server_loop(mock_server_instance)

    mock_uvicorn_config_constructor.assert_called_once()
    _, kwargs_config = mock_uvicorn_config_constructor.call_args

    assert kwargs_config.get("log_config") == sample_log_config
    assert "log_level" not in kwargs_config

    mock_uvicorn_server_constructor.assert_called_once_with(
        mock_uvicorn_config_constructor.return_value
    )
    mock_server_instance._serve.assert_awaited_once_with(sockets=None)

    # Signal the mock to finish and wait for run_http_async to return.
    serve_finished_event.set()
    await asyncio.wait_for(server_task, timeout=2.0)


@patch("fastmcp.server.mixins.transport.uvicorn.Server")
@patch("fastmcp.server.mixins.transport.uvicorn.Config")
async def test_uvicorn_logging_custom_log_config_overrides_log_level_param(
    mock_uvicorn_config_constructor: Mock,
    mock_uvicorn_server_constructor: Mock,
    mcp_server: FastMCP,
):
    """Tests log_config precedence if log_level is also passed to run_http_async."""
    mock_server_instance = MagicMock()
    mock_server_instance._serve = AsyncMock()
    mock_server_instance.shutdown = AsyncMock()
    mock_server_instance.capture_signals.return_value = nullcontext()
    mock_uvicorn_server_constructor.return_value = mock_server_instance
    serve_finished_event = anyio.Event()
    mock_server_instance._serve.side_effect = partial(
        _wait_for_event, serve_finished_event
    )

    sample_log_config = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "test_formatter": {
                "()": "tests.server.test_logging.CustomLogFormatterForTest"
            }
        },
        "handlers": {
            "test_handler": {
                "formatter": "test_formatter",
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
            }
        },
        "loggers": {
            "uvicorn.error": {
                "handlers": ["test_handler"],
                "level": "INFO",
                "propagate": False,
            }
        },
    }
    explicit_log_level = "debug"

    server_task = asyncio.create_task(
        mcp_server.run_http_async(
            log_level=explicit_log_level,
            uvicorn_config={"log_config": sample_log_config},
            port=8005,
        )
    )
    await _wait_for_server_loop(mock_server_instance)

    mock_uvicorn_config_constructor.assert_called_once()
    _, kwargs_config = mock_uvicorn_config_constructor.call_args

    assert kwargs_config.get("log_config") == sample_log_config
    assert "log_level" not in kwargs_config

    mock_uvicorn_server_constructor.assert_called_once_with(
        mock_uvicorn_config_constructor.return_value
    )
    mock_server_instance._serve.assert_awaited_once_with(sockets=None)

    # Signal the mock to finish and wait for run_http_async to return.
    serve_finished_event.set()
    await asyncio.wait_for(server_task, timeout=2.0)