backend/tests/test_notes_parser.py

from __future__ import annotations

from backend.core import notes_parser, template_discoverer
from backend.core.notes_parser import UNASSIGNED
from backend.core.rics_canonical_l3 import iter_canonical_leaf_sections


def _schema(master_docx):
    schema, _ = template_discoverer.discover_schema(master_docx)
    return schema


def test_section_anchors_initialized():
    assert notes_parser.initialize_section_anchors() == len(iter_canonical_leaf_sections())


def test_routes_by_semantic_similarity(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "The chimney is cracked.\n\nSlipped slate tiles on external roof coverings.", schema
    )
    by_id = {n.section_id: n.text for n in notes}
    assert "chimney" in by_id.get("D1", "").lower()
    assert "slate" in by_id.get("D2", "").lower()


def test_semantic_chimney_stack_routes_d1(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "significant cracking noted to the chimney stack", schema
    )
    by_id = {n.section_id: n.text for n in notes}
    assert "cracking" in by_id.get("D1", "").lower()


def test_unmatched_goes_to_unassigned(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "Zebras grazing savannah extraterrestrial wildlife xyzzy plugh.", schema
    )
    ids = {n.section_id for n in notes}
    assert UNASSIGNED in ids


def test_no_rating_value_when_schema_has_none(master_docx):
    schema = _schema(master_docx)
    schema.rating_system.detected = False
    notes = notes_parser.parse_notes_to_sections("chimney stack rating 2 cracking", schema)
    d1 = next(n for n in notes if n.section_id == "D1")
    assert d1.rating_value is None


def test_raw_observations_populated(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "significant cracking noted to the chimney stack", schema
    )
    d1 = next(n for n in notes if n.section_id == "D1")
    assert len(d1.raw_observations) >= 1
    assert "cracking" in d1.raw_observations[0].lower()


def test_notes_with_pii_preserved_verbatim(master_docx):
    """Surveyor notes are not redacted before parsing (RAG ingest is scrubbed, not notes)."""
    schema = _schema(master_docx)
    raw = "chimney stack cracking at 14 Acacia Road SW1A 1AA and contact john@client.com"
    notes = notes_parser.parse_notes_to_sections(raw, schema)
    d1 = next(n for n in notes if n.section_id == "D1")
    assert "SW1A 1AA" in d1.text
    assert "john@client.com" in d1.text


def test_rating_from_schema_values(master_docx_rated):
    from backend.models.schema import RatingSystem, RatingValue

    schema, _ = template_discoverer.discover_schema(master_docx_rated)
    schema.rating_system = RatingSystem(
        detected=True,
        type="numeric",
        values=[RatingValue(value="1"), RatingValue(value="2"), RatingValue(value="3")],
    )
    notes = notes_parser.parse_notes_to_sections("chimney stack CR2 cracking", schema)
    d1 = next(n for n in notes if n.section_id == "D1")
    assert d1.rating_value == "2"


def test_semantic_insulation_routes_j1(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections("insulation looks thin in loft attic", schema)
    j1 = next(n for n in notes if n.section_id == "J1")
    assert "insulation" in j1.text.lower()


def test_semantic_cavity_walls_routes_d4(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections("Cavity walls show erosion and render loss", schema)
    d4 = next(n for n in notes if n.section_id == "D4")
    assert "cavity" in d4.text.lower()


def test_external_walls_cavity_routes_d4_not_c_group(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "External walls: 300mm cavity wall construction with step cracking", schema
    )
    by_id = {n.section_id: n.text for n in notes}
    assert "D4" in by_id
    assert "cavity wall" in by_id["D4"].lower()
    assert not any(sid.startswith("C") for sid in by_id if sid != "UNASSIGNED")


def test_explicit_d9_prefix_routes_d9(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "D9: External walls: 300mm cavity wall construction with step cracking", schema
    )
    d9 = next(n for n in notes if n.section_id == "D9")
    assert "External walls" in d9.text
    assert not d9.text.upper().startswith("D9:")


def test_rainwater_fittings_routes_d3(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "Rainwater fittings: UPVC gutters and gullies present. Rainwater downpipe discharges below ground.",
        schema,
    )
    d3 = next(n for n in notes if n.section_id == "D3")
    assert "gutter" in d3.text.lower()


def test_g1_comma_prefix_routes_g1_strips_token(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "G1, garage roof pitched design with concrete tiles", schema
    )
    g1 = next(n for n in notes if n.section_id == "G1")
    assert "garage roof" in g1.text.lower()
    assert "G1," not in g1.text


def test_semantic_asbestos_routes_i3(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "Suspected chrysotile artex to ceiling in garage.", schema
    )
    assert any(n.section_id == "I3" for n in notes)


def test_roof_structure_routes_e1(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "Loft hatch access: cut rafters and purlins visible in roof structure.",
        schema,
    )
    assert any(n.section_id == "E1" for n in notes)


def test_external_slates_route_d2_not_e1(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "External roof coverings: slipped clay tiles and lead flashing to valley gutter.",
        schema,
    )
    by_id = {n.section_id: n.text for n in notes}
    assert "D2" in by_id
    assert "E1" not in by_id or "clay tiles" not in by_id.get("E1", "").lower()


def test_vague_note_stays_unassigned(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "Zebras grazing savannah extraterrestrial wildlife xyzzy plugh.", schema
    )
    unassigned = next(n for n in notes if n.section_id == UNASSIGNED)
    assert unassigned.raw_observations


def test_nonsense_note_stays_unassigned(master_docx):
    schema = _schema(master_docx)
    notes = notes_parser.parse_notes_to_sections(
        "Z9: xyzzy plugh quantum frobnicator", schema
    )
    assert any(n.section_id == UNASSIGNED for n in notes)
    assert not any(n.section_id == "Z9" for n in notes)


def test_route_note_to_section_api(master_docx):
    schema = _schema(master_docx)
    assert notes_parser.route_note_to_section("chimney stack crack", schema) == "D1"
    assert notes_parser.route_note_to_section(
        "slipped slate tiles on external roof coverings south slope", schema
    ) == "D2"
    assert notes_parser.route_note_to_section("xyzzy plugh quantum frobnicator", schema) == UNASSIGNED


def test_confidence_threshold_gate(master_docx, monkeypatch):
    from backend.config import settings

    schema = _schema(master_docx)
    monkeypatch.setattr(settings, "note_routing_mode", "rag")
    monkeypatch.setattr(settings, "note_rag_match_min_score", 0.99)
    notes_parser.reset_section_anchors()
    notes_parser.initialize_section_anchors()
    # Keywords win before RAG — chimney vocabulary is deterministic D1.
    assert notes_parser.route_note_to_section("chimney stack cracked", schema) == "D1"
    # Nonsense with no keyword hit must still respect the RAG confidence gate.
    assert notes_parser.route_note_to_section("xyzzy plugh quantum frobnicator", schema) == UNASSIGNED