fastmcp/tests/client/test_logs.py

import logging

import pytest
from mcp import LoggingLevel

from fastmcp import Client, Context, FastMCP
from fastmcp.client.logging import LogMessage


class LogHandler:
    def __init__(self):
        self.logs: list[LogMessage] = []
        self.logger = logging.getLogger(__name__)
        # Backwards-compatible way to get the log level mapping
        if hasattr(logging, "getLevelNamesMapping"):
            # For Python 3.11+
            self.LOGGING_LEVEL_MAP = logging.getLevelNamesMapping()  # pyright: ignore [reportAttributeAccessIssue]
        else:
            # For older Python versions
            self.LOGGING_LEVEL_MAP = logging._nameToLevel

    async def handle_log(self, message: LogMessage) -> None:
        self.logs.append(message)

        level = self.LOGGING_LEVEL_MAP[message.level.upper()]
        msg = message.data.get("msg")
        extra = message.data.get("extra")
        self.logger.log(level, msg, extra=extra)


@pytest.fixture
def fastmcp_server():
    mcp = FastMCP()

    @mcp.tool
    async def log(context: Context) -> None:
        await context.info(message="hello?")

    @mcp.tool
    async def echo_log(
        message: str,
        context: Context,
        level: LoggingLevel | None = None,
        logger: str | None = None,
    ) -> None:
        await context.log(message=message, level=level)

    return mcp


class TestClientLogs:
    async def test_log(self, fastmcp_server: FastMCP, caplog):
        caplog.set_level(logging.INFO, logger=__name__)

        log_handler = LogHandler()
        async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
            await client.call_tool("log", {})

        assert len(log_handler.logs) == 1
        assert log_handler.logs[0].data["msg"] == "hello?"
        assert log_handler.logs[0].level == "info"

        assert len(caplog.records) == 1
        assert caplog.records[0].msg == "hello?"
        assert caplog.records[0].levelname == "INFO"

    async def test_echo_log(self, fastmcp_server: FastMCP, caplog):
        caplog.set_level(logging.INFO, logger=__name__)

        log_handler = LogHandler()
        async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
            await client.call_tool("echo_log", {"message": "this is a log"})

            assert len(log_handler.logs) == 1
            assert len(caplog.records) == 1
            await client.call_tool(
                "echo_log", {"message": "this is a warning log", "level": "warning"}
            )
            assert len(log_handler.logs) == 2
            assert len(caplog.records) == 2

        assert log_handler.logs[0].data["msg"] == "this is a log"
        assert log_handler.logs[0].level == "info"
        assert log_handler.logs[1].data["msg"] == "this is a warning log"
        assert log_handler.logs[1].level == "warning"

        assert caplog.records[0].msg == "this is a log"
        assert caplog.records[0].levelname == "INFO"
        assert caplog.records[1].msg == "this is a warning log"
        assert caplog.records[1].levelname == "WARNING"


class TestSetLoggingLevel:
    async def test_set_logging_level(self, fastmcp_server: FastMCP):
        """Client can set the minimum log level and lower-level messages are suppressed."""
        log_handler = LogHandler()
        async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
            await client.set_logging_level("warning")
            await client.call_tool(
                "echo_log", {"message": "debug msg", "level": "debug"}
            )
            await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
            await client.call_tool(
                "echo_log", {"message": "warning msg", "level": "warning"}
            )
            await client.call_tool(
                "echo_log", {"message": "error msg", "level": "error"}
            )

        assert len(log_handler.logs) == 2
        assert log_handler.logs[0].data["msg"] == "warning msg"
        assert log_handler.logs[1].data["msg"] == "error msg"

    async def test_set_logging_level_debug_allows_all(self, fastmcp_server: FastMCP):
        """Setting level to debug allows all messages through."""
        log_handler = LogHandler()
        async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
            await client.set_logging_level("debug")
            await client.call_tool(
                "echo_log", {"message": "debug msg", "level": "debug"}
            )
            await client.call_tool("echo_log", {"message": "info msg", "level": "info"})

        assert len(log_handler.logs) == 2

    async def test_default_level_allows_all(self, fastmcp_server: FastMCP):
        """Without calling set_logging_level, all messages are sent."""
        log_handler = LogHandler()
        async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
            await client.call_tool(
                "echo_log", {"message": "debug msg", "level": "debug"}
            )
            await client.call_tool("echo_log", {"message": "info msg", "level": "info"})

        assert len(log_handler.logs) == 2

    async def test_server_default_client_log_level(self):
        """Server-wide client_log_level filters messages for all sessions."""
        mcp = FastMCP(client_log_level="error")

        @mcp.tool
        async def echo_log(
            message: str, context: Context, level: LoggingLevel | None = None
        ) -> None:
            await context.log(message=message, level=level)

        log_handler = LogHandler()
        async with Client(mcp, log_handler=log_handler.handle_log) as client:
            await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
            await client.call_tool(
                "echo_log", {"message": "warning msg", "level": "warning"}
            )
            await client.call_tool(
                "echo_log", {"message": "error msg", "level": "error"}
            )

        assert len(log_handler.logs) == 1
        assert log_handler.logs[0].data["msg"] == "error msg"

    async def test_session_level_overrides_server_default(self):
        """Per-session setLevel overrides the server's client_log_level."""
        mcp = FastMCP(client_log_level="error")

        @mcp.tool
        async def echo_log(
            message: str, context: Context, level: LoggingLevel | None = None
        ) -> None:
            await context.log(message=message, level=level)

        log_handler = LogHandler()
        async with Client(mcp, log_handler=log_handler.handle_log) as client:
            await client.set_logging_level("warning")
            await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
            await client.call_tool(
                "echo_log", {"message": "warning msg", "level": "warning"}
            )
            await client.call_tool(
                "echo_log", {"message": "error msg", "level": "error"}
            )

        assert len(log_handler.logs) == 2
        assert log_handler.logs[0].data["msg"] == "warning msg"
        assert log_handler.logs[1].data["msg"] == "error msg"


