"""Per-note RAG threshold gating before baseline mapping.""" from __future__ import annotations from pathlib import Path import pytest from docx import Document from backend.config import settings from backend.core import ingest, section_mapper, template_discoverer from backend.core.observation_matcher import ( format_unmatched_observation_tag, observation_matches_baseline, partition_observations_for_baseline, ) from backend.core.section_mapper import _format_unmatched_section_block def _seed_roof_tenant(tmp_path: Path, monkeypatch, tenant: str = "obs-match") -> None: monkeypatch.setattr(settings, "data_dir", str(tmp_path / "data")) tpl = tmp_path / "template.docx" doc = Document() doc.add_paragraph("D2 Roof coverings") doc.save(str(tpl)) past = tmp_path / "past.docx" doc2 = Document() doc2.add_paragraph("D2\nRoof coverings") doc2.add_paragraph( "The roof covering comprised slate tiles laid to pitched timber rafters. " "Inspection was limited to ground level using binoculars." ) doc2.save(str(past)) ingest.ingest_report_template(tenant, tpl) ingest.ingest_reference(tenant, past) def test_format_unmatched_observation_tag(): assert format_unmatched_observation_tag("asbestos noted") == ( "[UNMATCHED_OBSERVATION: asbestos noted]" ) def test_unmatched_section_block_format(): block = _format_unmatched_section_block(["slate slipped south slope", "moss on north"]) assert "### UNMATCHED_OBSERVATION" in block assert "* slate slipped south slope" in block assert "* moss on north" in block def test_partition_splits_low_match_notes(tmp_path, monkeypatch): tenant = "obs-part" _seed_roof_tenant(tmp_path, monkeypatch, tenant) schema = template_discoverer.load_schema(tenant) assert schema is not None baseline = ( "The roof covering comprised slate tiles laid to pitched timber rafters. " "Inspection was limited to ground level using binoculars." ) matched, unmatched = partition_observations_for_baseline( tenant, [ "slate tile slipped south slope", "completely invented asbestos contamination in cavity", ], baseline, paragraph_section_id=schema.paragraph_section_id("D2"), interference_level="minimum", ) assert matched assert any("slate" in m.lower() for m in matched) assert unmatched assert any("asbestos" in u.lower() for u in unmatched) def test_unmatched_note_never_invented_in_section_text(tmp_path, monkeypatch): tenant = "obs-e2e" _seed_roof_tenant(tmp_path, monkeypatch, tenant) monkeypatch.setattr(settings, "note_rag_match_min_score", 0.38) result = section_mapper.generate_report( tenant, "slate tile slipped south slope\nasbestos contamination in roof void", interference_level="minimum", ) d3 = next(s for s in result.sections if s.section_id == "D2") assert d3.status == "OK" mapped_body, _, _ = d3.text.partition("### UNMATCHED_OBSERVATION") assert "slate" in mapped_body.lower() or "slate" in d3.text.lower() assert "asbestos" not in mapped_body.lower() for sec in result.sections: if sec.section_id != "I3": assert "asbestos" not in (sec.text or "").lower() def test_observation_matches_baseline_uses_lexical_overlap(tmp_path, monkeypatch): tenant = "obs-lex" _seed_roof_tenant(tmp_path, monkeypatch, tenant) baseline = "The chimney stacks were inspected from ground level using binoculars." assert observation_matches_baseline( tenant, "chimney stack hairline crack noted", baseline, paragraph_section_id="D1", interference_level="minimum", )