Files
rag-from-scratch/tests/chunks/test_packer.py
T

119 lines
3.9 KiB
Python

import pytest
from chunks.models import Block
from chunks.config import ChunkerConfig
from chunks.packer import pack
def make_block(idx: int, content: str, kind: str = "paragraph", atomic: bool = False,
header_path=None) -> Block:
hp = header_path or [{"level": 1, "text": "Titolo"}]
return Block(
id=f"blk_{idx:04d}", kind=kind, content=content, plain_text=content,
atomic=atomic, start_line=idx * 2, end_line=idx * 2 + 1,
header_path=hp, chars=len(content),
)
@pytest.fixture
def cfg():
return ChunkerConfig(min_chars=10, target_chars=50, max_chars=100)
def test_single_block_becomes_single_chunk(cfg):
blocks = [make_block(1, "Testo breve.")]
chunks = pack(blocks, cfg, "test")
assert len(chunks) == 1
assert chunks[0].content_original == "Testo breve."
def test_chunk_id_format(cfg):
blocks = [make_block(1, "Testo.")]
chunks = pack(blocks, cfg, "test")
assert chunks[0].chunk_id.startswith("chk_")
def test_neighbors_populated(cfg):
blocks = [make_block(i, "x" * 60, header_path=[{"level": 1, "text": "T"}]) for i in range(1, 4)]
chunks = pack(blocks, cfg, "test")
assert len(chunks) >= 2
assert chunks[0].neighbors["next_chunk_id"] == chunks[1].chunk_id
assert chunks[1].neighbors["previous_chunk_id"] == chunks[0].chunk_id
assert chunks[-1].neighbors["next_chunk_id"] is None
def test_blocks_merged_until_target(cfg):
blocks = [make_block(i, "x" * 20) for i in range(1, 4)]
chunks = pack(blocks, cfg, "test")
assert len(chunks) >= 2
def test_oversized_paragraph_split(cfg):
long_content = ("Questa è la prima frase completa. " * 5).strip()
blocks = [make_block(1, long_content)]
chunks = pack(blocks, cfg, "test")
for c in chunks:
assert c.chars <= cfg.max_chars or c.flags["is_overflow"]
def test_atomic_block_not_split(cfg):
atomic_content = "```python\n" + "codice\n" * 20 + "```"
blocks = [make_block(1, atomic_content, kind="code", atomic=True)]
chunks = pack(blocks, cfg, "test")
assert len(chunks) == 1
assert chunks[0].flags["is_overflow"] is True
assert chunks[0].content_type == "overflow"
def test_heading_path_break_creates_new_chunk(cfg):
hp1 = [{"level": 1, "text": "Sezione 1"}]
hp2 = [{"level": 1, "text": "Sezione 2"}]
blocks = [make_block(1, "x" * 30, header_path=hp1), make_block(2, "x" * 30, header_path=hp2)]
chunks = pack(blocks, cfg, "test")
assert len(chunks) == 2
assert chunks[0].header_path == hp1
assert chunks[1].header_path == hp2
def test_content_for_embedding_has_header_prefix(cfg):
blocks = [make_block(1, "Contenuto.", header_path=[{"level": 1, "text": "Guida"},
{"level": 2, "text": "Intro"}])]
chunks = pack(blocks, cfg, "test")
assert "Guida" in chunks[0].content_for_embedding
assert "Intro" in chunks[0].content_for_embedding
assert "Contenuto." in chunks[0].content_for_embedding
def test_thematic_break_flushes_chunk(cfg):
blocks = [
make_block(1, "x" * 30),
make_block(2, "---", kind="thematic_break"),
make_block(3, "x" * 30),
]
chunks = pack(blocks, cfg, "test")
assert len(chunks) == 2
def test_flags_has_code(cfg):
blocks = [make_block(1, "```\ncode\n```", kind="code", atomic=True)]
chunks = pack(blocks, cfg, "test")
assert chunks[0].flags["has_code"] is True
def test_flags_has_table(cfg):
blocks = [make_block(1, "| A | B |\n|---|---|\n| 1 | 2 |", kind="table", atomic=True)]
chunks = pack(blocks, cfg, "test")
assert chunks[0].flags["has_table"] is True
def test_chunk_index_sequential(cfg):
blocks = [make_block(i, "x" * 60) for i in range(1, 5)]
chunks = pack(blocks, cfg, "test")
for i, c in enumerate(chunks):
assert c.chunk_index == i
def test_assets_empty(cfg):
blocks = [make_block(1, "Testo.")]
chunks = pack(blocks, cfg, "test")
assert chunks[0].assets == []