Initial commit
This commit is contained in:
423
tests/test_rag.py
Normal file
423
tests/test_rag.py
Normal file
@ -0,0 +1,423 @@
|
||||
"""
|
||||
Suite de tests unitaires pour le système RAG (Indexer, Datastore, Retriever)
|
||||
|
||||
Adapté à votre implémentation spécifique avec:
|
||||
- BaseDatastore (pas BaseDataStore)
|
||||
- Datastore (pas DataStore)
|
||||
- DataItem avec content/source (pas text/metadata)
|
||||
|
||||
Pour exécuter:
|
||||
pytest tests/test_rag.py -v
|
||||
|
||||
Pour exécuter avec couverture:
|
||||
pytest tests/test_rag.py --cov=src --cov-report=html
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Configuration du PYTHONPATH
|
||||
project_root = Path(__file__).parent.parent
|
||||
if str(project_root) not in sys.path:
|
||||
sys.path.insert(0, str(project_root))
|
||||
|
||||
import pytest
|
||||
import tempfile
|
||||
import shutil
|
||||
from typing import List
|
||||
|
||||
# Imports adaptés à votre structure
|
||||
from src.interface.base_datastore import BaseDatastore, DataItem
|
||||
from src.impl.datastore import Datastore
|
||||
|
||||
# Tentative d'import des autres composants (adapter selon vos fichiers)
|
||||
try:
|
||||
from src.interface.base_indexer import BaseIndexer
|
||||
from src.impl.indexer import Indexer
|
||||
HAS_INDEXER = True
|
||||
except ImportError:
|
||||
HAS_INDEXER = False
|
||||
print("⚠️ Indexer non trouvé - tests d'indexation désactivés")
|
||||
|
||||
try:
|
||||
from src.interface.base_retriever import BaseRetriever
|
||||
from src.impl.retriever import Retriever
|
||||
HAS_RETRIEVER = True
|
||||
except ImportError:
|
||||
HAS_RETRIEVER = False
|
||||
print("⚠️ Retriever non trouvé - tests de récupération désactivés")
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# FIXTURES - Configuration des tests
|
||||
# ============================================================================
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir():
|
||||
"""Crée un répertoire temporaire pour les tests."""
|
||||
temp_path = tempfile.mkdtemp()
|
||||
yield temp_path
|
||||
shutil.rmtree(temp_path, ignore_errors=True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_items():
|
||||
"""Crée des DataItems de test."""
|
||||
return [
|
||||
DataItem(
|
||||
content="L'intelligence artificielle (IA) est un domaine de l'informatique.",
|
||||
source="doc1.pdf"
|
||||
),
|
||||
DataItem(
|
||||
content="Python est un langage de programmation populaire pour l'IA.",
|
||||
source="doc2.pdf"
|
||||
),
|
||||
DataItem(
|
||||
content="Les réseaux de neurones sont utilisés en deep learning.",
|
||||
source="doc3.pdf"
|
||||
),
|
||||
DataItem(
|
||||
content="Le machine learning permet aux ordinateurs d'apprendre sans être explicitement programmés.",
|
||||
source="doc4.pdf"
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def datastore(temp_dir, monkeypatch):
|
||||
"""Crée une instance du Datastore avec une base temporaire."""
|
||||
# Modifier le chemin de la DB pour les tests
|
||||
test_db_path = str(Path(temp_dir) / "test-lancedb")
|
||||
monkeypatch.setattr("src.impl.datastore.Datastore.DB_PATH", test_db_path)
|
||||
|
||||
ds = Datastore()
|
||||
ds.reset_table() # S'assurer que la table est vide
|
||||
yield ds
|
||||
|
||||
# Cleanup
|
||||
try:
|
||||
ds.vector_db.drop_table(ds.DB_TABLE_NAME)
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def populated_datastore(datastore, sample_items):
|
||||
"""Datastore pré-rempli avec des items."""
|
||||
datastore.add_items(sample_items)
|
||||
return datastore
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def indexer():
|
||||
"""Crée une instance de l'Indexer si disponible."""
|
||||
if not HAS_INDEXER:
|
||||
pytest.skip("Indexer non disponible")
|
||||
return Indexer()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def retriever(datastore):
|
||||
"""Crée une instance du Retriever si disponible."""
|
||||
if not HAS_RETRIEVER:
|
||||
pytest.skip("Retriever non disponible")
|
||||
return Retriever(datastore=datastore)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS DATAITEM
|
||||
# ============================================================================
|
||||
|
||||
class TestDataItem:
|
||||
"""Tests pour la classe DataItem."""
|
||||
|
||||
def test_dataitem_creation(self):
|
||||
"""Test la création d'un DataItem."""
|
||||
item = DataItem(content="Test content", source="test.pdf")
|
||||
assert item.content == "Test content"
|
||||
assert item.source == "test.pdf"
|
||||
|
||||
def test_dataitem_default_values(self):
|
||||
"""Test les valeurs par défaut."""
|
||||
item = DataItem()
|
||||
assert item.content == ""
|
||||
assert item.source == ""
|
||||
|
||||
def test_dataitem_validation(self):
|
||||
"""Test la validation Pydantic."""
|
||||
# Pydantic devrait accepter ces types
|
||||
item = DataItem(content="text", source="source.pdf")
|
||||
assert isinstance(item.content, str)
|
||||
assert isinstance(item.source, str)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS DATASTORE
|
||||
# ============================================================================
|
||||
|
||||
class TestDatastore:
|
||||
"""Tests pour la classe Datastore."""
|
||||
|
||||
def test_datastore_initialization(self, datastore):
|
||||
"""Test l'initialisation du Datastore."""
|
||||
assert isinstance(datastore, BaseDatastore)
|
||||
assert datastore.vector_dimensions == 384
|
||||
assert datastore.model is not None
|
||||
assert datastore.table is not None
|
||||
|
||||
def test_reset_table(self, datastore):
|
||||
"""Test le reset de la table."""
|
||||
# Ajouter des items
|
||||
items = [DataItem(content="test", source="test.pdf")]
|
||||
datastore.add_items(items)
|
||||
|
||||
# Reset
|
||||
table = datastore.reset_table()
|
||||
assert table is not None
|
||||
|
||||
# Vérifier que la table est vide
|
||||
results = datastore.search_datastore("test", top_k=10)
|
||||
assert len(results) == 0
|
||||
|
||||
def test_add_single_item(self, datastore):
|
||||
"""Test l'ajout d'un seul item."""
|
||||
item = DataItem(
|
||||
content="Test de l'ajout d'un item unique",
|
||||
source="test_single.pdf"
|
||||
)
|
||||
datastore.add_items([item])
|
||||
|
||||
# Vérifier que l'item peut être retrouvé
|
||||
results = datastore.search_datastore("test ajout", top_k=5)
|
||||
assert len(results) > 0
|
||||
|
||||
def test_add_multiple_items(self, datastore, sample_items):
|
||||
"""Test l'ajout de plusieurs items."""
|
||||
datastore.add_items(sample_items)
|
||||
|
||||
# Vérifier que plusieurs items ont été ajoutés
|
||||
results = datastore.search_datastore("intelligence", top_k=10)
|
||||
assert len(results) > 0
|
||||
|
||||
def test_add_empty_list(self, datastore):
|
||||
"""Test l'ajout d'une liste vide."""
|
||||
# Ne devrait pas crasher
|
||||
datastore.add_items([])
|
||||
# Pas d'exception = succès
|
||||
|
||||
def test_create_vector(self, datastore):
|
||||
"""Test la création de vecteurs."""
|
||||
vector = datastore.create_vector("Test content")
|
||||
|
||||
assert isinstance(vector, list)
|
||||
assert len(vector) == 384 # Dimension du modèle all-MiniLM-L6-v2
|
||||
assert all(isinstance(v, float) for v in vector)
|
||||
|
||||
def test_create_vector_consistency(self, datastore):
|
||||
"""Test que le même texte produit le même vecteur."""
|
||||
text = "Texte de test pour la cohérence"
|
||||
vector1 = datastore.create_vector(text)
|
||||
vector2 = datastore.create_vector(text)
|
||||
|
||||
# Les vecteurs devraient être identiques (ou très proches)
|
||||
import numpy as np
|
||||
similarity = np.dot(vector1, vector2) / (np.linalg.norm(vector1) * np.linalg.norm(vector2))
|
||||
assert similarity > 0.99 # Très haute similarité
|
||||
|
||||
def test_search_basic(self, populated_datastore):
|
||||
"""Test une recherche basique."""
|
||||
results = populated_datastore.search_datastore("intelligence artificielle", top_k=3)
|
||||
|
||||
assert len(results) > 0
|
||||
assert len(results) <= 3
|
||||
assert all(isinstance(r, str) for r in results)
|
||||
|
||||
def test_search_relevance(self, populated_datastore):
|
||||
"""Test la pertinence des résultats."""
|
||||
results = populated_datastore.search_datastore("Python programmation", top_k=5)
|
||||
|
||||
assert len(results) > 0
|
||||
# Le premier résultat devrait contenir "Python"
|
||||
assert any("Python" in results[0] or "python" in results[0].lower()
|
||||
for _ in [results[0]])
|
||||
|
||||
def test_search_top_k_limit(self, populated_datastore):
|
||||
"""Test que top_k limite correctement les résultats."""
|
||||
results_2 = populated_datastore.search_datastore("test", top_k=2)
|
||||
results_4 = populated_datastore.search_datastore("test", top_k=4)
|
||||
|
||||
assert len(results_2) <= 2
|
||||
assert len(results_4) <= 4
|
||||
|
||||
def test_search_empty_query(self, populated_datastore):
|
||||
"""Test avec une requête vide."""
|
||||
results = populated_datastore.search_datastore("", top_k=5)
|
||||
# Devrait retourner des résultats ou liste vide, pas d'erreur
|
||||
assert isinstance(results, list)
|
||||
|
||||
def test_search_no_results(self, datastore):
|
||||
"""Test une recherche sur une base vide."""
|
||||
results = datastore.search_datastore("query inexistante", top_k=5)
|
||||
assert results == []
|
||||
|
||||
def test_search_special_characters(self, populated_datastore):
|
||||
"""Test avec des caractères spéciaux."""
|
||||
results = populated_datastore.search_datastore("l'intelligence", top_k=5)
|
||||
assert isinstance(results, list)
|
||||
|
||||
def test_multiple_searches_consistency(self, populated_datastore):
|
||||
"""Test la cohérence sur plusieurs recherches."""
|
||||
query = "intelligence artificielle"
|
||||
results1 = populated_datastore.search_datastore(query, top_k=3)
|
||||
results2 = populated_datastore.search_datastore(query, top_k=3)
|
||||
|
||||
# Les résultats devraient être identiques
|
||||
assert results1 == results2
|
||||
|
||||
def test_large_batch_add(self, datastore):
|
||||
"""Test l'ajout d'un grand nombre d'items."""
|
||||
large_batch = [
|
||||
DataItem(content=f"Document numéro {i} avec du contenu varié.", source=f"doc_{i}.pdf")
|
||||
for i in range(100)
|
||||
]
|
||||
|
||||
# Ne devrait pas crasher
|
||||
datastore.add_items(large_batch)
|
||||
|
||||
# Vérifier que des résultats sont retournés
|
||||
results = datastore.search_datastore("document", top_k=10)
|
||||
assert len(results) > 0
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS INDEXER (si disponible)
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.skipif(not HAS_INDEXER, reason="Indexer non disponible")
|
||||
class TestIndexer:
|
||||
"""Tests pour la classe Indexer."""
|
||||
|
||||
def test_indexer_initialization(self, indexer):
|
||||
"""Test l'initialisation de l'Indexer."""
|
||||
assert isinstance(indexer, BaseIndexer)
|
||||
|
||||
def test_index_documents(self, indexer, temp_dir):
|
||||
"""Test l'indexation de documents."""
|
||||
# Créer un document de test
|
||||
doc_path = Path(temp_dir) / "test_doc.pdf"
|
||||
doc_path.write_text("Contenu de test pour l'indexation.")
|
||||
|
||||
items = indexer.index([str(doc_path)])
|
||||
|
||||
assert len(items) > 0
|
||||
assert all(isinstance(item, DataItem) for item in items)
|
||||
assert all(hasattr(item, 'content') for item in items)
|
||||
assert all(hasattr(item, 'source') for item in items)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS RETRIEVER (si disponible)
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.skipif(not HAS_RETRIEVER, reason="Retriever non disponible")
|
||||
class TestRetriever:
|
||||
"""Tests pour la classe Retriever."""
|
||||
|
||||
def test_retriever_initialization(self, retriever):
|
||||
"""Test l'initialisation du Retriever."""
|
||||
assert isinstance(retriever, BaseRetriever)
|
||||
assert retriever.datastore is not None
|
||||
|
||||
def test_retrieve_basic(self, retriever, populated_datastore):
|
||||
"""Test une récupération basique."""
|
||||
retriever.datastore = populated_datastore
|
||||
results = retriever.search_retriever("intelligence artificielle", top_k=3)
|
||||
|
||||
assert len(results) > 0
|
||||
assert len(results) <= 3
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS D'INTÉGRATION
|
||||
# ============================================================================
|
||||
|
||||
class TestIntegration:
|
||||
"""Tests d'intégration du système complet."""
|
||||
|
||||
@pytest.mark.skipif(not HAS_INDEXER, reason="Indexer requis")
|
||||
def test_full_pipeline(self, indexer, datastore, temp_dir):
|
||||
"""Test du pipeline complet: indexation → stockage → recherche."""
|
||||
# 1. Créer un document
|
||||
doc_path = Path(temp_dir) / "integration_test.pdf"
|
||||
doc_path.write_text("""
|
||||
L'intelligence artificielle transforme le monde.
|
||||
Python est le langage privilégié pour l'IA.
|
||||
Les algorithmes de machine learning sont puissants.
|
||||
""")
|
||||
|
||||
# 2. Indexation
|
||||
items = indexer.index([str(doc_path)])
|
||||
assert len(items) > 0
|
||||
|
||||
# 3. Stockage
|
||||
datastore.add_items(items)
|
||||
|
||||
# 4. Recherche
|
||||
results = datastore.search_datastore("intelligence artificielle Python", top_k=3)
|
||||
assert len(results) > 0
|
||||
|
||||
# 5. Vérifier la pertinence
|
||||
top_result = results[0].lower()
|
||||
assert "intelligence" in top_result or "python" in top_result
|
||||
|
||||
def test_incremental_addition(self, datastore, sample_items):
|
||||
"""Test l'ajout incrémental de données."""
|
||||
# Ajouter en plusieurs fois
|
||||
datastore.add_items(sample_items[:2])
|
||||
results_1 = datastore.search_datastore("test", top_k=10)
|
||||
|
||||
datastore.add_items(sample_items[2:])
|
||||
results_2 = datastore.search_datastore("test", top_k=10)
|
||||
|
||||
# Le second devrait avoir plus de résultats potentiels
|
||||
assert len(results_2) >= len(results_1)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS DE ROBUSTESSE
|
||||
# ============================================================================
|
||||
|
||||
class TestRobustness:
|
||||
"""Tests de robustesse et gestion d'erreurs."""
|
||||
|
||||
def test_unicode_content(self, datastore):
|
||||
"""Test avec du contenu Unicode."""
|
||||
items = [
|
||||
DataItem(content="Émojis: 🎉 🎨 🚀", source="unicode.pdf"),
|
||||
DataItem(content="Caractères spéciaux: é à ç ñ", source="special.pdf"),
|
||||
]
|
||||
|
||||
datastore.add_items(items)
|
||||
results = datastore.search_datastore("émojis caractères", top_k=5)
|
||||
assert isinstance(results, list)
|
||||
|
||||
def test_very_long_content(self, datastore):
|
||||
"""Test avec du contenu très long."""
|
||||
long_content = "Test. " * 1000 # ~6000 caractères
|
||||
item = DataItem(content=long_content, source="long.pdf")
|
||||
|
||||
# Ne devrait pas crasher
|
||||
datastore.add_items([item])
|
||||
results = datastore.search_datastore("test", top_k=1)
|
||||
assert len(results) >= 0
|
||||
|
||||
def test_empty_content(self, datastore):
|
||||
"""Test avec du contenu vide."""
|
||||
items = [DataItem(content="", source="empty.pdf")]
|
||||
|
||||
# Ne devrait pas crasher
|
||||
datastore.add_items(items)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v", "--tb=short"])
|
||||
Reference in New Issue
Block a user