Testing — Test Suites, Benchmarks & Quality Assurance

1. Test suite organization

Tests live in backend/tests/ and are organized by system phase:

backend/tests/
├── test_phase1_boot.py           # System startup, config, logging
├── test_phase2_all_types.py      # Data model validation (Pydantic)
├── test_phase3_chunking.py       # Document chunking logic
├── test_phase4_embeddings.py     # BGE-M3 embedding generation
├── test_phase5_vector_store.py   # ChromaDB operations
├── test_phase678_retrieval.py    # Dense/sparse/hybrid retrieval + reranking
├── test_integration_e2e.py       # End-to-end RAG pipeline
├── test_gpu_manager.py           # GPU resource management (37 tests)
├── test_pipeline_config.py       # Runtime pipeline reconfiguration
├── test_workspace.py             # Workspace management + folder watcher
├── test_context_hint.py          # Context hint retrieval
├── test_citation_validation.py   # Citation integrity checks
├── test_source_parsing.py        # Source citation parsing
├── test_think_splitter.py        # Chain-of-thought extraction
├── test_abstain_and_temp.py      # Abstain signal + temperature control
├── test_post_tool_gate.py        # Post-tool reasoning gate
└── test_api_provider.py          # Cloud API provider integration

Phase-based testing philosophy

Each phase tests a single layer of the stack:

1. Phase 1 (Boot) — Can we start? Config loading, logging, directory creation
2. Phase 2 (Types) — Are data structures valid? Pydantic model validation
3. Phase 3 (Chunking) — Can we split documents? Chunk size, overlap, section respect
4. Phase 4 (Embeddings) — Can we embed text? BGE-M3 inference, normalization
5. Phase 5 (Vector Store) — Can we store/retrieve vectors? ChromaDB operations
6. Phase 6-8 (Retrieval) — Can we search? Dense, sparse, hybrid, reranking
7. Integration — Does the full pipeline work end-to-end?

This makes debugging trivial: a Phase 3 failure means chunking broke, not embeddings.

2. Running tests

Run all tests

bashcd backend
pytest tests/ -v

Run a specific phase

bashcd backend
pytest tests/test_phase3_chunking.py -v

Run a specific test function

bashcd backend
pytest tests/test_phase3_chunking.py::test_chunking_respects_sections -v

Run with coverage

bashcd backend
pytest tests/ --cov=legal_retrieval --cov-report=html

Coverage report → backend/htmlcov/index.html

Run only fast tests (skip slow integration)

bashcd backend
pytest tests/ -v -m "not slow"

Run integration tests only

bashcd backend
pytest tests/test_integration_e2e.py -v

3. Test environment setup

Tests use in-memory ChromaDB and mock embeddings by default (fast, no GPU needed):

python# Typical test setup
import pytest
from legal_retrieval.config import reset_config

@pytest.fixture(autouse=True)
def reset_state():
    """Reset global state before each test."""
    reset_config()
    yield

For tests that need real models (integration), use the gpu marker:

python@pytest.mark.gpu
def test_real_embedding():
    """Requires GPU and real BGE-M3 model."""
    from legal_retrieval.embedder import get_embedder
    embedder = get_embedder()
    vec = embedder.embed("test query")
    assert vec.shape == (1024,)

Run GPU tests separately:

bashcd backend
pytest tests/ -v -m gpu

4. Phase 1: Boot & Configuration

File: test_phase1_boot.py

Tests system startup essentials:

What's tested

✅ Configuration loading

• Default values from config.py
• Environment variable overrides (PLR_*)
• Config singleton behavior

✅ Logging setup

• Log level from PLR_LOG_LEVEL
• File handler creation (logs/legal_retrieval.log)
• Rotating file handler (10MB, 5 backups)

✅ Directory creation

• ensure_dirs() creates all required directories
• Workspace directory creation

✅ Device resolution

• resolve_device("auto") → cuda | mps | cpu
• Respects explicit device override

Example tests

