""" Tests de la dimension, de la normalisation et de la cle de cache des embeddings. Chaque test correspond a un mecanisme qui n'existait pas : - output_dimensionality n'etait pas demande a l'API ; - la renormalisation des sorties tronquees (Matryoshka) n'existait pas, alors que l'operateur <=> de pgvector suppose des vecteurs unitaires ; - le chemin batch n'avait ni garde de dimension ni controle du nombre de vecteurs rendus ; - la cle de cache ignorait la dimension (un vecteur 3072 aurait ete resservi sous configuration 1536) et minusculait le texte. """ import json import httpx import numpy as np import pytest from app.services import embedding_service as module from app.services.embedding_service import EmbeddingService, EmbeddingServiceError @pytest.fixture def recorder(monkeypatch): """Doublure HTTP de Voyage qui enregistre le corps de chaque requete.""" state = {"payloads": [], "n_returned": None, "dim": None, "scale": 9.0} def _repondre(requete: httpx.Request) -> httpx.Response: payload = json.loads(requete.content) state["payloads"].append(payload) dim = state["dim"] or payload["output_dimension"] n = state["n_returned"] if state["n_returned"] is not None else len(payload["input"]) rng = np.random.default_rng(7) # Vecteurs NON unitaires : c'est le service qui doit normaliser. return httpx.Response(200, json={"data": [ {"index": i, "embedding": (rng.random(dim) * state["scale"]).tolist()} for i in range(n) ]}) monkeypatch.setattr(module, "transport_http", httpx.MockTransport(_repondre)) return state @pytest.fixture def service(recorder): return EmbeddingService(use_cache=False) class TestDimension: def test_output_dimension_is_requested(self, service, recorder): service.generate_embedding("Article premier du code civil") assert recorder["payloads"][-1]["output_dimension"] == EmbeddingService.EMBEDDING_DIM # Valeur absolue et non derivee : le but est de remarquer un changement # accidentel de dimension, pas de le suivre. assert EmbeddingService.EMBEDDING_DIM == 1024 def test_embedding_has_configured_dimension(self, service): embedding = service.generate_embedding("Article premier") assert embedding.shape == (EmbeddingService.EMBEDDING_DIM,) def test_batch_rejects_wrong_dimension(self, service, recorder): recorder["dim"] = 512 # l'API renvoie autre chose que ce qui est demande with pytest.raises(EmbeddingServiceError, match="Dimension"): service.generate_batch_embeddings(["a", "b"]) def test_batch_rejects_truncated_response(self, service, recorder): """ Deux vecteurs pour trois textes : zip() perdait silencieusement la queue du lot, puis un KeyError tombait bien plus loin. """ recorder["n_returned"] = 2 with pytest.raises(EmbeddingServiceError, match="incomplète"): service.generate_batch_embeddings(["a", "b", "c"]) class TestNormalisation: def test_embedding_is_unit_norm(self, service): embedding = service.generate_embedding("Responsabilité des dirigeants") assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5 def test_batch_embeddings_are_unit_norm(self, service): embeddings = service.generate_batch_embeddings(["un", "deux", "trois"]) for embedding in embeddings: assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5 def test_normalisation_applies_even_when_not_requested(self, service): """ La normalisation est INCONDITIONNELLE, a toute dimension y compris la native. Une version precedente la sautait quand normalize=False a la dimension native, en supposant que l'API renvoie deja un vecteur unitaire. C'est vrai de l'API, mais ca fait dependre d'un reglage une garantie dont tout l'aval depend : similarity() est un np.dot nu, et le score 1 - distance borne a [0, 1] ne vaut que sur la sphere unite. """ embedding = service.generate_embedding("Test", normalize=False) assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5 class TestCacheKey: def test_key_isolates_dimension(self, service): key_native = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT) original = EmbeddingService.EMBEDDING_DIM try: # Une dimension differente de celle configuree, sinon les deux cles # sont identiques et le test ne verifie rien. service.EMBEDDING_DIM = 512 key_512 = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT) finally: service.EMBEDDING_DIM = original assert key_native != key_512 def test_key_isolates_task_type(self, service): as_document = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT) as_query = service._cache_key("texte", EmbeddingService.TASK_QUERY) assert as_document != as_query def test_key_is_case_sensitive(self, service): """ La cle minusculait le texte : "Article PREMIER" et "article premier" partageaient une entree alors que le modele les encode differemment. """ upper = service._cache_key("Article PREMIER", EmbeddingService.TASK_DOCUMENT) lower = service._cache_key("article premier", EmbeddingService.TASK_DOCUMENT) assert upper != lower class TestCacheRead: @pytest.mark.asyncio async def test_cache_read_rejects_wrong_dimension(self, recorder, db_session): """ Une entree ecrite sous une autre dimension (ici 3072, la configuration precedente) doit etre ignoree, pas servie : elle serait rejetee par la colonne, ou pire, comparee de travers. """ from sqlalchemy import text as sql_text service = EmbeddingService(use_cache=True) key = service._cache_key("texte périmé", EmbeddingService.TASK_DOCUMENT) await db_session.execute( sql_text( "INSERT INTO embedding_cache (text_hash, embedding_json, expires_at) " "VALUES (:key, :data, now() + interval '1 day')" ), {"key": key, "data": json.dumps([0.1] * 3072)}, ) await db_session.commit() assert service._get_from_pg_cache(key) is None class TestOffLoop: @pytest.mark.asyncio async def test_generate_embedding_async_runs_off_the_event_loop(self, service): """ La recherche semantique appelait generate_embedding depuis une coroutine : la boucle etait gelee pour l'aller-retour reseau, le time.sleep du backoff et les E/S psycopg2 du cache. """ import threading loop_thread = threading.get_ident() seen = {} original = service._produire def _spy(*args, **kwargs): seen["thread"] = threading.get_ident() return original(*args, **kwargs) service._produire = _spy embedding = await service.generate_embedding_async("Une question") assert embedding.shape == (EmbeddingService.EMBEDDING_DIM,) assert seen["thread"] != loop_thread @pytest.mark.asyncio async def test_async_wrapper_defaults_to_query_task(self, service, recorder): """Son seul appelant encode une question, pas un document.""" await service.generate_embedding_async("Une question") assert recorder["payloads"][-1]["input_type"] == "query" def test_batch_encodes_documents(self, service, recorder): """ L'autre moitie de l'asymetrie de la recherche : une inversion encoderait les articles comme des questions, sans erreur. """ service.generate_batch_embeddings(["Article premier", "Article 2"]) assert recorder["payloads"][-1]["input_type"] == "document" class TestRequestTimeout: """ Le client Gemini doit imposer un delai maximal. Sans http_options, le client attend indefiniment. Observe en conditions reelles pendant l'ingestion du corpus : le processus est reste 40 minutes endormi sur une lecture de prise reseau, vivant, sans consommer d'UC. Aucune exception n'etant levee, la boucle de reprise du service ne se declenchait jamais — un blocage silencieux et sans fin. """ def test_embedding_client_has_a_timeout(self): # Sans doublure : c'est le VRAI client Gemini qu'on inspecte. Sa # construction n'ouvre aucune connexion. Les delais de Voyage sont # verifies dans test_embedding_voyage.py. from app.core.config import settings from app.services.embedding_service import GeminiProvider service = EmbeddingService(use_cache=False, provider=GeminiProvider(api_key="x")) configured = service.client._api_client._http_options.timeout assert configured == settings.GEMINI_TIMEOUT_S * 1000 # millisecondes assert configured > 0 def test_generation_client_has_a_timeout(self): from app.core.config import settings from app.services.gemini_service import GeminiService service = GeminiService() assert service.client._api_client._http_options.timeout == ( settings.GEMINI_TIMEOUT_S * 1000 ) class TestDailyQuota: """ Un quota JOURNALIER ne se rejoue pas. Google renvoie 429 aussi bien pour un quota par minute que par jour, et suggere un retryDelay d'une trentaine de secondes dans les deux cas. Pour un quota journalier ce delai est trompeur : reessayer trois fois brule une minute avant d'echouer de toute facon, et le lot suivant recommence. Seul le quotaId permet de trancher. """ def test_recognises_a_daily_quota(self): from app.services.embedding_service import _is_daily_quota_error error = Exception( "429 RESOURCE_EXHAUSTED {'quotaId': " "'EmbedContentRequestsPerDayPerUserPerProjectPerModel-FreeTier'}" ) assert _is_daily_quota_error(error) is True def test_a_per_minute_quota_stays_retryable(self): from app.services.embedding_service import _is_daily_quota_error error = Exception( "429 RESOURCE_EXHAUSTED {'quotaId': 'EmbedContentRequestsPerMinutePerProject'}" ) assert _is_daily_quota_error(error) is False def test_other_errors_are_not_quota(self): from app.services.embedding_service import _is_daily_quota_error assert _is_daily_quota_error(Exception("500 INTERNAL")) is False assert _is_daily_quota_error(Exception("401 UNAUTHENTICATED")) is False def test_batch_raises_immediately_without_retrying(self, monkeypatch): """Aucune tentative supplementaire : on leve au premier 429 journalier.""" from app.services.embedding_service import GeminiProvider, QuotaExhaustedError service = EmbeddingService(use_cache=False, provider=GeminiProvider(api_key="x")) calls = {"n": 0} def _always_quota(**kwargs): calls["n"] += 1 raise RuntimeError( "429 RESOURCE_EXHAUSTED {'quotaId': " "'EmbedContentRequestsPerDayPerUserPerProjectPerModel-FreeTier'}" ) monkeypatch.setattr(service.client.models, "embed_content", _always_quota) monkeypatch.setattr(module.time, "sleep", lambda *_: None) with pytest.raises(QuotaExhaustedError): service.generate_batch_embeddings(["a", "b"]) assert calls["n"] == 1, "un quota journalier ne doit pas etre rejoue"