prompts/tests/web/test_mcp_skills.py

from __future__ import annotations

import json

import pytest

pytestmark = pytest.mark.smoke


REQUIRED_TOOL_NAMES = (
    "search_patterns",
    "get_pattern_by_id",
    "get_skill_document_by_id",
    "search_prompts",
    "get_prompt_by_id",
)


REQUIRED_RESOURCE_URIS = (
    "resource://catalog/skills_index",
    "resource://catalog/prompts_index",
)

TOOL_NAME_PARAMETERS = tuple(
    pytest.param(
        tool_name,
        id=tool_name.replace("_", "-"),
    )
    for tool_name in REQUIRED_TOOL_NAMES
)


SEARCH_QUERY_PARAMETERS = (
    pytest.param(
        "pytest",
        {"pytesting"},
        id="query-pytest",
    ),
    pytest.param(
        "asyncio",
        {"pytesting", "async-fastapi-sqlmodel"},
        id="query-asyncio",
    ),
    pytest.param(
        "fastapi testing",
        {"pytesting"},
        id="query-fastapi-testing",
    ),
    pytest.param(
        "asyncio fastapi testing deterministic pytest",
        {"pytesting"},
        id="query-composite-async-fastapi-testing-deterministic-pytest",
    ),
)


RESOURCE_URI_PARAMETERS = tuple(
    pytest.param(
        resource_uri,
        id=resource_uri.removeprefix("resource://").replace("/", "-"),
    )
    for resource_uri in REQUIRED_RESOURCE_URIS
)


class TestMcpCatalogSurface:
    """Covers smoke-level MCP catalog discovery and tool execution paths."""

    class TestTools:
        """Covers MCP tool-list and tool-call smoke behavior."""

        @pytest.mark.parametrize("tool_name", TOOL_NAME_PARAMETERS)
        @pytest.mark.asyncio
        async def test_lists_core_catalog_tools(
            self,
            mcp_session_factory,
            tool_name: str,
        ) -> None:
            """Ensures tools/list exposes each required core catalog tool name."""
            async with mcp_session_factory() as mcp_session:
                result = await mcp_session.list_tools()
            tool_names = {tool.name for tool in result.tools}

            assert tool_name in tool_names

        @pytest.mark.asyncio
        async def test_calls_search_patterns_tool(self, mcp_session_factory) -> None:
            """Ensures tools/call succeeds for search_patterns with basic args."""
            async with mcp_session_factory() as mcp_session:
                result = await mcp_session.call_tool(
                    "search_patterns",
                    {
                        "query": "pytest",
                        "limit": 5,
                    },
                )

            assert result.isError is False
            assert result.content

        @pytest.mark.parametrize(
            ("query", "expected_skill_ids"),
            SEARCH_QUERY_PARAMETERS,
        )
        @pytest.mark.asyncio
        async def test_search_patterns_matches_expected_skills_for_query_terms(
            self,
            mcp_session_factory,
            query: str,
            expected_skill_ids: set[str],
        ) -> None:
            """Ensures query terms return expected skill IDs from search_patterns."""
            async with mcp_session_factory() as mcp_session:
                result = await mcp_session.call_tool(
                    "search_patterns",
                    {
                        "query": query,
                        "limit": 20,
                    },
                )

            assert result.isError is False
            assert result.content

            payload = json.loads(result.content[0].text)
            found_skill_ids = {pattern["id"] for pattern in payload["patterns"]}

            assert expected_skill_ids.issubset(found_skill_ids)

    class TestResources:
        """Covers MCP resource and resource-template discovery."""

        @pytest.mark.parametrize("resource_uri", RESOURCE_URI_PARAMETERS)
        @pytest.mark.asyncio
        async def test_lists_catalog_resources(
            self,
            mcp_session_factory,
            resource_uri: str,
        ) -> None:
            """Ensures resources/list exposes each required catalog resource URI."""
            async with mcp_session_factory() as mcp_session:
                result = await mcp_session.list_resources()
            resource_uris = {str(resource.uri) for resource in result.resources}

            assert resource_uri in resource_uris

        @pytest.mark.asyncio
        async def test_lists_resource_templates(self, mcp_session_factory) -> None:
            """Ensures resources/templates/list includes skills and prompt templates."""
            async with mcp_session_factory() as mcp_session:
                result = await mcp_session.list_resource_templates()
            template_uris = {template.uriTemplate for template in result.resourceTemplates}

            assert "resource://skills/{skill_id}/document" in template_uris
            assert "resource://prompts/{prompt_id}/document" in template_uris

        @pytest.mark.asyncio
        async def test_reads_mcp_details_skill_document(self, mcp_session_factory) -> None:
            """Ensures read_resource resolves the mcp-details skill document URI."""
            async with mcp_session_factory() as mcp_session:
                result = await mcp_session.call_tool(
                    "read_resource",
                    {"uri": "resource://skills/mcp-details/document"},
                )

            assert result.isError is False
            assert result.content

    class TestPrompts:
        """Covers MCP prompt discovery surface."""

        @pytest.mark.asyncio
        async def test_lists_registered_prompts(self, mcp_session_factory) -> None:
            """Ensures prompts/list returns at least one registered prompt."""
            async with mcp_session_factory() as mcp_session:
                result = await mcp_session.list_prompts()

            assert result.prompts