File size: 3,942 Bytes
547ce6e | 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 | """Chunking invariants, checked against the real corpus."""
from __future__ import annotations
import pytest
from app.core.settings import Settings
from app.rag.chunk import build_header, chunk_documents
from app.rag.parse import ParsedDocument
pytestmark = pytest.mark.integration
@pytest.fixture(scope="module")
def chunked(parsed: list[ParsedDocument], settings: Settings):
return chunk_documents(parsed, settings)
def test_no_chunk_exceeds_the_encoder_window(chunked, settings: Settings):
"""The load-bearing property.
Past 512 tokens bge-small truncates its input, so the tail of an over-long chunk
contributes nothing to the vector while still being displayed as retrieved evidence.
"""
chunks, stats = chunked
assert stats.over_window == 0
assert stats.tokens_max <= settings.embedding_max_tokens
assert all(chunk.token_count <= settings.embedding_max_tokens for chunk in chunks)
def test_chunks_never_straddle_an_article(chunked):
"""A chunk spanning two articles could not be cited honestly."""
chunks, _ = chunked
for chunk in chunks:
assert chunk.text.startswith(f"{chunk.law_label} - Article {chunk.article_no}")
def test_every_chunk_carries_its_context_header(chunked):
"""Statutory text never names its own law or article; the header supplies both."""
chunks, _ = chunked
for chunk in chunks:
header, _, body = chunk.text.partition("\n")
assert str(chunk.article_no) in header
assert chunk.law_label in header
assert body.strip()
def test_article_text_is_fully_covered_by_its_chunks(parsed, chunked):
"""No article may lose text to chunking.
Overlap means the concatenation is longer than the original, but every article's
first and last words must appear somewhere in its chunk set.
"""
chunks, _ = chunked
by_article: dict[tuple[str, str, int], list[str]] = {}
for chunk in chunks:
by_article.setdefault((chunk.law_id, chunk.part_id, chunk.article_no), []).append(
chunk.text
)
for document in parsed:
for article in document.articles:
joined = " ".join(by_article[(article.law_id, article.part_id, article.article_no)])
words = article.text.split()
assert words[0] in joined
assert " ".join(words[-4:]) in joined or words[-1] in joined
def test_chunk_ids_are_unique_and_content_addressed(chunked):
chunks, _ = chunked
ids = [chunk.chunk_id for chunk in chunks]
assert len(ids) == len(set(ids))
def test_reindexing_is_deterministic(parsed, settings: Settings):
"""Identical input must produce identical ids, so a rebuild is a no-op."""
first, _ = chunk_documents(parsed, settings)
second, _ = chunk_documents(parsed, settings)
assert [c.chunk_id for c in first] == [c.chunk_id for c in second]
def test_smaller_target_produces_more_chunks(parsed, settings: Settings):
small, small_stats = chunk_documents(parsed, settings, target_tokens=300)
large, _ = chunk_documents(parsed, settings, target_tokens=600)
assert len(small) > len(large)
assert small_stats.tokens_max <= 320
def test_uncapped_chunking_exceeds_the_window(parsed, settings: Settings):
"""The cap is worth something: without it, chunks are silently truncated by bge."""
_, stats = chunk_documents(parsed, settings, target_tokens=1000, enforce_encoder_window=False)
assert stats.over_window > 0
assert stats.tokens_max > settings.embedding_max_tokens
def test_header_prefers_the_article_title_then_the_section(parsed):
tenancy = next(d for d in parsed if d.law_id == "dubai-tenancy-law")
titled = next(a for a in tenancy.articles if a.article_no == 5)
untitled = next(a for a in tenancy.articles if a.article_no == 6)
assert build_header(titled).endswith("Term of Lease Contract")
assert "(Term of Lease Contract)" in build_header(untitled)
|