pythondef test_config_loads_defaults():
    """Config loads with sensible defaults."""
    cfg = get_config()
    assert cfg.models.embedding_dim == 1024
    assert cfg.search.top_k_final == 5

def test_logging_creates_file():
    """Logging setup creates log file."""
    setup_logging(force=True)
    log_file = get_config().paths.logs_dir / "legal_retrieval.log"
    assert log_file.exists()

5. Phase 2: Data Model Validation

File: test_phase2_all_types.py

Tests all Pydantic models for structural integrity:

What's tested

✅ Request/Response models (models.py)

• ChatRequest validation
• QueryResponse structure
• StreamEvent types

✅ Document models

• Document (file_name, page, section, text)
• Chunk (chunk_id, metadata)

✅ Configuration dataclasses

• ModelConfig, SearchConfig, ChunkingConfig
• Frozen attributes (immutability)

Example tests

pythondef test_chat_request_validation():
    """ChatRequest rejects invalid input."""
    with pytest.raises(ValidationError):
        ChatRequest(query="", mode="invalid_mode")

def test_chunk_id_generation():
    """Chunk IDs are stable and collision-free."""
    chunk1 = Chunk(file_name="a.pdf", page=1, text="foo")
    chunk2 = Chunk(file_name="a.pdf", page=1, text="foo")
    assert chunk1.chunk_id == chunk2.chunk_id  # stable

6. Phase 3: Chunking

File: test_phase3_chunking.py

Tests the StructureAwareChunker:

What's tested

✅ Section respect

• Chunks never split across section boundaries
• Headers stay with their content

✅ Chunk size targets

• Chunks stay near chunk_size (480 tokens default)
• Minimum chunk size enforced (32 tokens)

✅ Overlap handling

• Overlap tokens correctly copied between chunks
• No duplicate text when overlap=0

✅ Edge cases

• Single-word documents
• Empty sections
• Very long sections (must split eventually)

Example tests

pythondef test_chunking_respects_sections():
    """Chunks never split section boundaries."""
    doc = Document(
        text="# Section A\nContent A.\n\n# Section B\nContent B.",
        file_name="test.md",
        section_title="Main"
    )
    chunks = chunker.chunk_document(doc)
    # Each chunk's text must not span two section headers
    for chunk in chunks:
        assert chunk.text.count("# Section") <= 1

7. Phase 4: Embeddings

File: test_phase4_embeddings.py

Tests BGE-M3 embedding generation:

What's tested

✅ Embedding shape

• Output is 1024-dim for BGE-M3
• L2-normalized (unit vectors)

✅ Batch processing

• Batch embedding produces same results as single
• Batch size respected

✅ Semantic similarity

• Similar texts have high cosine similarity
• Unrelated texts have low similarity

✅ Device handling

• Respects device config (cuda/cpu)
• Falls back gracefully when GPU unavailable

Example tests

pythondef test_embedding_shape():
    """BGE-M3 produces 1024-dim L2-normalized vectors."""
    embedder = get_embedder()
    vec = embedder.embed("test query")
    assert vec.shape == (1024,)
    assert abs(np.linalg.norm(vec) - 1.0) < 1e-5  # L2-normalized

def test_semantic_similarity():
    """Similar texts have high cosine similarity."""
    embedder = get_embedder()
    vec1 = embedder.embed("contract termination")
    vec2 = embedder.embed("agreement cancellation")
    vec3 = embedder.embed("weather forecast")
    
    sim_similar = np.dot(vec1, vec2)  # cosine (normalized vecs)
    sim_different = np.dot(vec1, vec3)
    assert sim_similar > 0.7
    assert sim_different < 0.3

8. Phase 5: Vector Store

File: test_phase5_vector_store.py

Tests ChromaDB operations via VectorStore:

What's tested

✅ Add/retrieve chunks

• add_chunks() stores vectors + metadata
• query() retrieves top-k by cosine similarity

✅ Metadata filtering

• Filter by file_name, page_number, section_title
• Combined filters work correctly

✅ Deduplication

• Upserting same chunk_id updates, doesn't duplicate
• Collection size remains correct

