83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from corpus_mcp.database import (
|
|
CorpusError,
|
|
build_database,
|
|
iter_chunks,
|
|
read_corpus_info,
|
|
)
|
|
|
|
|
|
def test_indexes_supported_files_recursively(tmp_path: Path) -> None:
|
|
source = tmp_path / "documents"
|
|
nested = source / "nested"
|
|
nested.mkdir(parents=True)
|
|
(source / "root.TXT").write_text("Root text", encoding="utf-8-sig")
|
|
(nested / "page.Md").write_text("Nested markdown", encoding="utf-8")
|
|
(nested / "ignored.rst").write_text("Not indexed", encoding="utf-8")
|
|
database = tmp_path / "output" / "corpus.sqlite"
|
|
|
|
count = build_database(source, database, description=" Example corpus ")
|
|
|
|
assert count == 2
|
|
info = read_corpus_info(database)
|
|
assert info["description"] == "Example corpus"
|
|
assert info["document_count"] == 2
|
|
with sqlite3.connect(database) as connection:
|
|
paths = [
|
|
row[0]
|
|
for row in connection.execute("SELECT path FROM documents ORDER BY path")
|
|
]
|
|
assert paths == ["nested/page.Md", "root.TXT"]
|
|
|
|
|
|
def test_empty_file_is_recorded(tmp_path: Path) -> None:
|
|
source = tmp_path / "documents"
|
|
source.mkdir()
|
|
(source / "empty.txt").write_text("", encoding="utf-8")
|
|
database = tmp_path / "corpus.sqlite"
|
|
|
|
assert build_database(source, database) == 1
|
|
with sqlite3.connect(database) as connection:
|
|
assert connection.execute("SELECT count(*) FROM documents").fetchone()[0] == 1
|
|
assert connection.execute("SELECT count(*) FROM chunks").fetchone()[0] == 0
|
|
|
|
|
|
def test_rebuild_is_atomic_when_a_source_cannot_be_decoded(tmp_path: Path) -> None:
|
|
source = tmp_path / "documents"
|
|
source.mkdir()
|
|
(source / "valid.txt").write_text("original", encoding="utf-8")
|
|
database = tmp_path / "corpus.sqlite"
|
|
build_database(source, database, description="Original")
|
|
original_bytes = database.read_bytes()
|
|
(source / "invalid.txt").write_bytes(b"\xff\xfe\xfa")
|
|
|
|
with pytest.raises(CorpusError, match="invalid.txt"):
|
|
build_database(source, database)
|
|
|
|
assert database.read_bytes() == original_bytes
|
|
assert read_corpus_info(database)["description"] == "Original"
|
|
|
|
|
|
def test_chunks_overlap_and_reconstruct_offsets() -> None:
|
|
text = "paragraph one\n\n" + ("word " * 1_200) + "the end"
|
|
chunks = list(iter_chunks(text))
|
|
|
|
assert len(chunks) >= 2
|
|
assert chunks[0][0] == 0
|
|
assert chunks[-1][1] == len(text)
|
|
for previous, current in zip(chunks, chunks[1:]):
|
|
assert current[0] < previous[1]
|
|
assert text[current[0] : current[1]] == current[2]
|
|
|
|
|
|
def test_rejects_non_directory_source(tmp_path: Path) -> None:
|
|
source = tmp_path / "file.txt"
|
|
source.write_text("text", encoding="utf-8")
|
|
|
|
with pytest.raises(CorpusError, match="not a directory"):
|
|
build_database(source, tmp_path / "corpus.sqlite")
|