from collections.abc import Callable

import pytest

import mcp.types as types
from mcp.client._memory import InMemoryTransport
from mcp.client.session import ClientSession
from mcp.server import Server
from mcp.server.fastmcp import FastMCP
from mcp.types import ListToolsRequest, ListToolsResult

from .conftest import StreamSpyCollection

pytestmark = pytest.mark.anyio


@pytest.fixture
async def full_featured_server():
    """Create a server with tools, resources, prompts, and templates."""
    server = FastMCP("test")

    # pragma: no cover on handlers below - these exist only to register items with the
    # server so list_* methods return results. The handlers themselves are never called
    # because these tests only verify pagination/cursor behavior, not tool/resource invocation.
    @server.tool()
    def greet(name: str) -> str:  # pragma: no cover
        """Greet someone by name."""
        return f"Hello, {name}!"

    @server.resource("test://resource")
    def test_resource() -> str:  # pragma: no cover
        """A test resource."""
        return "Test content"

    @server.resource("test://template/{id}")
    def test_template(id: str) -> str:  # pragma: no cover
        """A test resource template."""
        return f"Template content for {id}"

    @server.prompt()
    def greeting_prompt(name: str) -> str:  # pragma: no cover
        """A greeting prompt."""
        return f"Please greet {name}."

    return server


@pytest.mark.parametrize(
    "method_name,request_method",
    [
        ("list_tools", "tools/list"),
        ("list_resources", "resources/list"),
        ("list_prompts", "prompts/list"),
        ("list_resource_templates", "resources/templates/list"),
    ],
)
async def test_list_methods_params_parameter(
    stream_spy: Callable[[], StreamSpyCollection],
    full_featured_server: FastMCP,
    method_name: str,
    request_method: str,
):
    """Test that the params parameter is accepted and correctly passed to the server.

    Covers: list_tools, list_resources, list_prompts, list_resource_templates

    See: https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/pagination#request-format
    """
    transport = InMemoryTransport(full_featured_server)
    async with transport.connect() as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            spies = stream_spy()

            # Test without params (omitted)
            method = getattr(session, method_name)
            _ = await method()
            requests = spies.get_client_requests(method=request_method)
            assert len(requests) == 1
            assert requests[0].params is None

            spies.clear()

            # Test with params containing cursor
            _ = await method(params=types.PaginatedRequestParams(cursor="from_params"))
            requests = spies.get_client_requests(method=request_method)
            assert len(requests) == 1
            assert requests[0].params is not None
            assert requests[0].params["cursor"] == "from_params"

            spies.clear()

            # Test with empty params
            _ = await method(params=types.PaginatedRequestParams())
            requests = spies.get_client_requests(method=request_method)
            assert len(requests) == 1
            # Empty params means no cursor
            assert requests[0].params is None or "cursor" not in requests[0].params


async def test_list_tools_with_strict_server_validation(
    full_featured_server: FastMCP,
):
    """Test pagination with a server that validates request format strictly."""
    transport = InMemoryTransport(full_featured_server)
    async with transport.connect() as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            result = await session.list_tools(params=types.PaginatedRequestParams())
            assert isinstance(result, ListToolsResult)
            assert len(result.tools) > 0


async def test_list_tools_with_lowlevel_server():
    """Test that list_tools works with a lowlevel Server using params."""
    server = Server("test-lowlevel")

    @server.list_tools()
    async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult:
        # Echo back what cursor we received in the tool description
        cursor = request.params.cursor if request.params else None
        return ListToolsResult(
            tools=[
                types.Tool(
                    name="test_tool",
                    description=f"cursor={cursor}",
                    input_schema={},
                )
            ]
        )

    transport = InMemoryTransport(server)
    async with transport.connect() as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            result = await session.list_tools(params=types.PaginatedRequestParams())
            assert result.tools[0].description == "cursor=None"

            result = await session.list_tools(params=types.PaginatedRequestParams(cursor="page2"))
            assert result.tools[0].description == "cursor=page2"