Files
davide ed42e1ba7b feat(chunks): aggiunge validator.py — invarianti + metriche
Implementa validate() che controlla chunk_id duplicati, fence rotti,
size compliance (esclusi overflow) e calcola metriche aggregate.
Fix fixture cfg: min_chars=40 per allinearla ai chunk da 50 chars nel test.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-09 08:02:35 +02:00

91 lines
3.1 KiB
Python

import pytest
from chunks.models import Chunk, ChunkingResult, Diagnostics
from chunks.config import ChunkerConfig
from chunks.validator import validate
def make_chunk(idx: int, content: str, chars: int = None,
is_overflow: bool = False, content_type: str = "section_fragment") -> Chunk:
chars = chars or len(content)
return Chunk(
chunk_id=f"chk_{idx:06d}", chunk_index=idx,
content_original=content, content_for_embedding=content,
content_type=content_type, chars=chars,
start_line=idx * 3, end_line=idx * 3 + 2,
header_path=[{"level": 1, "text": "T"}],
block_ids=[f"blk_{idx:04d}"],
flags={"has_code": False, "has_table": False, "has_math": False,
"is_overflow": is_overflow, "is_sparse": False},
neighbors={"previous_chunk_id": None, "next_chunk_id": None},
assets=[],
)
@pytest.fixture
def cfg():
return ChunkerConfig(max_chars=200, min_chars=40, fail_on_content_loss=False)
def test_no_errors_on_valid_chunks(cfg):
result = ChunkingResult(
stem="test", source_path="sources/test.md",
chunks=[make_chunk(1, "Testo valido.")],
diagnostics=Diagnostics([], [], {}),
)
diag = validate(result, "# T\n\nTesto valido.\n", cfg)
assert diag.errors == []
def test_broken_fence_detected(cfg):
result = ChunkingResult(
stem="test", source_path="sources/test.md",
chunks=[make_chunk(1, "```python\ncodice senza chiusura")],
diagnostics=Diagnostics([], [], {}),
)
diag = validate(result, "", cfg)
assert any("fence" in e.lower() for e in diag.errors)
def test_no_duplicate_chunk_ids(cfg):
c1 = make_chunk(1, "Testo A.")
c2 = make_chunk(1, "Testo B.") # stesso chunk_id
result = ChunkingResult(
stem="test", source_path="sources/test.md",
chunks=[c1, c2], diagnostics=Diagnostics([], [], {}),
)
diag = validate(result, "", cfg)
assert any("duplicat" in e.lower() for e in diag.errors)
def test_metrics_populated(cfg):
chunks = [make_chunk(i, "x" * 50) for i in range(1, 4)]
result = ChunkingResult(
stem="test", source_path="sources/test.md",
chunks=chunks, diagnostics=Diagnostics([], [], {}),
)
diag = validate(result, "", cfg)
assert diag.metrics["total_chunks"] == 3
assert diag.metrics["avg_chars"] == 50
assert diag.metrics["size_compliance"] == 1.0
def test_overflow_not_counted_in_size_compliance(cfg):
chunks = [make_chunk(1, "x" * 300, is_overflow=True, content_type="overflow")]
result = ChunkingResult(
stem="test", source_path="sources/test.md",
chunks=chunks, diagnostics=Diagnostics([], [], {}),
)
diag = validate(result, "", cfg)
assert diag.metrics["overflow_count"] == 1
assert diag.errors == []
def test_empty_chunks_list(cfg):
result = ChunkingResult(
stem="test", source_path="sources/test.md",
chunks=[], diagnostics=Diagnostics([], [], {}),
)
diag = validate(result, "", cfg)
assert diag.warnings
assert diag.metrics["total_chunks"] == 0