test: юнит-суит L1/L2/L3 + ГОСТ (41 тест) и гейт в CI перед деплоем

Первый настоящий автоматический тест-суит проекта — раньше регрессии ловились
руками. Покрыта чистая логика детекции и форматирования (без БД/Redis/GPU/Ollama):

- worker-indexer: L1 Winnowing (точные совпадения, идемпотентность отпечатка,
  диапазон signed int64) и L2 MinHash LSH (шинглы, Jaccard, near-duplicate +
  upsert через in-memory-фолбэк).
- worker-gpu: L3 FAISS — возврат doc_id из PostgreSQL (IndexIDMap2),
  идемпотентность add_vectors (remove-before-add, без дублей), self-match ≈ 1,
  ранжирование. Прямо стережёт баги, из-за которых индекс переписывался.
- worker-gost: ГОСТ 7.1-2003 и 7.0.5-2008 — авторы (≤3 / 4+ «и др.»/et al.),
  статья/книга/web, DOI, порядок сортировки кириллица→латиница, стр. в ссылке.

Обвязка: per-service pytest.ini/conftest/requirements-test. scripts/run_tests.sh
гоняет тесты в изолированных python:3.11-slim контейнерах (не засоряя хост),
через Tsinghua-зеркало. CI: job `test` теперь гейтит `deploy` (needs: test) —
падение тестов блокирует прод-деплой. make test / make test-one SVC=...

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jze9
2026-08-11 17:11:08 +05:00
parent a2d3625f2a
commit ecc10a4413
17 changed files with 480 additions and 4 deletions

View File

@@ -0,0 +1,88 @@
"""Юнит-тесты FAISSManager (уровень 3 — семантический индекс).
Проверяют то, что было сломано в старой реализации и переписано:
- возврат doc_id из PostgreSQL (IndexIDMap2), а не позиционного номера;
- идемпотентность add_vectors по doc_id (remove-before-add, без дублей);
- корректность self-match (косинус ≈ 1) и ранжирования.
Работают целиком в памяти — диск (FAISS_INDEX_PATH) не трогается.
"""
import numpy as np
import pytest
pytest.importorskip("faiss")
from app.config import settings # noqa: E402
from app.faiss_manager import FAISSManager # noqa: E402
DIM = settings.EMBED_DIM
def _unit(vecs: np.ndarray) -> np.ndarray:
"""Нормировать строки к единичной длине (для косинуса через inner product)."""
vecs = np.asarray(vecs, dtype=np.float32)
norms = np.linalg.norm(vecs, axis=1, keepdims=True)
norms[norms == 0] = 1.0
return vecs / norms
@pytest.fixture(autouse=True)
def fresh_index():
"""Свежий пустой индекс в памяти на каждый тест; диск не читаем и не пишем."""
FAISSManager._index = FAISSManager._new_index()
yield
FAISSManager._index = None
def test_empty_search_returns_nothing():
assert FAISSManager.search(np.zeros(DIM, dtype=np.float32)) == []
assert FAISSManager.total_vectors() == 0
def test_add_and_self_search_returns_postgres_doc_id():
rng = np.random.default_rng(42)
vecs = _unit(rng.standard_normal((3, DIM)))
FAISSManager.add_vectors(vecs, [101, 202, 303])
assert FAISSManager.total_vectors() == 3
results = FAISSManager.search(vecs[1], k=1)
assert results, "поиск ничего не вернул"
top_id, score = results[0]
assert top_id == 202 # IndexIDMap2 возвращает doc_id, а не позицию
assert score == pytest.approx(1.0, abs=1e-4) # self-match: косинус ≈ 1
def test_add_is_idempotent_by_doc_id():
rng = np.random.default_rng(0)
FAISSManager.add_vectors(_unit(rng.standard_normal((2, DIM))), [1, 2])
# Повторный эмбеддинг тех же id не должен плодить дубли (remove-before-add)
updated = _unit(rng.standard_normal((2, DIM)))
FAISSManager.add_vectors(updated, [1, 2])
assert FAISSManager.total_vectors() == 2
# ...и поиск возвращает ОБНОВЛЁННЫЙ вектор
top_id, score = FAISSManager.search(updated[0], k=1)[0]
assert top_id == 1
assert score == pytest.approx(1.0, abs=1e-4)
def test_search_ranks_nearest_first():
rng = np.random.default_rng(7)
v = _unit(rng.standard_normal((1, DIM)))[0]
near = _unit((v + 0.01 * rng.standard_normal(DIM))[None, :])[0]
far = _unit(rng.standard_normal((1, DIM)))[0]
FAISSManager.add_vectors(np.stack([near, far]), [10, 20])
results = FAISSManager.search(v, k=2)
assert [doc_id for doc_id, _ in results][0] == 10 # ближайший — near
def test_search_respects_k_and_index_size():
rng = np.random.default_rng(1)
FAISSManager.add_vectors(_unit(rng.standard_normal((2, DIM))), [5, 6])
# k больше числа векторов не должно приводить к падению или id == -1
results = FAISSManager.search(_unit(rng.standard_normal((1, DIM)))[0], k=100)
assert len(results) == 2
assert all(doc_id in (5, 6) for doc_id, _ in results)