423 lines
15 KiB
Python
423 lines
15 KiB
Python
"""
|
|
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"]) |