rar-autopass/tests/test_embedding_cache.py

"""Teste pentru app/embedding_cache.py -- cache persistent de vectori in SQLite.

Acopera: round-trip serializare, blob corupt = miss, validare dimensiune la scriere,
chunking >500 randuri, save partial esuat, purge orfane + modele vechi, purge NU
ruleaza cand embed_fn esueaza.
"""
from __future__ import annotations

import os
import sqlite3
import tempfile
from array import array

import pytest

from app.embedding_cache import (
    EMB_DIM,
    blob_to_vector,
    load_cached_vectors,
    purge_stale,
    save_vectors,
    sync_corpus_vectors,
    text_hash,
    vector_to_blob,
)


@pytest.fixture()
def conn(monkeypatch):
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "embcache.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, get_connection
    init_db()
    c = get_connection()
    yield c
    c.close()
    get_settings.cache_clear()


def _vec(seed: float = 1.0, dim: int = EMB_DIM) -> list[float]:
    return [seed + i * 0.001 for i in range(dim)]


# --------------------------------------------------------------------------- #
# Serializare                                                                  #
# --------------------------------------------------------------------------- #

def test_roundtrip_vector_to_blob_blob_to_vector():
    v = _vec(3.5)
    blob = vector_to_blob(v)
    assert isinstance(blob, bytes)
    assert len(blob) == EMB_DIM * 4
    out = blob_to_vector(blob)
    assert isinstance(out, array)
    assert out == pytest.approx(v, rel=1e-6)


def test_text_hash_deterministic_and_distinct():
    assert text_hash("SCHIMB ULEI") == text_hash("SCHIMB ULEI")
    assert text_hash("SCHIMB ULEI") != text_hash("SCHIMB FILTRU")


# --------------------------------------------------------------------------- #
# load_cached_vectors                                                         #
# --------------------------------------------------------------------------- #

def test_load_cached_vectors_roundtrip(conn):
    h = text_hash("SCHIMB ULEI")
    save_vectors(conn, "model-a", [(h, _vec(1.0))])
    out = load_cached_vectors(conn, "model-a", [h])
    assert h in out
    assert out[h] == pytest.approx(_vec(1.0), rel=1e-6)


def test_load_cached_vectors_miss_on_missing_hash(conn):
    out = load_cached_vectors(conn, "model-a", [text_hash("NECUNOSCUT")])
    assert out == {}


def test_load_cached_vectors_blob_corupt_e_miss(conn):
    h = text_hash("SCHIMB ULEI")
    conn.execute("BEGIN")
    conn.execute(
        "INSERT INTO embedding_cache (text_hash, model, vector) VALUES (?, ?, ?)",
        (h, "model-a", b"\x00\x01\x02"),  # lungime gresita
    )
    conn.execute("COMMIT")
    out = load_cached_vectors(conn, "model-a", [h])
    assert h not in out


def test_load_cached_vectors_scoped_pe_model(conn):
    h = text_hash("SCHIMB ULEI")
    save_vectors(conn, "model-a", [(h, _vec(1.0))])
    out = load_cached_vectors(conn, "model-b", [h])
    assert out == {}


# --------------------------------------------------------------------------- #
# save_vectors: validare dimensiune + chunking                                #
# --------------------------------------------------------------------------- #

def test_save_vectors_respinge_dimensiune_gresita(conn):
    h = text_hash("SCHIMB ULEI")
    save_vectors(conn, "model-a", [(h, [1.0, 2.0, 3.0])])  # nu e EMB_DIM
    out = load_cached_vectors(conn, "model-a", [h])
    assert h not in out


def test_save_vectors_chunking_peste_500_randuri(conn):
    items = [(text_hash(f"text-{i}"), _vec(float(i))) for i in range(1200)]
    save_vectors(conn, "model-a", items)
    hashes = [h for h, _ in items]
    out = load_cached_vectors(conn, "model-a", hashes)
    assert len(out) == 1200
    for h, vec in items:
        assert out[h] == pytest.approx(vec, rel=1e-6)


def test_save_vectors_insert_or_replace_idempotent(conn):
    h = text_hash("SCHIMB ULEI")
    save_vectors(conn, "model-a", [(h, _vec(1.0))])
    save_vectors(conn, "model-a", [(h, _vec(2.0))])  # populare intrerupta -> completare
    row = conn.execute(
        "SELECT COUNT(*) AS n FROM embedding_cache WHERE model=? AND text_hash=?",
        ("model-a", h),
    ).fetchone()
    assert row["n"] == 1
    out = load_cached_vectors(conn, "model-a", [h])
    assert out[h] == pytest.approx(_vec(2.0), rel=1e-6)


# --------------------------------------------------------------------------- #
# purge_stale                                                                 #
# --------------------------------------------------------------------------- #

def test_purge_stale_sterge_orfane_model_curent(conn):
    h1, h2 = text_hash("A"), text_hash("B")
    save_vectors(conn, "model-a", [(h1, _vec(1.0)), (h2, _vec(2.0))])
    purge_stale(conn, "model-a", corpus_hashes=[h1])  # h2 nu mai e in corpus
    out = load_cached_vectors(conn, "model-a", [h1, h2])
    assert h1 in out
    assert h2 not in out


