"""rerank() must rank an unembedded chunk, not crash on it. Split from test_model_contract_shapes.py because tools.retrieval imports psycopg: this module skips on a bare checkout and runs inside the ai-api image, where the whole suite runs with nothing skipped: docker run --rm --entrypoint python yau/ai-api:local -m pytest /app/tests -q """ import pytest pytest.importorskip("psycopg") from tools.retrieval import Chunk, rerank # noqa: E402 def chunk(cid: int, text: str, similarity: float | None) -> Chunk: return Chunk( id=cid, source_file="procedures/X.md", doc_type="procedure", doc_number="WRPS-X-001", revision="0", effective_date=None, page=1, section_title="S", equipment_id=None, chunk_text=text, similarity=similarity, ) def test_rerank_survives_a_null_similarity(): """The bug: `0.75 * None` raised TypeError and 500'd the question. A chunk ingested with --no-embed has a NULL embedding, so `1 - (NULL <=> vec)` comes back NULL. """ chunks = [chunk(1, "interlock bypass pump", None), chunk(2, "unrelated text", 0.4)] assert len(rerank(chunks, "interlock bypass procedure", top_n=2)) == 2 def test_unembedded_chunk_still_ranks_on_lexical_overlap(): """It should degrade to the lexical half, not vanish and not win.""" strong_words = chunk(1, "interlock bypass procedure pump", None) weak_words = chunk(2, "boiler feedwater chemistry", None) assert rerank([weak_words, strong_words], "interlock bypass procedure", top_n=1) == [strong_words] def test_embedded_chunk_outranks_an_unembedded_one_on_equal_text(): text = "interlock bypass procedure" embedded = chunk(1, text, 0.9) unembedded = chunk(2, text, None) assert rerank([unembedded, embedded], text, top_n=1) == [embedded] # --------------------------------------------------------------------------- # Step sections are withheld from the procedural branch # --------------------------------------------------------------------------- from tools.retrieval import STEP_SECTION_RE # noqa: E402 def test_step_sections_are_recognised(): """These are the sections that ARE the instructions. find_procedure must not hand them to the model - an interlock exists because somebody assessed a hazard, and a bypass reassembled from fragments is a safety document nobody approved.""" for title in ["3. Procedure", "Steps", "4. Method", "Instructions", "5. Execution", "Work Instruction", "6. Restoration"]: assert STEP_SECTION_RE.search(title), title def test_identifying_sections_are_not_withheld(): """The header carries the title and authorising role; prerequisites are what a procedural answer quotes. Withholding either was the bug.""" for title in ["DEMO DOCUMENT — NOT A CONTROLLED DOCUMENT", "1. Purpose", "2. Prerequisites", "Scope", "Hazards", "References"]: assert not STEP_SECTION_RE.search(title), title def test_a_missing_section_title_is_not_treated_as_a_step(): assert not STEP_SECTION_RE.search("")