✅ Persistence

• ChromaDB survives restart (PersistentClient)
• Collection metadata persists

Example tests

pythondef test_vector_store_add_and_query():
    """Add chunks, query retrieves them correctly."""
    store = VectorStore()
    chunks = [
        Chunk(file_name="a.pdf", page=1, text="contract law"),
        Chunk(file_name="b.pdf", page=2, text="weather data"),
    ]
    embeddings = embedder.embed_batch([c.text for c in chunks])
    store.add_chunks(chunks, embeddings)
    
    results = store.query("legal agreement", top_k=1)
    assert results[0].file_name == "a.pdf"  # semantic match

9. Phase 6-8: Retrieval & Reranking

File: test_phase678_retrieval.py

Tests the full retrieval pipeline (RetrievalPipeline):

What's tested

✅ Dense retrieval (ChromaDB)

• Returns top-k by cosine similarity
• Respects top_k_retrieval parameter

✅ Sparse retrieval (BM25)

• Keyword matching works
• BM25 scores are reasonable

✅ Hybrid merge strategies

• combine_and_dedup — union of dense + sparse
• rrf — reciprocal rank fusion
• boost_only — dense results, BM25 boosts scores

✅ Reranking (BGE-Reranker-v2-M3)

• Cross-encoder re-scores candidates
• Top-k after rerank differs from pre-rerank
• Rerank scores in [0, 1]

✅ Skip rerank

• skip_rerank=True bypasses reranker
• Falls back to retrieval scores

Example tests

pythondef test_hybrid_retrieval():
    """Hybrid mode merges dense + sparse results."""
    pipeline = RetrievalPipeline()
    response = pipeline.query("rescission contract", top_k=5)
    assert len(response.results) <= 5
    assert all(r.score >= 0 for r in response.results)

def test_rerank_changes_order():
    """Reranker re-orders retrieval results."""
    pipeline = RetrievalPipeline()
    resp_no_rerank = pipeline.query("test", skip_rerank=True)
    resp_rerank = pipeline.query("test", skip_rerank=False)
    
    # Top result may differ after reranking
    assert resp_no_rerank.results[0].chunk_id != resp_rerank.results[0].chunk_id

10. Integration: End-to-End RAG

File: test_integration_e2e.py

Tests the complete RAG flow from query to answer:

What's tested

✅ Full pipeline

• Ingest document → query → retrieve → rerank → answer
• All components integrated

✅ Streaming chat

• SSE events stream correctly
• done event contains answer + sources

✅ Tool execution

• search_documents tool works
• verify_relevance tool works
• ingest_file tool works

✅ Citation generation

• Answer includes [1], [2] markers
• Citations match returned sources

Example tests

python@pytest.mark.slow
def test_end_to_end_rag():
    """Full RAG pipeline: ingest → query → answer."""
    # Ingest test document
    pipeline = RetrievalPipeline()
    pipeline.ingest_file("tests/fixtures/sample.pdf")
    
    # Query
    response = pipeline.query("What is the payment term?", top_k=3)
    assert len(response.results) > 0
    
    # Generate answer (via chat pipeline)
    from legal_retrieval.chat import create_chat_pipeline
    chat = create_chat_pipeline()
    answer = chat.chat("What is the payment term?", mode="rag")
    assert "[1]" in answer  # citation present

11. GPU Manager Tests

File: test_gpu_manager.py (37 tests)

Tests smart GPU resource management (no GPU required — pure logic tests):

What's tested

✅ Registration

• Models register with priority, VRAM estimate, promote/demote functions

✅ Promotion/demotion

• acquire() promotes model to VRAM
• Models demote per residency mode (eager/adaptive/persistent)

✅ Priority eviction

• Lower-priority models evicted first under pressure
• Higher-priority models never evicted for lower ones

✅ Hotness tracking

• Models become "hot" after hot_min_uses within hot_window_seconds
• Old uses expire from sliding window

✅ Residency modes

