tests/test_mcp_server.py

from __future__ import annotations

from fastapi.testclient import TestClient

from app.container import build_container
from app.mcp.registry import (
    AUDIO_TOOLS,
    IMAGE_TOOLS,
    PROBE_TOOLS,
    SYSTEM_TOOLS,
    TEMPLATE_TOOLS,
    VIDEO_TOOLS,
    WHISPER_TOOLS,
    YTDLP_TOOLS,
    MCPRegistry,
    MediaInput,
)
from app.mcp.server import create_mcp_server
from main import create_app


async def test_mcp_registers_all_tools_resources_and_prompts(settings) -> None:
    server = create_mcp_server(build_container(settings))

    tools = {tool.name for tool in await server.list_tools()}
    expected_tools = set(
        VIDEO_TOOLS
        + AUDIO_TOOLS
        + IMAGE_TOOLS
        + WHISPER_TOOLS
        + YTDLP_TOOLS
        + PROBE_TOOLS
        + SYSTEM_TOOLS
        + TEMPLATE_TOOLS
    )
    expected_tools.update(
        {
            "social.list_providers",
            "social.get_capabilities",
            "social.list_media_assets",
            "social.register_media_asset",
            "social.list_accounts",
            "social.get_account",
            "social.create_post",
            "social.publish_post",
            "social.schedule_post",
            "social.cancel_post",
            "social.get_post",
            "social.get_job",
            "social.get_analytics",
            "ai.capabilities",
            "ai.generate",
            "ai.list_jobs",
            "ai.get_job",
            "ai.cancel_job",
        }
    )
    assert tools == expected_tools

    resources = {str(resource.uri) for resource in await server.list_resources()}
    assert resources == {
        "media://operations",
        "media://formats",
        "media://codecs",
        "media://health",
        "media://configuration",
        "media://version",
    }

    prompts = {prompt.name for prompt in await server.list_prompts()}
    assert prompts == {
        "compress_for_social_media",
        "youtube_to_mp3",
        "download_and_transcribe",
        "generate_subtitles",
        "extract_audio",
        "make_thumbnail",
        "probe_media",
        "instagram_reel",
        "tiktok_video",
        "podcast_audio",
    }


async def test_mcp_errors_use_safe_structured_envelope(settings) -> None:
    registry = MCPRegistry(build_container(settings))

    response = await registry.run_probe(
        "probe_media",
        MediaInput(temp_path=str(settings.temp_dir / "missing.mp4")),
    )

    assert response["success"] is False
    assert response["request_id"]
    assert response["processing_time"] >= 0
    assert response["error"]["code"] == "INVALID_INPUT"
    assert "traceback" not in str(response).lower()


def test_rest_and_mcp_coexist_in_one_application(settings) -> None:
    application = create_app(settings)
    mounts = {getattr(route, "path", None) for route in application.routes}
    assert "/mcp" in mounts

    with TestClient(application) as client:
        response = client.get("/health")

    assert response.status_code == 200
    assert response.json()["success"] is True
    assert application.state.mcp_server is not None