"""Integration tests for mcp-maildir with a real Qdrant instance. Env vars (QDRANT_URL, COLLECTION_NAME, EMBEDDING_MODEL_NAME) are set in tests/conftest.py — don't set them here, they must match server.py's import-time snapshot for every module in the session. """ import os import uuid import pytest from server import get_qdrant_client, get_embedding_model, search_emails, read_email from qdrant_client.http import models # Requires a live Qdrant on localhost:6333 (see pytest.ini: deselected by # default; run with `pytest -m integration`, nightly CI does exactly that). pytestmark = pytest.mark.integration TEST_EMAILS = [ { "message_id": "", "date": "2026-01-15T10:00:00", "sender": "alice@example.com", "sender_raw": "Alice ", "receiver": "bob@example.com", "receiver_raw": "Bob ", "subject": "Hello World", "body_text": "This is a test email about Python programming and vector databases.", "attachments": [], }, { "message_id": "", "date": "2026-01-16T14:30:00", "sender": "carol@other.com", "sender_raw": "Carol ", "receiver": "alice@example.com", "receiver_raw": "Alice ", "subject": "Qdrant setup help", "body_text": "Can you help me set up Qdrant for semantic email search?", "attachments": ["screenshot.png"], }, { "message_id": "", "date": "2026-02-01T09:00:00", "sender": "bob@example.com", "sender_raw": "Bob ", "receiver": "alice@example.com", "receiver_raw": "Alice ", "subject": "Meeting next week", "body_text": "Let's discuss the project roadmap and data pipeline.", "attachments": [], }, ] @pytest.fixture(scope="module") def qdrant_setup(): """Create collection and index test emails once per test run.""" client = get_qdrant_client() model = get_embedding_model() collection_name = os.environ["COLLECTION_NAME"] # Debug instrumentalisation — truth about which server we are talking to # and what it holds (CI 404 diagnosis). import logging dbg = logging.getLogger("integration.debug") try: dbg.warning("qdrant client url: %s", client.http._client.base_url) except Exception: dbg.warning("qdrant client url: ") server_info = None try: server_info = client.get_collections() dbg.warning("collections BEFORE: %s", [c.name for c in server_info.collections]) except Exception as e: dbg.warning("get_collections FAILED: %s", e) # Clean up from previous runs try: client.delete_collection(collection_name) except Exception: pass # Create collection probe = next(iter(model.embed(["dimension_probe"]))) client.create_collection( collection_name=collection_name, vectors_config=models.VectorParams( size=len(probe), distance=models.Distance.COSINE ), ) # Assert the collection is actually visible right after creation names_after = [c.name for c in client.get_collections().collections] dbg.warning("collections AFTER create: %s", names_after) if collection_name not in names_after: raise RuntimeError( f"Fixture created collection {collection_name} but it is not in " f"get_collections(): {names_after}" ) # Index test emails for email in TEST_EMAILS: vector_text = ( f"Date: {email['date']}\n" f"From: {email['sender']}\n" f"To: {email['receiver']}\n" f"Subject: {email['subject']}\n\n" f"{email['body_text']}\n\n" f"Attachments: {', '.join(email['attachments']) if email['attachments'] else 'None'}" ) vector = list(model.embed([vector_text]))[0].tolist() # Create payload indexes on first upsert client.create_payload_index( collection_name=collection_name, field_name="sender", field_schema=models.PayloadSchemaType.KEYWORD, ) client.create_payload_index( collection_name=collection_name, field_name="receiver", field_schema=models.PayloadSchemaType.KEYWORD, ) client.create_payload_index( collection_name=collection_name, field_name="date", field_schema=models.PayloadSchemaType.DATETIME, ) client.upsert( collection_name=collection_name, points=[ models.PointStruct( id=str(uuid.uuid5(uuid.NAMESPACE_OID, email["message_id"])), vector=vector, payload={ "message_id": email["message_id"], "date": email["date"], "sender": email["sender"], "sender_raw": email["sender_raw"], "receiver": email["receiver"], "receiver_raw": email["receiver_raw"], "subject": email["subject"], "body_text": email["body_text"], "attachments": email["attachments"], }, ) ], ) yield client # Cleanup try: client.delete_collection(collection_name) except Exception: pass class TestSearchEmails: def test_search_by_content(self, qdrant_setup): """Semantic search returns the most relevant email first. Searching for 'Python programming' should match email 001 which discusses Python and vector databases. """ result = search_emails(query="Python programming") assert result["count"] >= 1 assert "results" in result # The top result should be email 001 (most semantically relevant) messages = [r["message_id"] for r in result["results"]] assert "" in messages def test_search_with_participant_filter(self, qdrant_setup): """Search emails sent by alice.""" result = search_emails(query="help", participant="alice@example.com") assert result["count"] >= 1 messages = [r["message_id"] for r in result["results"]] assert "" in messages def test_search_with_date_filter(self, qdrant_setup): """Search emails after a specific date.""" result = search_emails(query="meeting", start_date="2026-02-01") assert result["count"] >= 1 for r in result["results"]: date = r.get("date", "") assert date >= "2026-02-01" def test_search_no_results_by_date(self, qdrant_setup): """Search with a filter that matches no emails returns empty results. With semantic (vector) search, any text query always has nearest neighbors, so we use a date filter that matches nothing instead. """ result = search_emails( query="anything", start_date="2099-01-01", end_date="2099-12-31", ) assert result["count"] == 0 assert result["results"] == [] class TestReadEmail: def test_read_existing_email(self, qdrant_setup): """Read an email by its message_id.""" result = read_email(message_id="") assert "error" not in result assert result["message_id"] == "" assert "This is a test email" in result.get("body_text", "") def test_read_nonexistent_email(self, qdrant_setup): """Read an email that doesn't exist.""" result = read_email(message_id="") assert "error" in result assert "No email found" in result["error"]