• eager — demote always
• adaptive — keep hot models, evict cold
• persistent — keep all

✅ Auto-batch tuning

• suggest_batch_size() scales with free VRAM
• Never exceeds configured cap
• CPU fallback keeps configured batch

✅ Ollama eviction

• ollama stop called under VRAM pressure when enabled

Example tests

pythondef test_gpu_manager_priority_eviction():
    """Lower-priority models evicted first."""
    gpu = GPUManager(total_vram_mb=4000)
    gpu.register("embedder", vram_estimate_mb=1500, priority=0)
    gpu.register("reranker", vram_estimate_mb=2000, priority=1)
    
    # Acquire both (total 3500MB fits in 4000MB)
    gpu.acquire("embedder")
    gpu.acquire("reranker")
    assert gpu._models["embedder"].state == "vram"
    assert gpu._models["reranker"].state == "vram"
    
    # Register LLM (5000MB) — reranker evicted, embedder stays
    gpu.register("llm", vram_estimate_mb=5000, priority=2)
    gpu.acquire("llm")
    assert gpu._models["embedder"].state == "vram"  # high priority
    assert gpu._models["reranker"].state == "ram"   # evicted

12. Pipeline Config Tests

File: test_pipeline_config.py

Tests runtime pipeline reconfiguration (POST /api/pipeline/config):

What's tested

✅ Node presence → config derivation

• Rerank node present → skip_rerank=False
• Rerank node absent → skip_rerank=True
• Dense+sparse → search_mode="hybrid"
• Dense only → search_mode="dense"

✅ Validation

• Unknown node types dropped
• Dangling edges dropped
• Duplicate IDs deduplicated

✅ Query overrides

• ?top_k=8 wins over layout-derived value
• ?dense_weight=0.7 overrides fusion weights

Example tests

pythondef test_pipeline_config_derives_skip_rerank():
    """Rerank node absence sets skip_rerank=True."""
    layout = {
        "nodes": [
            {"id": "dense", "type": "denseRetrieval"},
            {"id": "answer", "type": "answer"}
        ],
        "edges": [{"id": "e1", "source": "dense", "target": "answer"}]
    }
    config = apply_pipeline_layout(layout)
    assert config["skip_rerank"] is True

13. Workspace Tests

File: test_workspace.py

Tests workspace management and folder watcher:

What's tested

✅ Workspace lifecycle

• set_workspace() → storage paths resolve under workspace
• reset_workspace() → falls back to project data/

✅ Folder fingerprinting

• Fingerprint captures (size, mtime) per file
• Excludes workspace storage dirs
• Excludes temp/lock files

✅ Auto-ingestion

• Watcher detects new files
• Ingests only new files (skips already-indexed)
• Serializes with manual ingestion via INGEST_LOCK

✅ Watcher status

• status() returns thread-safe snapshot
• Last scan timestamp, files found/indexed, chunks

Example tests

pythondef test_workspace_watcher_auto_ingests_new_file(tmp_path):
    """Watcher auto-ingests newly added documents."""
    set_workspace(tmp_path)
    (tmp_path / "contract.pdf").write_text("fake pdf")
    
    watcher = get_workspace_watcher()
    watcher.start(tmp_path)
    watcher._last_fingerprint = None  # force "changed"
    watcher._scan_once()
    
    status = watcher.status()
    assert "contract.pdf" in status["files_indexed"]
    assert status["chunks_indexed"] > 0

14. Other Test Files

test_context_hint.py

Tests context hint retrieval (native right-click → exact chunk citation):

• Extracts chunk from user-selected text
• Exact match in vector store
• Fallback to semantic search when exact match fails

test_citation_validation.py

Tests citation integrity:

• Citation markers [N] validated against source count
• Malformed markers stripped ([18] when only 5 sources)
• Lexical entailment check (chunk supports claim)

test_source_parsing.py

Tests citation string parsing:

• "[Contract.pdf, p.3]" → {file, page}
• Multiple citations extracted
• Malformed citations handled gracefully

test_think_splitter.py

Tests chain-of-thought extraction:

