atara57769's picture
feat: implement modular RAG architecture, query routing, metadata-filtered retrieval, and a comprehensive offline testing suite.
af355c6
Raw
History Blame Contribute Delete
2.81 kB
import os
import sys
from unittest.mock import MagicMock
# Ensure the workspace root is in python path
workspace_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if workspace_root not in sys.path:
sys.path.insert(0, workspace_root)
# 1. Custom Mock Classes to emulate langchain / qdrant / groq behaviors
class MockHuggingFaceEmbeddings:
def __init__(self, model_name=None, *args, **kwargs):
self.model_name = model_name
class MockQdrantVectorStore:
def __init__(self, client=None, collection_name=None, embedding=None, *args, **kwargs):
self.client = client
self.collection_name = collection_name
self.embedding = embedding
self.docs = []
def similarity_search(self, query, k=3, filter=None, **kwargs):
return self.docs
def add_texts(self, texts, metadatas=None, ids=None, **kwargs):
return ids or [f"mock-uuid-{i}" for i in range(len(texts))]
class MockQdrantClient:
def __init__(self, *args, **kwargs):
pass
def collection_exists(self, collection_name):
return True
def create_collection(self, *args, **kwargs):
pass
def delete_collection(self, *args, **kwargs):
pass
def create_payload_index(self, *args, **kwargs):
pass
class MockGroq:
def __init__(self, api_key=None, *args, **kwargs):
self.api_key = api_key
self.chat = MockChat()
class MockChat:
def __init__(self):
self.completions = MockCompletions()
class MockCompletions:
def __init__(self):
pass
def create(self, *args, **kwargs):
mock_response = MagicMock()
mock_choice = MagicMock()
mock_message = MagicMock()
mock_message.content = "mock response content"
mock_choice.message = mock_message
mock_response.choices = [mock_choice]
return mock_response
# 2. Inject the mocks into sys.modules BEFORE any service is imported
sys.modules['langchain_huggingface'] = MagicMock()
sys.modules['langchain_huggingface'].HuggingFaceEmbeddings = MockHuggingFaceEmbeddings
sys.modules['langchain_qdrant'] = MagicMock()
sys.modules['langchain_qdrant'].QdrantVectorStore = MockQdrantVectorStore
sys.modules['qdrant_client'] = MagicMock()
sys.modules['qdrant_client'].QdrantClient = MockQdrantClient
# Add models schema support to mock qdrant_client.models
mock_models = MagicMock()
sys.modules['qdrant_client.models'] = mock_models
sys.modules['qdrant_client'].models = mock_models
sys.modules['groq'] = MagicMock()
sys.modules['groq'].Groq = MockGroq
# 3. Setup common fixtures
import pytest
@pytest.fixture(autouse=True)
def clean_mock_states():
"""Fixture to ensure a fresh, clean slate between each test run."""
# Reset mock responses or states if needed
pass