"""Teste de flux warmup + cache persistent la nivelul `ensure_embeddings_corpus` (app/mapping.py): cold/warm start, incremental, model schimbat, hash pe lista filtrata, concurenta warmup/request sub lock (A9), echivalenta ranking cu toleranta float32 (A5). Backend mock determinist (fara fastembed real). Toate testele seteaza explicit AUTOPASS_EMBEDDINGS_ENABLED=1 (A13d). """ from __future__ import annotations import hashlib import os import tempfile import threading import time import pytest from app.embedding_cache import EMB_DIM def _det_vector(text: str, dim: int = EMB_DIM) -> list[float]: """Vector determinist (384-dim) derivat din hash-ul textului. Suficient pentru ranking cosine in teste -- nu evaluam calitatea semantica, doar alinierea/cache-ul.""" digest = hashlib.sha256(text.encode("utf-8")).digest() return [((digest[i % len(digest)] + i) % 256) / 255.0 for i in range(dim)] class CountingMockBackend: """Backend determinist care numara textele primite la fiecare apel embed().""" def __init__(self): self.calls: list[list[str]] = [] def embed(self, texts): self.calls.append(list(texts)) return [_det_vector(t) for t in texts] @property def total_texts(self) -> int: return sum(len(c) for c in self.calls) @pytest.fixture() def env(monkeypatch): tmp = tempfile.mkdtemp() monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "warmup.db")) monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "false") monkeypatch.setenv("AUTOPASS_EMBEDDINGS_ENABLED", "true") # A13d: anti-vacuos from app.config import get_settings get_settings.cache_clear() from app.db import init_db init_db() yield monkeypatch get_settings.cache_clear() @pytest.fixture() def conn(env): from app.db import get_connection c = get_connection() yield c c.close() def _inject_engine(backend): import app.embeddings as emb from app.embeddings import EmbeddingEngine emb._engine = EmbeddingEngine(backend=backend) return emb def _seed_silver(conn, rows): """rows = [(denumire_normalizata, cod, is_nul)].""" conn.executemany( "INSERT OR IGNORE INTO mapping_suggestions " "(denumire_normalizata, cod_prestatie, is_nul, source, confidence) VALUES (?, ?, ?, 'llm_seed', 0.7)", rows, ) conn.commit() # --------------------------------------------------------------------------- # # Cold / warm / incremental start (US-005) # # --------------------------------------------------------------------------- # def test_cold_start_trimite_exact_n_texte_si_populeaza_cache(conn): backend = CountingMockBackend() emb = _inject_engine(backend) denumiri = ["SCHIMB ULEI MOTOR", "INLOCUIT PLACUTE FRANA", "VERIFICARE DIRECTIE"] _seed_silver(conn, [(d, "OE-1", 0) for d in denumiri]) from app.mapping import ensure_embeddings_corpus ensure_embeddings_corpus(conn) assert backend.total_texts == 3 from app.embedding_cache import load_cached_vectors, text_hash cached = load_cached_vectors(conn, emb.FASTEMBED_MODEL, [text_hash(d) for d in denumiri]) assert len(cached) == 3 def test_warm_start_al_doilea_proces_zero_texte_embed(conn): denumiri = ["SCHIMB ULEI MOTOR", "INLOCUIT PLACUTE FRANA"] _seed_silver(conn, [(d, "OE-1", 0) for d in denumiri]) from app.mapping import ensure_embeddings_corpus backend1 = CountingMockBackend() _inject_engine(backend1) ensure_embeddings_corpus(conn) assert backend1.total_texts == 2 # Simuleaza un al doilea proces: engine NOU, acelasi conn (cache-ul e in DB, nu in RAM). backend2 = CountingMockBackend() emb2 = _inject_engine(backend2) ensure_embeddings_corpus(conn) assert backend2.total_texts == 0 assert emb2.has_corpus() res = emb2.suggest_nearest("SCHIMB ULEI MOTOR", top_k=1) assert res and res[0]["cod"] == "OE-1" def test_incremental_un_rand_nou_trimite_exact_un_text(conn): from app.mapping import ensure_embeddings_corpus _seed_silver(conn, [("SCHIMB ULEI MOTOR", "OE-3", 0)]) _inject_engine(CountingMockBackend()) ensure_embeddings_corpus(conn) _seed_silver(conn, [("INLOCUIT BATERIE", "OE-1", 0)]) backend2 = CountingMockBackend() _inject_engine(backend2) ensure_embeddings_corpus(conn) assert backend2.total_texts == 1 assert backend2.calls == [["INLOCUIT BATERIE"]] # --------------------------------------------------------------------------- # # Model schimbat (US-004) # # --------------------------------------------------------------------------- # def test_model_schimbat_zero_hituri_cache_reindexare_completa_si_purjare(conn, monkeypatch): import app.embeddings as emb_module from app.mapping import ensure_embeddings_corpus denumiri = ["SCHIMB ULEI MOTOR", "INLOCUIT PLACUTE FRANA"] _seed_silver(conn, [(d, "OE-1", 0) for d in denumiri]) model_vechi = emb_module.FASTEMBED_MODEL backend_old = CountingMockBackend() _inject_engine(backend_old) ensure_embeddings_corpus(conn) assert backend_old.total_texts == 2 monkeypatch.setattr(emb_module, "FASTEMBED_MODEL", "model-nou-v2") backend_new = CountingMockBackend() _inject_engine(backend_new) ensure_embeddings_corpus(conn) assert backend_new.total_texts == 2 # zero hit-uri sub noul model -> re-vectorizare integrala from app.embedding_cache import load_cached_vectors, text_hash hashes = [text_hash(d) for d in denumiri] # Intrarile modelului vechi au fost purjate la reindexarea reusita sub noul model (A3). assert load_cached_vectors(conn, model_vechi, hashes) == {} assert len(load_cached_vectors(conn, "model-nou-v2", hashes)) == 2 # --------------------------------------------------------------------------- # # Hash pe lista FILTRATA (denumire goala exclusa din corpus) # # --------------------------------------------------------------------------- # def test_denumire_goala_nu_strica_alinierea_si_nu_produce_miss_permanent(conn): from app.mapping import ensure_embeddings_corpus _seed_silver(conn, [ ("", "OE-9", 0), # denumire_normalizata goala -- exclusa din corpus la filtrare ("SCHIMB ULEI MOTOR", "OE-3", 0), ("INLOCUIT PLACUTE FRANA", "OE-1", 0), ]) backend1 = CountingMockBackend() _inject_engine(backend1) ensure_embeddings_corpus(conn) assert backend1.total_texts == 2 # doar cele 2 randuri cu denumire nevida backend2 = CountingMockBackend() emb2 = _inject_engine(backend2) ensure_embeddings_corpus(conn) assert backend2.total_texts == 0 # warm: fara miss permanent din cauza filtrarii assert emb2.suggest_nearest("SCHIMB ULEI MOTOR", top_k=1)[0]["cod"] == "OE-3" # --------------------------------------------------------------------------- # # Concurenta warmup/request sub lock de modul (A9) # # --------------------------------------------------------------------------- # def test_concurenta_doua_threaduri_fara_dubla_vectorizare(conn): class ReentrancyDetectingBackend: """Detecteaza executie concurenta reala in embed(): daca lock-ul de modul NU serializeaza secventa hash->embed->save->purge, `max_active` ar depasi 1.""" def __init__(self): self.active = 0 self.max_active = 0 self.total_calls = 0 self._guard = threading.Lock() def embed(self, texts): with self._guard: self.active += 1 self.max_active = max(self.max_active, self.active) self.total_calls += 1 time.sleep(0.05) # largeste deliberat fereastra de suprapunere with self._guard: self.active -= 1 return [_det_vector(t) for t in texts] backend = ReentrancyDetectingBackend() emb = _inject_engine(backend) denumiri = ["SCHIMB ULEI MOTOR", "INLOCUIT PLACUTE FRANA"] _seed_silver(conn, [(d, "OE-1", 0) for d in denumiri]) from app.db import get_connection from app.mapping import ensure_embeddings_corpus errors: list[Exception] = [] def _run(): try: c = get_connection() try: ensure_embeddings_corpus(c, block=True) finally: c.close() except Exception as exc: # pragma: no cover - vizibil doar la regresie errors.append(exc) threads = [threading.Thread(target=_run) for _ in range(2)] for t in threads: t.start() for t in threads: t.join(timeout=5) assert not errors assert backend.max_active <= 1 # lock-ul de modul serializeaza cele doua treceri from app.embedding_cache import load_cached_vectors, text_hash cached = load_cached_vectors(conn, emb.FASTEMBED_MODEL, [text_hash(d) for d in denumiri]) assert len(cached) == 2 # niciun rand nou nu a fost purjat fals de trecerea concurenta # --------------------------------------------------------------------------- # # Echivalenta ranking cu toleranta float32 (T6/A5) # # --------------------------------------------------------------------------- # def test_block_false_nu_asteapta_dupa_warmup_in_curs(conn): """Calea de request (block=False) nu trebuie sa blocheze cat warmup-ul (block=True) tine lock-ul -- trebuie sa iasa imediat, nu sa astepte pana termina warmup-ul.""" warmup_poate_continua = threading.Event() warmup_a_intrat_in_embed = threading.Event() class SlowBackend: def embed(self, texts): warmup_a_intrat_in_embed.set() warmup_poate_continua.wait(timeout=5) return [_det_vector(t) for t in texts] emb = _inject_engine(SlowBackend()) _seed_silver(conn, [("SCHIMB ULEI MOTOR", "OE-1", 0)]) from app.db import get_connection from app.mapping import ensure_embeddings_corpus warmup_thread = threading.Thread( target=lambda: ensure_embeddings_corpus(get_connection(), block=True) ) warmup_thread.start() assert warmup_a_intrat_in_embed.wait(timeout=5), "warmup trebuia sa ajunga in embed()" t0 = time.monotonic() ensure_embeddings_corpus(conn, block=False) # nu trebuie sa astepte lock-ul durata_request = time.monotonic() - t0 assert durata_request < 1.0, "block=False nu are voie sa astepte warmup-ul in curs" assert not emb.has_corpus() # warmup-ul nu a terminat inca, request-ul a iesit fara sa faca nimic warmup_poate_continua.set() warmup_thread.join(timeout=5) assert emb.has_corpus() # warmup-ul a terminat normal, neblocat de request def test_indexare_esuata_nu_purjeaza_cache_ul_vechi(conn): """Cand embed-ul unui text NOU esueaza, indexarea nu se termina cu succes -> purge_stale nu trebuie sa ruleze, altfel randuri inca valide (disparute doar din setul CERUT, nu esecul lor) ar fi sterse fals din cache.""" denumiri_initiale = ["SCHIMB ULEI MOTOR", "INLOCUIT PLACUTE FRANA"] _seed_silver(conn, [(d, "OE-1", 0) for d in denumiri_initiale]) _inject_engine(CountingMockBackend()) from app.mapping import ensure_embeddings_corpus ensure_embeddings_corpus(conn) from app.embedding_cache import load_cached_vectors, text_hash hashes_initiale = [text_hash(d) for d in denumiri_initiale] assert len(load_cached_vectors(conn, "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2", hashes_initiale)) == 2 # Corpusul cerut se schimba: "INLOCUIT PLACUTE FRANA" dispare, apare un text nou # a carui vectorizare va esua -- indexarea intreaga trebuie sa rateze. conn.execute( "DELETE FROM mapping_suggestions WHERE denumire_normalizata=?", ("INLOCUIT PLACUTE FRANA",), ) conn.commit() _seed_silver(conn, [("TEXT NOU CARE ESUEAZA", "OE-2", 0)]) class BrokenOnNewText: def embed(self, texts): raise RuntimeError("model indisponibil pentru text nou") emb = _inject_engine(BrokenOnNewText()) ensure_embeddings_corpus(conn) # esueaza intern, prins de degradarea gratioasa # Randul disparut din corpusul cerut RAMANE in cache -- purge nu a rulat. out = load_cached_vectors(conn, emb.FASTEMBED_MODEL, hashes_initiale) assert len(out) == 2 def test_ranking_echivalent_index_direct_vs_din_cache_float32(conn): from app.embedding_cache import sync_corpus_vectors from app.embeddings import EmbeddingEngine corpus = [ {"denumire": "SCHIMB ULEI MOTOR", "cod": "OE-3"}, {"denumire": "REPARATIE CUTIE VITEZE", "cod": "OE-1"}, {"denumire": "VERIFICARE DIRECTIE VOLAN", "cod": "OE-4"}, {"denumire": "INLOCUIT PLACUTE FRANA", "cod": "OE-2"}, ] texts = [item["denumire"] for item in corpus] backend = CountingMockBackend() engine_direct = EmbeddingEngine(backend=backend) engine_direct.index_corpus(corpus) # embed complet, vectori float64 in RAM vectors_cache = sync_corpus_vectors(conn, "model-test-ranking", texts, backend.embed) engine_cache = EmbeddingEngine(backend=backend) engine_cache.index_corpus(corpus, vectors=vectors_cache) # round-trip float32 din cache for query in ("SCHIMB ULEI", "VERIFICARE DIRECTIE"): r_direct = engine_direct.suggest_nearest(query, top_k=len(corpus)) r_cache = engine_cache.suggest_nearest(query, top_k=len(corpus)) assert [r["cod"] for r in r_direct] == [r["cod"] for r in r_cache]