• <think>...</think> blocks extracted
• Post-tool reasoning segments detected
• Adaptive reasoning window (short segments only)

test_abstain_and_temp.py

Tests abstain signal and temperature control:

• answer_found=false when pipeline abstains
• Temperature parameter respected in LLM calls

test_post_tool_gate.py

Tests post-tool reasoning gate:

• Untagged text after tool result treated as CoT
• Max segments and char limits enforced
• Long segments (answer body) excluded from CoT

test_api_provider.py

Tests cloud API provider integration:

• OpenAI-compatible provider configuration
• API key validation
• Model list fetching
• Embedding/chat API calls

15. Test fixtures

Common test fixtures in tests/conftest.py (or inline):

python@pytest.fixture
def temp_chroma_db(tmp_path):
    """Temporary ChromaDB instance."""
    from legal_retrieval.vector_store import VectorStore
    store = VectorStore(persist_directory=str(tmp_path / "chroma"))
    yield store
    store.client.delete_collection(store.collection_name)

@pytest.fixture
def sample_chunks():
    """Sample chunks for testing."""
    return [
        Chunk(file_name="a.pdf", page=1, text="contract law"),
        Chunk(file_name="b.pdf", page=2, text="weather data"),
    ]

@pytest.fixture(autouse=True)
def reset_config():
    """Reset config singleton before each test."""
    from legal_retrieval.config import reset_config, reset_workspace
    reset_config()
    reset_workspace()
    yield

16. Running tests in CI

GitHub Actions workflow (.github/workflows/ci.yml):

yamlname: CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: |
          cd backend
          pip install -r requirements.txt
          pip install pytest pytest-cov
      - name: Run tests
        run: |
          cd backend
          pytest tests/ -v --cov=legal_retrieval
      - name: Upload coverage
        uses: codecov/codecov-action@v3

17. Test coverage goals

Target coverage by module:

18. Writing new tests

Test naming convention

pythondef test_<component>_<behavior>():
    """<What it tests in one sentence>."""

Examples:

• test_chunker_respects_sections()
• test_embedder_normalizes_vectors()
• test_gpu_manager_evicts_low_priority()

Test structure (Arrange-Act-Assert)

pythondef test_example():
    """Example test following AAA pattern."""
    # Arrange — set up test data
    chunks = [Chunk(...), Chunk(...)]
    
    # Act — perform the operation
    result = chunker.chunk_document(doc)
    
    # Assert — verify the outcome
    assert len(result) == 2
    assert result[0].text.startswith("Chapter")

Use descriptive assertion messages

python# Bad
assert len(results) > 0

# Good
assert len(results) > 0, f"Expected results, got empty list for query: {query}"

Parametrize for multiple cases

python@pytest.mark.parametrize("query,expected_count", [
    ("contract law", 5),
    ("weather", 0),
    ("", 0),
])
def test_retrieval_various_queries(query, expected_count):
    """Test retrieval with various query types."""
    results = pipeline.query(query, top_k=5)
    assert len(results.results) == expected_count

19. Troubleshooting test failures

ChromaDB permission errors

PermissionError: [WinError 32] The process cannot access the file

Cause: ChromaDB client not closed between tests

Fix: Use tmp_path fixture for isolated DBs:

pythondef test_example(tmp_path):
    store = VectorStore(persist_directory=str(tmp_path / "chroma"))
    # ... test logic
    store.client.reset()  # cleanup

CUDA out of memory in tests

RuntimeError: CUDA out of memory

Cause: Real models loaded in tests

Fix: Use mocks or skip GPU tests:

python@pytest.mark.gpu  # mark as GPU-requiring
def test_real_embedding():
    ...

Run without GPU tests:

bashpytest tests/ -v -m "not gpu"

Slow test suite

Cause: Integration tests run on every invocation

Fix: Mark slow tests:

python@pytest.mark.slow
def test_full_rag_pipeline():
    ...

Run fast tests only:

bashpytest tests/ -v -m "not slow"

20. Related documentation