237 lines
7.4 KiB
Python
237 lines
7.4 KiB
Python
"""Testes do indexador: manifesto incremental, reprocessamento e falhas parciais."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from app.bible_parser import BibleParser # noqa: E402
|
|
from app.embedding import HashEmbedder # noqa: E402
|
|
from app.extraction import PageText # noqa: E402
|
|
from app.indexer import Indexer, IndexStatus # noqa: E402
|
|
from app.vector_store import InMemoryVectorStore # noqa: E402
|
|
|
|
|
|
class FakeExtractor:
|
|
"""Extrator controlado: devolve as páginas pré-definidas por arquivo."""
|
|
|
|
def __init__(self, pages_by_file: dict[str, list[PageText]]):
|
|
self.pages_by_file = pages_by_file
|
|
self.calls: list[str] = []
|
|
|
|
def extract_document(self, path: Path, file_hash: str | None = None):
|
|
self.calls.append(Path(path).name)
|
|
yield from self.pages_by_file.get(Path(path).name, [])
|
|
|
|
|
|
GENESIS_PAGE = (
|
|
"GÊNESIS\n"
|
|
"[1]\n"
|
|
"1 No princípio criou Deus os céus e a terra.\n"
|
|
"2 A terra era sem forma e vazia.\n"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def settings(tmp_path):
|
|
from app.config import Settings
|
|
|
|
return Settings(
|
|
pdf_dir=tmp_path / "pdfs",
|
|
vector_db_dir=tmp_path / "vector_db",
|
|
embedding_model="hash-test",
|
|
)
|
|
|
|
|
|
def build_indexer(settings, extractor, embedder=None, store=None):
|
|
embedder = embedder or HashEmbedder(dimensions=32)
|
|
store = store or InMemoryVectorStore(embedder=embedder)
|
|
indexer = Indexer(
|
|
settings=settings,
|
|
extractor=extractor,
|
|
parser=BibleParser(),
|
|
embedder=embedder,
|
|
store=store,
|
|
)
|
|
return indexer, store
|
|
|
|
|
|
def write_pdf(directory: Path, name: str, content: bytes = b"%PDF-1.3") -> Path:
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
path = directory / name
|
|
path.write_bytes(content)
|
|
return path
|
|
|
|
|
|
def test_reindex_indexes_pdfs_and_reports_counts(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, store = build_indexer(settings, extractor)
|
|
|
|
status = indexer.reindex()
|
|
|
|
assert status.state == "ready"
|
|
assert status.pdfs_total == 1
|
|
assert status.pdfs_indexed == 1
|
|
assert status.chunks == 2
|
|
assert store.count() == 2
|
|
|
|
|
|
def test_reindex_is_incremental_when_hash_is_unchanged(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
indexer.reindex()
|
|
indexer.reindex()
|
|
|
|
assert extractor.calls == ["biblia.pdf"], "o PDF não deveria ser reprocessado"
|
|
|
|
|
|
def test_reindex_reprocesses_file_when_content_changes(settings):
|
|
pdf_path = write_pdf(settings.pdf_dir, "biblia.pdf", b"%PDF-1.3 original")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
indexer.reindex()
|
|
pdf_path.write_bytes(b"%PDF-1.3 modified content")
|
|
indexer.reindex()
|
|
|
|
assert extractor.calls == ["biblia.pdf", "biblia.pdf"]
|
|
|
|
|
|
def test_reindex_with_force_reprocesses_everything(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
indexer.reindex()
|
|
indexer.reindex(force=True)
|
|
|
|
assert extractor.calls == ["biblia.pdf", "biblia.pdf"]
|
|
|
|
|
|
def test_manifest_is_persisted(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
indexer.reindex()
|
|
|
|
manifest = json.loads(settings.manifest_path.read_text(encoding="utf-8"))
|
|
assert "biblia.pdf" in manifest
|
|
assert manifest["biblia.pdf"]["state"] == "complete"
|
|
assert manifest["biblia.pdf"]["chunks"] == 2
|
|
assert manifest["biblia.pdf"]["sha256"]
|
|
|
|
|
|
def test_partial_file_is_recorded_as_partial(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor(
|
|
{
|
|
"biblia.pdf": [
|
|
PageText(page=6, text=GENESIS_PAGE),
|
|
PageText(page=7, text="", failed=True),
|
|
]
|
|
}
|
|
)
|
|
indexer, store = build_indexer(settings, extractor)
|
|
|
|
status = indexer.reindex()
|
|
|
|
manifest = json.loads(settings.manifest_path.read_text(encoding="utf-8"))
|
|
assert manifest["biblia.pdf"]["state"] == "partial"
|
|
assert manifest["biblia.pdf"]["failed_pages"] == [7]
|
|
assert status.errors
|
|
|
|
|
|
def test_failed_file_is_recorded_as_error(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text="", failed=True)]})
|
|
indexer, store = build_indexer(settings, extractor)
|
|
|
|
indexer.reindex()
|
|
|
|
manifest = json.loads(settings.manifest_path.read_text(encoding="utf-8"))
|
|
assert manifest["biblia.pdf"]["state"] == "error"
|
|
assert store.count() == 0
|
|
|
|
|
|
def test_reindex_replaces_old_chunks_of_the_same_file(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, store = build_indexer(settings, extractor)
|
|
|
|
indexer.reindex()
|
|
indexer.reindex(force=True)
|
|
|
|
assert store.count() == 2, "não deve haver duplicação após reindexação"
|
|
|
|
|
|
def test_status_reports_chunks_from_the_store(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, store = build_indexer(settings, extractor)
|
|
|
|
indexer.reindex()
|
|
status = indexer.status()
|
|
|
|
assert isinstance(status, IndexStatus)
|
|
assert status.chunks == store.count()
|
|
|
|
|
|
def test_empty_pdf_directory_is_ready_with_zero_chunks(settings):
|
|
settings.pdf_dir.mkdir(parents=True, exist_ok=True)
|
|
extractor = FakeExtractor({})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
status = indexer.reindex()
|
|
|
|
assert status.state == "ready"
|
|
assert status.pdfs_total == 0
|
|
assert status.chunks == 0
|
|
|
|
|
|
def test_non_pdf_files_are_ignored(settings):
|
|
settings.pdf_dir.mkdir(parents=True, exist_ok=True)
|
|
(settings.pdf_dir / "notas.txt").write_text("ignorado", encoding="utf-8")
|
|
extractor = FakeExtractor({})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
status = indexer.reindex()
|
|
|
|
assert status.pdfs_total == 0
|
|
assert extractor.calls == []
|
|
|
|
|
|
def test_status_reports_indexing_while_work_is_in_progress(settings):
|
|
"""Durante a indexação o status deve mostrar progresso, não zero."""
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
indexer.mark_indexing()
|
|
status = indexer.status()
|
|
|
|
assert status.state == "indexing"
|
|
assert status.pdfs_total == 1
|
|
|
|
|
|
def test_status_counts_pages_processed_during_indexing(settings):
|
|
write_pdf(settings.pdf_dir, "biblia.pdf")
|
|
extractor = FakeExtractor({"biblia.pdf": [PageText(page=6, text=GENESIS_PAGE)]})
|
|
indexer, _ = build_indexer(settings, extractor)
|
|
|
|
indexer.mark_indexing()
|
|
indexer.report_progress(pdf_name="biblia.pdf", pages_done=120, pages_total=3364)
|
|
|
|
status = indexer.status()
|
|
|
|
assert status.pages_processed == 120
|
|
assert status.pages_total == 3364
|