forked from innovacion/searchbox
Anibal Angulo
cda334a45d
This commit renames the package from vector-search-mcp to searchbox. The package imports and executable name are updated accordingly.
130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
"""Tests for the MCP server implementation."""
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastembed import TextEmbedding
|
|
from fastmcp import Client
|
|
from mcp.types import TextContent
|
|
|
|
|
|
class TestMCPServer:
|
|
"""Test the MCP server implementation."""
|
|
|
|
def test_server_import(self):
|
|
"""Test that MCP server can be imported successfully."""
|
|
from searchbox.mcp_server import server
|
|
|
|
assert hasattr(server, 'mcp')
|
|
assert hasattr(server, 'engine')
|
|
|
|
def test_server_initialization(self):
|
|
"""Test that the MCP server initializes correctly."""
|
|
from searchbox.mcp_server import server
|
|
from searchbox.engine import Backend
|
|
|
|
# Verify server module attributes exist
|
|
assert hasattr(server, 'mcp')
|
|
assert hasattr(server, 'engine')
|
|
|
|
# The engine should be created during module import
|
|
# We can't easily test the exact call without complex mocking
|
|
# but we can verify the engine exists and is properly typed
|
|
assert server.engine is not None
|
|
|
|
def test_run_function_exists(self):
|
|
"""Test that the run function exists in the package init."""
|
|
from searchbox.mcp_server import run
|
|
|
|
assert callable(run)
|
|
|
|
def test_run_function_signature(self):
|
|
"""Test that run function has correct signature and docstring."""
|
|
from searchbox.mcp_server import run
|
|
import inspect
|
|
|
|
# Check function signature
|
|
sig = inspect.signature(run)
|
|
params = list(sig.parameters.values())
|
|
|
|
assert len(params) == 1
|
|
assert params[0].name == "transport"
|
|
assert params[0].default == "sse"
|
|
|
|
# Check docstring
|
|
assert run.__doc__ is not None
|
|
assert "transport" in run.__doc__.lower()
|
|
|
|
def test_run_function_type_annotations(self):
|
|
"""Test that run function has proper type annotations."""
|
|
from searchbox.mcp_server import run
|
|
|
|
# Verify function exists and is callable
|
|
assert callable(run)
|
|
|
|
# The function should accept Transport type
|
|
import inspect
|
|
sig = inspect.signature(run)
|
|
assert "transport" in sig.parameters
|
|
|
|
|
|
class TestMCPIntegration:
|
|
"""Integration tests for the MCP server."""
|
|
|
|
async def test_call_tool(self, embedding_model: TextEmbedding, run_mcp: str):
|
|
"""Test calling the semantic search tool via MCP."""
|
|
input = "Quien es el mas guapo?"
|
|
collection = "dummy_collection"
|
|
|
|
embedding: list[float] = list(embedding_model.embed(input))[0].tolist()
|
|
|
|
client = Client(run_mcp)
|
|
|
|
async with client:
|
|
name = "semantic_search"
|
|
body = {"embedding": embedding, "collection": collection}
|
|
result = await client.call_tool(name, body)
|
|
|
|
content_block = result.content[0]
|
|
|
|
assert isinstance(content_block, TextContent)
|
|
|
|
deserialized_result = json.loads(content_block.text)
|
|
|
|
top_result = deserialized_result[0]
|
|
|
|
assert top_result["chunk_id"] == "0"
|
|
assert top_result["score"] > 0.7
|
|
assert top_result["payload"] == {"text": "Rick es el mas guapo"}
|
|
|
|
def test_semantic_search_tool_registration(self):
|
|
"""Test that semantic_search tool registration is accessible."""
|
|
from searchbox.mcp_server.server import mcp
|
|
|
|
# Just verify the mcp object exists and is properly configured
|
|
# The actual tool registration happens during import
|
|
assert mcp is not None
|
|
assert hasattr(mcp, 'tool') # Has the decorator method
|
|
|
|
def test_server_module_attributes(self):
|
|
"""Test that server module has expected attributes."""
|
|
from searchbox.mcp_server import server
|
|
|
|
assert hasattr(server, 'mcp')
|
|
assert hasattr(server, 'engine')
|
|
|
|
# Verify mcp is a FastMCP instance
|
|
from fastmcp import FastMCP
|
|
assert isinstance(server.mcp, FastMCP)
|
|
|
|
def test_package_init_exports(self):
|
|
"""Test that package __init__ exports the run function."""
|
|
from searchbox.mcp_server import run
|
|
|
|
assert callable(run)
|
|
|
|
# Test the docstring exists
|
|
assert run.__doc__ is not None
|
|
assert "transport" in run.__doc__.lower()
|