119 lines
4.6 KiB
Python
119 lines
4.6 KiB
Python
import hashlib
|
|
|
|
import pytest
|
|
|
|
from tht.corpus.chunk import ChunkPolicy, chunk
|
|
from tht.corpus.models import CanonicalDocument, CorpusManifest
|
|
|
|
|
|
def document(content: str) -> CanonicalDocument:
|
|
normalized = content.replace("\r\n", "\n").replace("\r", "\n")
|
|
digest = hashlib.sha256(normalized.encode()).hexdigest()
|
|
return CanonicalDocument(
|
|
document_id="doc:abc",
|
|
source_id="source:a",
|
|
source_uri="https://host/a.md",
|
|
source_fingerprint="etag:abc",
|
|
content_hash=f"sha256:{digest}",
|
|
title="A",
|
|
content=normalized,
|
|
media_type="text/markdown",
|
|
pipeline_version="pipe:v1",
|
|
metadata={"owner": "docs"},
|
|
)
|
|
|
|
|
|
def other_document(content: str) -> CanonicalDocument:
|
|
return document(content).model_copy(
|
|
update={
|
|
"document_id": "doc:def",
|
|
"source_id": "source:b",
|
|
"source_uri": "https://host/b.md",
|
|
}
|
|
)
|
|
|
|
|
|
def test_chunk_ids_are_stable_for_same_content_and_repeat_runs():
|
|
policy = ChunkPolicy(version="paragraph:v1", max_chars=8)
|
|
first = chunk(document("A\n\nB"), policy)
|
|
second = chunk(document("A\r\n\r\nB"), policy)
|
|
repeated = chunk(document("A\n\nB"), policy)
|
|
assert first == second == repeated
|
|
|
|
|
|
def test_policy_version_changes_ids_without_changing_boundaries():
|
|
doc = document("alpha\n\nbeta")
|
|
first = chunk(doc, ChunkPolicy(version="paragraph:v1", max_chars=6))
|
|
second = chunk(doc, ChunkPolicy(version="paragraph:v2", max_chars=6))
|
|
assert [item.content for item in first] == [item.content for item in second]
|
|
assert [item.chunk_id for item in first] != [item.chunk_id for item in second]
|
|
|
|
|
|
def test_same_policy_version_with_different_boundary_config_changes_ids():
|
|
doc = document("alpha beta")
|
|
first = chunk(doc, ChunkPolicy(version="paragraph:v1", max_chars=6))
|
|
second = chunk(doc, ChunkPolicy(version="paragraph:v1", max_chars=7))
|
|
assert first[0].chunk_id != second[0].chunk_id
|
|
|
|
|
|
def test_identical_content_in_different_documents_cannot_collide_in_manifest():
|
|
policy = ChunkPolicy(version="paragraph:v1", max_chars=20)
|
|
first = document("same")
|
|
second = other_document("same")
|
|
chunks = [*chunk(first, policy), *chunk(second, policy)]
|
|
manifest = CorpusManifest(
|
|
pipeline_version="pipe:v1", documents=[first, second], chunks=chunks
|
|
)
|
|
assert len({item.chunk_id for item in manifest.chunks}) == 2
|
|
|
|
|
|
def test_long_non_ascii_tokens_are_hard_split_by_unicode_characters():
|
|
chunks = chunk(document("ééééé世界"), ChunkPolicy(version="chars:v1", max_chars=3))
|
|
assert [item.content for item in chunks] == ["ééé", "éé世", "界"]
|
|
assert all(len(item.content) <= 3 for item in chunks)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content",
|
|
[
|
|
"alpha beta\tgamma\n\ndelta",
|
|
"line with markdown hard break \nnext line\n```\na b\n```",
|
|
" \t\n\n \n",
|
|
"supercalifragilisticexpialidocious",
|
|
"é 世界\r\nnext",
|
|
],
|
|
)
|
|
def test_chunks_preserve_every_character_and_respect_max_chars(content):
|
|
doc = document(content)
|
|
chunks = chunk(doc, ChunkPolicy(version="exact:v1", max_chars=9))
|
|
assert "".join(item.content for item in chunks) == doc.content
|
|
assert all(0 < len(item.content) <= 9 for item in chunks)
|
|
|
|
|
|
def test_chunks_have_contiguous_ordinals_hashes_and_provenance_metadata():
|
|
doc = document("alpha beta gamma")
|
|
chunks = chunk(doc, ChunkPolicy(version="words:v1", max_chars=7))
|
|
assert [item.ordinal for item in chunks] == list(range(len(chunks)))
|
|
assert len({item.chunk_id for item in chunks}) == len(chunks)
|
|
for item in chunks:
|
|
assert item.source_uri == doc.source_uri
|
|
assert item.document_id == doc.document_id
|
|
assert item.pipeline_version == doc.pipeline_version
|
|
assert item.metadata["chunk_policy"]["max_chars"] == 7
|
|
assert item.metadata["chunk_policy"]["version"] == "words:v1"
|
|
assert item.metadata["chunk_policy"]["fingerprint"].startswith("sha256:")
|
|
assert item.metadata["document"] == {"owner": "docs"}
|
|
assert item.content_hash == "sha256:" + hashlib.sha256(item.content.encode()).hexdigest()
|
|
|
|
|
|
def test_duplicate_chunk_content_cannot_collide_across_ordinals():
|
|
chunks = chunk(document("samesame"), ChunkPolicy(version="paragraph:v1", max_chars=4))
|
|
assert [item.content for item in chunks] == ["same", "same"]
|
|
assert chunks[0].chunk_id != chunks[1].chunk_id
|
|
|
|
|
|
def test_empty_document_has_no_chunks_and_invalid_policy_is_rejected():
|
|
assert chunk(document(""), ChunkPolicy(version="v1", max_chars=4)) == []
|
|
with pytest.raises(ValueError):
|
|
ChunkPolicy(version="v1", max_chars=0)
|