def test_purge_stale_sterge_modele_vechi(conn):
    h = text_hash("A")
    save_vectors(conn, "model-old", [(h, _vec(1.0))])
    save_vectors(conn, "model-a", [(h, _vec(2.0))])
    purge_stale(conn, "model-a", corpus_hashes=[h])
    assert load_cached_vectors(conn, "model-old", [h]) == {}
    assert h in load_cached_vectors(conn, "model-a", [h])


# --------------------------------------------------------------------------- #
# sync_corpus_vectors                                                         #
# --------------------------------------------------------------------------- #

def test_sync_corpus_vectors_cold_start(conn):
    calls = []

    def embed_fn(texts):
        calls.append(list(texts))
        return [_vec(float(i)) for i in range(len(texts))]

    texts = ["A", "B", "C"]
    vecs = sync_corpus_vectors(conn, "model-a", texts, embed_fn)
    assert len(vecs) == 3
    assert calls == [texts]  # toate 3 lipsesc din cache -> toate trimise la embed
    hashes = [text_hash(t) for t in texts]
    cached = load_cached_vectors(conn, "model-a", hashes)
    assert len(cached) == 3


def test_sync_corpus_vectors_warm_start_zero_embed_calls(conn):
    calls = []

    def embed_fn(texts):
        calls.append(list(texts))
        return [_vec(float(i)) for i in range(len(texts))]

    texts = ["A", "B", "C"]
    sync_corpus_vectors(conn, "model-a", texts, embed_fn)
    calls.clear()
    vecs2 = sync_corpus_vectors(conn, "model-a", texts, embed_fn)
    assert calls == []  # nimic nou de vectorizat
    assert len(vecs2) == 3


def test_sync_corpus_vectors_incremental_un_text_nou(conn):
    calls = []

    def embed_fn(texts):
        calls.append(list(texts))
        return [_vec(float(i)) for i in range(len(texts))]

    sync_corpus_vectors(conn, "model-a", ["A", "B"], embed_fn)
    calls.clear()
    sync_corpus_vectors(conn, "model-a", ["A", "B", "C"], embed_fn)
    assert calls == [["C"]]


def test_sync_corpus_vectors_nu_purjeaza_singur_apelantul_decide(conn):
    """sync_corpus_vectors NU mai purjeaza -- e responsabilitatea apelantului, DUPA
    o indexare reusita (vezi ensure_embeddings_corpus). purge_stale ramane apelabil
    separat, explicit, de catre apelant."""
    def embed_fn(texts):
        return [_vec(float(i)) for i in range(len(texts))]

    sync_corpus_vectors(conn, "model-a", ["A", "B"], embed_fn)
    sync_corpus_vectors(conn, "model-a", ["A"], embed_fn)  # B disparut din corpus solicitat
    out = load_cached_vectors(conn, "model-a", [text_hash("A"), text_hash("B")])
    assert text_hash("A") in out
    assert text_hash("B") in out  # nepurjat automat -- sync_corpus_vectors nu mai face asta

    purge_stale(conn, "model-a", {text_hash("A")})  # apelantul purjeaza dupa indexare reusita
    out2 = load_cached_vectors(conn, "model-a", [text_hash("A"), text_hash("B")])
    assert text_hash("A") in out2
    assert text_hash("B") not in out2


def test_sync_corpus_vectors_nu_purjeaza_cand_embed_fn_esueaza(conn):
    def embed_fn_ok(texts):
        return [_vec(float(i)) for i in range(len(texts))]

    sync_corpus_vectors(conn, "model-a", ["A", "B"], embed_fn_ok)

    def embed_fn_broken(texts):
        raise RuntimeError("model indisponibil")

    with pytest.raises(RuntimeError):
        sync_corpus_vectors(conn, "model-a", ["A", "C"], embed_fn_broken)

    # "B" nu a fost purjat -- sync_corpus_vectors nu purjeaza niciodata singur.
    out = load_cached_vectors(conn, "model-a", [text_hash("A"), text_hash("B")])
    assert text_hash("A") in out
    assert text_hash("B") in out


class _LockingConn:
    """Wrapper peste o conexiune reala: simuleaza `database is locked` la BEGIN
    (exercita try/except-ul din save_vectors, nu il ocoleste)."""

    def __init__(self, real):
        self._real = real

    def execute(self, sql, *a, **kw):
        if sql.strip() == "BEGIN":
            raise sqlite3.OperationalError("database is locked")
        return self._real.execute(sql, *a, **kw)

    def executemany(self, *a, **kw):
        return self._real.executemany(*a, **kw)

    def __getattr__(self, name):
        return getattr(self._real, name)


def test_sync_corpus_vectors_save_partial_esuat_continua_din_ram(conn):
    """Daca save_vectors esueaza (ex. DB locked la BEGIN), vectorii noi tot se
    intorc din RAM -- indexarea continua, doar persistarea in cache rateaza."""
    locking = _LockingConn(conn)

    def embed_fn(texts):
        return [_vec(float(i)) for i in range(len(texts))]

    vecs = sync_corpus_vectors(locking, "model-a", ["A", "B"], embed_fn)
    assert len(vecs) == 2
    assert vecs[0] == pytest.approx(_vec(0.0), rel=1e-6)

    # Nimic nu a fost persistat (BEGIN a esuat la fiecare chunk).
    out = load_cached_vectors(conn, "model-a", [text_hash("A"), text_hash("B")])
    assert out == {}