class TestDefaultLogHandler:
    """Tests for default_log_handler with data as any JSON-serializable type."""

    async def test_default_handler_routes_to_correct_levels(self):
        """Test that default_log_handler routes server logs to appropriate Python log levels."""
        from unittest.mock import MagicMock, patch

        from mcp.types import LoggingMessageNotificationParams

        from fastmcp.client.logging import default_log_handler

        with patch("fastmcp.client.logging.from_server_logger") as mock_logger:
            # Set up mock methods
            mock_logger.debug = MagicMock()
            mock_logger.info = MagicMock()
            mock_logger.warning = MagicMock()
            mock_logger.error = MagicMock()
            mock_logger.critical = MagicMock()

            # Test each log level
            test_cases = [
                ("debug", mock_logger.debug, "Debug message"),
                ("info", mock_logger.info, "Info message"),
                ("notice", mock_logger.info, "Notice message"),  # notice -> info
                ("warning", mock_logger.warning, "Warning message"),
                ("error", mock_logger.error, "Error message"),
                ("critical", mock_logger.critical, "Critical message"),
                ("alert", mock_logger.critical, "Alert message"),  # alert -> critical
                (
                    "emergency",
                    mock_logger.critical,
                    "Emergency message",
                ),  # emergency -> critical
            ]

            for level, expected_method, msg in test_cases:
                # Reset mocks
                mock_logger.reset_mock()

                # Create log message with data as a string
                log_msg = LoggingMessageNotificationParams(
                    level=level,  # type: ignore[arg-type]  # ty:ignore[invalid-argument-type]
                    logger="test.logger",
                    data=msg,
                )

                # Call handler
                await default_log_handler(log_msg)

                # Verify correct method was called
                expected_method.assert_called_once_with(
                    msg=f"Received {level.upper()} from server (test.logger): {msg}"
                )

    async def test_default_handler_without_logger_name(self):
        """Test that default_log_handler works when logger name is None."""
        from unittest.mock import MagicMock, patch

        from mcp.types import LoggingMessageNotificationParams

        from fastmcp.client.logging import default_log_handler

        with patch("fastmcp.client.logging.from_server_logger") as mock_logger:
            mock_logger.info = MagicMock()

            log_msg = LoggingMessageNotificationParams(
                level="info",
                logger=None,
                data="Message without logger",
            )

            await default_log_handler(log_msg)

            mock_logger.info.assert_called_once_with(
                msg="Received INFO from server: Message without logger"
            )

    async def test_default_handler_with_dict_data(self):
        """Test that default_log_handler handles dict data correctly."""
        from unittest.mock import MagicMock, patch

        from mcp.types import LoggingMessageNotificationParams

        from fastmcp.client.logging import default_log_handler

        with patch("fastmcp.client.logging.from_server_logger") as mock_logger:
            mock_logger.info = MagicMock()

            log_msg = LoggingMessageNotificationParams(
                level="info",
                logger="test.logger",
                data={"key": "value", "count": 42},
            )

            await default_log_handler(log_msg)

            # Should log the entire dict as a string
            mock_logger.info.assert_called_once()
            call_args = mock_logger.info.call_args
            assert "Received INFO from server (test.logger):" in call_args[1]["msg"]
            assert "key" in call_args[1]["msg"]
            assert "value" in call_args[1]["msg"]

    async def test_default_handler_with_list_data(self):
        """Test that default_log_handler handles list data correctly."""
        from unittest.mock import MagicMock, patch

        from mcp.types import LoggingMessageNotificationParams

        from fastmcp.client.logging import default_log_handler

        with patch("fastmcp.client.logging.from_server_logger") as mock_logger:
            mock_logger.warning = MagicMock()

            log_msg = LoggingMessageNotificationParams(
                level="warning",
                logger="test.logger",
                data=["item1", "item2", "item3"],
            )

            await default_log_handler(log_msg)

            # Should log the entire list as a string
            mock_logger.warning.assert_called_once()
            call_args = mock_logger.warning.call_args
            assert "Received WARNING from server (test.logger):" in call_args[1]["msg"]
            assert "item1" in call_args[1]["msg"]

    async def test_default_handler_with_number_data(self):
        """Test that default_log_handler handles numeric data correctly."""
        from unittest.mock import MagicMock, patch

        from mcp.types import LoggingMessageNotificationParams

        from fastmcp.client.logging import default_log_handler

        with patch("fastmcp.client.logging.from_server_logger") as mock_logger:
            mock_logger.error = MagicMock()

            log_msg = LoggingMessageNotificationParams(
                level="error",
                logger=None,
                data=404,
            )

            await default_log_handler(log_msg)

            mock_logger.error.assert_called_once_with(
                msg="Received ERROR from server: 404"
            )