162 lines
5.3 KiB
Python
162 lines
5.3 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
|
|
from mcp_server_qdrant.embeddings.fastembed import FastEmbedProvider
|
|
from mcp_server_qdrant.qdrant import Entry, QdrantConnector
|
|
|
|
|
|
@pytest.fixture
|
|
async def embedding_provider():
|
|
"""Fixture to provide a FastEmbed embedding provider."""
|
|
return FastEmbedProvider(model_name="sentence-transformers/all-MiniLM-L6-v2")
|
|
|
|
|
|
@pytest.fixture
|
|
async def qdrant_connector(embedding_provider):
|
|
"""Fixture to provide a QdrantConnector with in-memory Qdrant client."""
|
|
# Use a random collection name to avoid conflicts between tests
|
|
collection_name = f"test_collection_{uuid.uuid4().hex}"
|
|
|
|
# Create connector with in-memory Qdrant
|
|
connector = QdrantConnector(
|
|
qdrant_url=":memory:",
|
|
qdrant_api_key=None,
|
|
collection_name=collection_name,
|
|
embedding_provider=embedding_provider,
|
|
)
|
|
|
|
yield connector
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_and_search(qdrant_connector):
|
|
"""Test storing an entry and then searching for it."""
|
|
# Store a test entry
|
|
test_entry = Entry(
|
|
content="The quick brown fox jumps over the lazy dog",
|
|
metadata={"source": "test", "importance": "high"},
|
|
)
|
|
await qdrant_connector.store(test_entry)
|
|
|
|
# Search for the entry
|
|
results = await qdrant_connector.search("fox jumps")
|
|
|
|
# Verify results
|
|
assert len(results) == 1
|
|
assert results[0].content == test_entry.content
|
|
assert results[0].metadata == test_entry.metadata
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_empty_collection(qdrant_connector):
|
|
"""Test searching in an empty collection."""
|
|
# Search in an empty collection
|
|
results = await qdrant_connector.search("test query")
|
|
|
|
# Verify results
|
|
assert len(results) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_entries(qdrant_connector):
|
|
"""Test storing and searching multiple entries."""
|
|
# Store multiple entries
|
|
entries = [
|
|
Entry(
|
|
content="Python is a programming language",
|
|
metadata={"topic": "programming"},
|
|
),
|
|
Entry(content="The Eiffel Tower is in Paris", metadata={"topic": "landmarks"}),
|
|
Entry(content="Machine learning is a subset of AI", metadata={"topic": "AI"}),
|
|
]
|
|
|
|
for entry in entries:
|
|
await qdrant_connector.store(entry)
|
|
|
|
# Search for programming-related entries
|
|
programming_results = await qdrant_connector.search("Python programming")
|
|
assert len(programming_results) > 0
|
|
assert any("Python" in result.content for result in programming_results)
|
|
|
|
# Search for landmark-related entries
|
|
landmark_results = await qdrant_connector.search("Eiffel Tower Paris")
|
|
assert len(landmark_results) > 0
|
|
assert any("Eiffel" in result.content for result in landmark_results)
|
|
|
|
# Search for AI-related entries
|
|
ai_results = await qdrant_connector.search(
|
|
"artificial intelligence machine learning"
|
|
)
|
|
assert len(ai_results) > 0
|
|
assert any("machine learning" in result.content.lower() for result in ai_results)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_collection_exists(qdrant_connector):
|
|
"""Test that the collection is created if it doesn't exist."""
|
|
# The collection shouldn't exist yet
|
|
assert not await qdrant_connector._client.collection_exists(
|
|
qdrant_connector._collection_name
|
|
)
|
|
|
|
# Storing an entry should create the collection
|
|
test_entry = Entry(content="Test content")
|
|
await qdrant_connector.store(test_entry)
|
|
|
|
# Now the collection should exist
|
|
assert await qdrant_connector._client.collection_exists(
|
|
qdrant_connector._collection_name
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metadata_handling(qdrant_connector):
|
|
"""Test that metadata is properly stored and retrieved."""
|
|
# Store entries with different metadata
|
|
metadata1 = {"source": "book", "author": "Jane Doe", "year": 2023}
|
|
metadata2 = {"source": "article", "tags": ["science", "research"]}
|
|
|
|
await qdrant_connector.store(
|
|
Entry(content="Content with structured metadata", metadata=metadata1)
|
|
)
|
|
await qdrant_connector.store(
|
|
Entry(content="Content with list in metadata", metadata=metadata2)
|
|
)
|
|
|
|
# Search and verify metadata is preserved
|
|
results = await qdrant_connector.search("metadata")
|
|
|
|
assert len(results) == 2
|
|
|
|
# Check that both metadata objects are present in the results
|
|
found_metadata1 = False
|
|
found_metadata2 = False
|
|
|
|
for result in results:
|
|
if result.metadata.get("source") == "book":
|
|
assert result.metadata.get("author") == "Jane Doe"
|
|
assert result.metadata.get("year") == 2023
|
|
found_metadata1 = True
|
|
elif result.metadata.get("source") == "article":
|
|
assert "science" in result.metadata.get("tags", [])
|
|
assert "research" in result.metadata.get("tags", [])
|
|
found_metadata2 = True
|
|
|
|
assert found_metadata1
|
|
assert found_metadata2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_entry_without_metadata(qdrant_connector):
|
|
"""Test storing and retrieving entries without metadata."""
|
|
# Store an entry without metadata
|
|
await qdrant_connector.store(Entry(content="Entry without metadata"))
|
|
|
|
# Search and verify
|
|
results = await qdrant_connector.search("without metadata")
|
|
|
|
assert len(results) == 1
|
|
assert results[0].content == "Entry without metadata"
|
|
assert results[0].metadata is None
|