File size: 5,174 Bytes
547ce6e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """Shared fixtures.
The suite runs against the REAL corpus and the REAL index rather than fixtures. A parser
tested on a synthetic PDF proves nothing about the government typesetting it actually has
to survive, and a retrieval test over invented chunks measures nothing about retrieval
quality. Tests that need the index are marked `integration` and skip cleanly when it has
not been built.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from app.core.models import Chunk, ScoredChunk
from app.core.settings import Settings, get_settings
from app.rag.parse import ParsedDocument, ParseDiagnostics, parse_corpus
@pytest.fixture(scope="session")
def settings() -> Settings:
return get_settings()
@pytest.fixture(scope="session")
def corpus_available(settings: Settings) -> bool:
return settings.pdf_dir.exists() and any(settings.pdf_dir.glob("*.pdf"))
def _lock_held_by_another_process(lock: Path) -> bool:
"""Whether another process holds the embedded store's advisory lock.
Deliberately does *not* construct a ``QdrantClient`` to find out. That constructor
opens the lock file and only then raises, leaving the caller no reference with which
to close the handle; the orphan surfaces as an unraisable ``ResourceWarning`` at
interpreter shutdown, which under ``-W error`` fails the whole run and points at a
lock file rather than at the process actually holding it. The previous version of
this fixture did exactly that, and its docstring claimed it *prevented* the warning.
Taking the same advisory lock on our own file object keeps ownership of the handle
here, where ``with`` closes it on every path. ``portalocker`` is what qdrant-client
itself uses, so this tests the same lock and not merely a similar one.
"""
try:
import portalocker
except ImportError: # pragma: no cover - qdrant-client always provides it
return False
try:
with lock.open("r+", encoding="utf-8") as handle:
try:
portalocker.lock(handle, portalocker.LOCK_EX | portalocker.LOCK_NB)
except portalocker.exceptions.BaseLockException:
return True
portalocker.unlock(handle)
except OSError:
return False
return False
@pytest.fixture(scope="session")
def index_available(settings: Settings) -> bool:
"""Whether the on-disk index is present AND usable by this process.
Embedded Qdrant allows a single writer, so a running `make dev` makes the store
unopenable here. Detect that and skip with an instruction, rather than failing
somewhere unrelated.
"""
if not (settings.chunks_path.exists() and settings.bm25_path.exists()):
return False
# A configured remote Qdrant is not lock-protected and supports concurrent readers.
if getattr(settings, "qdrant_url", None):
return True
lock = settings.qdrant_path / ".lock"
if lock.exists() and _lock_held_by_another_process(lock):
pytest.skip(
"var/qdrant is locked by another process. Embedded Qdrant allows one "
"writer at a time — run `make stop` before `make test`."
)
return True
@pytest.fixture(scope="session")
def parsed(corpus_available: bool) -> list[ParsedDocument]:
if not corpus_available:
pytest.skip("corpus PDFs not present; run `make corpus`")
return parse_corpus(diagnostics=ParseDiagnostics())
@pytest.fixture(scope="session")
def diagnostics(corpus_available: bool) -> ParseDiagnostics:
if not corpus_available:
pytest.skip("corpus PDFs not present; run `make corpus`")
diag = ParseDiagnostics()
parse_corpus(diagnostics=diag)
return diag
@pytest.fixture(scope="session")
def chunks(settings: Settings, index_available: bool) -> list[Chunk]:
if not index_available:
pytest.skip("index not built; run `make index`")
from app.rag.index import load_chunks
return load_chunks(settings)
def make_chunk(
*,
law_id: str = "uae-labour-law",
law_label: str = "Labour Law",
article_no: int = 51,
text: str = "Labour Law - Article 51\nEnd of service benefits are payable.",
part_id: str = "part-1",
) -> Chunk:
"""A minimal chunk for pure-logic tests that must not touch the index."""
return Chunk(
chunk_id=Chunk.make_id(law_id, part_id, article_no, 0, text),
law_id=law_id,
law_label=law_label,
law_title=f"{law_label} title",
part_id=part_id,
part_title="Part",
article_no=article_no,
article_title="End of Service Benefits",
section="",
seq=0,
seq_total=1,
page_start=1,
page_end=1,
text=text,
token_count=12,
)
def make_scored(chunk: Chunk | None = None, **kwargs: object) -> ScoredChunk:
from app.core.models import RetrievalSource
return ScoredChunk(chunk=chunk or make_chunk(), source=RetrievalSource.BOTH, **kwargs) # type: ignore[arg-type]
@pytest.fixture
def tmp_index_dir(tmp_path: Path) -> Path:
"""A scratch directory for tests that write index artefacts."""
return tmp_path
|