File size: 4,258 Bytes
b30f068 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """Shared pytest fixtures.
Generalizes the in-memory-Qdrant + local-hashing-embedder trick (previously
hand-wired in the root offline scripts) into reusable fixtures, and provides
temp-SQLite account fixtures plus HF-sync mocks — so the whole default suite
runs with NO credentials, network, or Docker.
"""
from __future__ import annotations
import hashlib
import os
import re
import sqlite3
import sys
from pathlib import Path
import pytest
# Repo root on sys.path + as CWD-independent anchor.
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
CDMS_DB = ROOT / "data" / "cdms_metadata.db"
_DIM = 1536
_TOKEN = re.compile(r"[a-z0-9]+")
# --------------------------------------------------------------------------- #
# Deterministic local embedder (stand-in for OpenAIEmbeddingService)
# --------------------------------------------------------------------------- #
def _embed(text: str):
import numpy as np
v = np.zeros(_DIM, dtype=np.float32)
for tok in _TOKEN.findall((text or "").lower()):
idx = int(hashlib.md5(tok.encode()).hexdigest(), 16) % _DIM
v[idx] += 1.0
n = np.linalg.norm(v)
if n > 0:
v /= n
return v.tolist()
class LocalEmbedder:
"""Drop-in replacement for the OpenAI embedding service (offline, deterministic)."""
def generate_embedding(self, text: str):
return _embed(text)
@pytest.fixture(scope="session")
def local_embedder():
return LocalEmbedder()
# --------------------------------------------------------------------------- #
# Accounts: temp SQLite store + service with HF sync disabled
# --------------------------------------------------------------------------- #
@pytest.fixture()
def account_store(tmp_path):
from src.accounts.store import AccountStore
return AccountStore(tmp_path / "accounts.db")
@pytest.fixture()
def accounts_service(tmp_path, monkeypatch):
"""Real AccountsService on a temp DB with sync OFF (no HF token) and a
known SESSION_SECRET so token tests are deterministic."""
monkeypatch.setenv("SESSION_SECRET", "unit-test-secret")
monkeypatch.delenv("HF_DATA_REPO", raising=False)
monkeypatch.delenv("HF_DATA_TOKEN", raising=False)
from src.accounts.service import AccountsService
svc = AccountsService(db_path=str(tmp_path / "accounts.db"), daily_quota=5)
assert svc.sync.enabled is False # local-dev guard: no sync without a token
return svc
# --------------------------------------------------------------------------- #
# Integration: real CDMS pipeline over the committed index (in-memory Qdrant)
# --------------------------------------------------------------------------- #
@pytest.fixture(scope="session")
def in_memory_rag():
"""Build an in-memory Qdrant from the committed chunk DB and wire the REAL
CDMSRAGSearch to it with the local embedder. Skips if the index or
qdrant-client is unavailable."""
if not CDMS_DB.exists():
pytest.skip(f"CDMS index not present ({CDMS_DB}); skipping integration test")
try:
from src.rag.vector_store import QdrantVectorStore
from src.cdms.rag_search import CDMSRAGSearch
except Exception as e: # missing qdrant-client etc.
pytest.skip(f"RAG deps unavailable: {e}")
conn = sqlite3.connect(str(CDMS_DB))
try:
rows = conn.execute(
"SELECT dc.id, dc.content, dc.page_number, dc.document_id, d.filename "
"FROM document_chunks dc JOIN documents d ON dc.document_id = d.id"
).fetchall()
finally:
conn.close()
if not rows:
pytest.skip("CDMS index has no chunks")
store = QdrantVectorStore() # Docker unavailable -> in-memory mode
for cid, content, page, docid, filename in rows:
if not content:
continue
store.add_document_chunk(
str(cid),
_embed(content),
{
"content": content,
"source_file": filename,
"page_number": page or 0,
"document_id": docid,
},
)
searcher = CDMSRAGSearch()
searcher.vector_store = store
searcher.embedding_service = LocalEmbedder()
return searcher
|