from __future__ import annotations from backend.core import pii_scrubber from backend.core.rag_store import TIER_MASTER, TIER_REFERENCE, Chunk, get_rag_store def test_reference_ingest_strips_cross_property_identifiers(): store = get_rag_store() raw = ( "Survey for Mr John Smith at 22 High Street, London SE1 2AB. " "Contact john.smith@example.com. Report ref NCS-100147." ) n = store.ingest_document( "t-ref", "reference:past_report.docx", [Chunk(text=raw, tier=TIER_REFERENCE)], tier=TIER_REFERENCE, ) assert n == 1 hits = store.search("t-ref", "survey high street", tier=TIER_REFERENCE, top_k=1) assert hits text = hits[0].text.lower() assert "john.smith@example.com" not in text assert "se1 2ab" not in text assert "ncs-100147" not in text assert hits[0].is_scrubbed is True def test_past_report_not_visible_to_generation_search(): store = get_rag_store() store.ingest_document( "t-gen", "ref:old", [Chunk(text="Property at 99 Baker Street NW1 6XE had subsidence.", tier=TIER_REFERENCE)], tier=TIER_REFERENCE, ) store.ingest_document( "t-gen", "master:m", [Chunk(text="The site was inspected externally.", section_id="A", tier=TIER_MASTER)], tier=TIER_MASTER, ) gen_hits = store.search_for_generation("t-gen", "property inspected", top_k=5) assert all(h.tier == TIER_MASTER for h in gen_hits) combined = " ".join(h.text for h in gen_hits).lower() assert "baker street" not in combined assert "nw1" not in combined def test_scrub_reference_drops_unsafe_chunk(): """Chunk is dropped when PII cannot be fully removed (simulated via detect).""" # Heavily redacted content should pass; empty after scrub should drop. result = pii_scrubber.scrub_reference_for_ingest(" ") assert result is None