procurement-ingest/lambdas/wo/email_processor/tests/test_validation_gate.py

"""Fail-closed validation-gate tests.

Each known drift / adversarial shape must be rejected with the expected reason
code, and direct unit tests exercise each individual gate rule.
"""

import pytest

from _wo_parser_support import load_email
from template_parser import (
    CONTRACT_KEYS,
    extract_update_plaintext,
    try_deterministic_parse,
    validate,
)

# Fixture stem -> expected fail-closed reason code.
EXPECTED_REASONS = {
    "unknown-subject": "subject_no_match",
    "missing-id": "subject_no_match",
    "single-space-work-order": "single_space_work_order",
    "nondigit-id": "missing_required_field",
    "empty-new-comment": "missing_required_field",
    "malformed-site-code": "malformed_site_code",
    "label-bleed-comment": "label_bleed",
    "unparseable-creation-time": "creation_time_unparseable",
    "cancellation-t1": "missing_required_field",
    "update-status-t1": "missing_required_field",
    "missing-building-and-comment": "missing_required_field",
    "t2-wo-id-mismatch": "wo_id_mismatch",
    "t2-missing-address": "missing_required_field",
    "t2-unparseable-date": "creation_time_unparseable",
}


@pytest.mark.parametrize("stem,reason", sorted(EXPECTED_REASONS.items()))
def test_gate_rejects_with_reason(stem, reason):
    parsed, method, _tid, got_reason = try_deterministic_parse(
        load_email("ai-fallback", stem)
    )
    assert parsed is None
    assert method == "ai_fallback"
    assert got_reason == reason


# --- Direct unit tests on validate() ---


def _good_t1_email():
    return load_email("update-plaintext", "update-plaintext-01")


def _good_candidate():
    return extract_update_plaintext(_good_t1_email())


def test_baseline_candidate_is_valid():
    ok, reason = validate(_good_candidate(), "update_plaintext", _good_t1_email())
    assert ok and reason == "ok"


def test_rule1_unknown_template():
    ok, reason = validate(_good_candidate(), "unknown", _good_t1_email())
    assert not ok and reason == "subject_no_match"


def test_rule2_extra_key_fails():
    cand = _good_candidate()
    cand["surprise"] = "x"
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "key_set_mismatch"


def test_rule2_missing_key_fails():
    cand = _good_candidate()
    del cand["address"]
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "key_set_mismatch"


def test_rule3_nondigit_wo():
    cand = _good_candidate()
    cand["work_order_id"] = "12A45"
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "missing_required_field"


def test_rule5_wrong_email_type():
    cand = _good_candidate()
    cand["email_type"] = "new_work_order"  # wrong for a T1 template
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "email_type_mismatch"


def test_rule6_bad_site_code():
    cand = _good_candidate()
    cand["site_code"] = "workshop"
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "malformed_site_code"


def test_rule7_bad_status():
    cand = _good_candidate()
    cand["status"] = "frobnicated"
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "malformed_site_code"


def test_rule8_empty_comment_text():
    cand = _good_candidate()
    cand["comment_text"] = "   "
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "missing_required_field"


def test_rule9_label_bleed_in_comment():
    cand = _good_candidate()
    cand["comment_text"] = "text that leaked Building: WCO0 into the value"
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "label_bleed"


def test_rule9_separator_bleed_in_address():
    cand = _good_candidate()
    cand["address"] = "123 Main St ________________"
    ok, reason = validate(cand, "update_plaintext", _good_t1_email())
    assert not ok and reason == "label_bleed"


def test_contract_keys_match_extraction_prompt():
    """The parser's key set must be exactly the AI EXTRACTION_PROMPT contract, so
    the deterministic and AI-fallback paths write identical shapes downstream."""
    import re

    import handler

    # The prompt's JSON skeleton uses union-type pseudo-values (not strict JSON)
    # and repeats some enum terms in prose bullets, so pull quoted "key": tokens
    # and assert every contract key is a field the AI is asked to emit (the
    # parser must never invent a key outside the AI contract).
    prompt_keys = set(re.findall(r'"([a-z_]+)":', handler.EXTRACTION_PROMPT))
    assert set(CONTRACT_KEYS).issubset(prompt_keys)
    assert len(CONTRACT_KEYS) == 16  # current EXTRACTION_PROMPT field count