Spaces:
Running
Running
File size: 3,108 Bytes
7cc81cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | 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
|