mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
273 lines
8.4 KiB
Python
273 lines
8.4 KiB
Python
"""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,
|
|
validate_ai_fallback,
|
|
)
|
|
|
|
# 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
|
|
|
|
|
|
# --- validate_ai_fallback unit tests -----------------------------------------
|
|
|
|
|
|
def _ai_candidate():
|
|
return {k: None for k in CONTRACT_KEYS}
|
|
|
|
|
|
def test_ai_fallback_baseline_is_valid():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "update"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert ok and reason == "ok"
|
|
|
|
|
|
def test_ai_fallback_nondigit_wo_id():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12A45"
|
|
cand["email_type"] = "update"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "missing_required_field"
|
|
|
|
|
|
def test_ai_fallback_null_wo_id():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = None
|
|
cand["email_type"] = "update"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "missing_required_field"
|
|
|
|
|
|
def test_ai_fallback_hash_in_wo_id():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "123#spoofed#deadbeef"
|
|
cand["email_type"] = "update"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "missing_required_field"
|
|
|
|
|
|
def test_ai_fallback_invalid_email_type():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "exploit"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "missing_required_field"
|
|
|
|
|
|
def test_ai_fallback_null_email_type():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = None
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "missing_required_field"
|
|
|
|
|
|
def test_ai_fallback_bad_status_enum():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "update"
|
|
cand["status"] = "frobnicated"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "malformed_site_code"
|
|
|
|
|
|
def test_ai_fallback_valid_status_ok():
|
|
for status in (
|
|
"new",
|
|
"assigned",
|
|
"in_progress",
|
|
"on_hold",
|
|
"completed",
|
|
"cancelled",
|
|
"unknown",
|
|
None,
|
|
):
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "update"
|
|
cand["status"] = status
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert ok, f"status={status} should pass"
|
|
|
|
|
|
def test_ai_fallback_bad_site_code():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "update"
|
|
cand["site_code"] = "workshop"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "malformed_site_code"
|
|
|
|
|
|
def test_ai_fallback_valid_site_codes():
|
|
for code in ("WIL1", "ZDL8", "AB12", None):
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "update"
|
|
cand["site_code"] = code
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert ok, f"site_code={code} should pass"
|
|
|
|
|
|
def test_ai_fallback_key_set_mismatch_extra():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "update"
|
|
cand["surprise"] = "x"
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "key_set_mismatch"
|
|
|
|
|
|
def test_ai_fallback_key_set_mismatch_missing():
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = "update"
|
|
del cand["address"]
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert not ok and reason == "key_set_mismatch"
|
|
|
|
|
|
def test_ai_fallback_all_valid_email_types():
|
|
for et in ("new_work_order", "update", "comment", "cancellation"):
|
|
cand = _ai_candidate()
|
|
cand["work_order_id"] = "12345"
|
|
cand["email_type"] = et
|
|
ok, reason = validate_ai_fallback(cand)
|
|
assert ok, f"email_type={et} should pass"
|