mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
Adam Moussa
89bf47c1bb
Cover the deterministic parser with golden-file tests over 55 real scrubbed .eml fixtures (both comment sub-shapes, username Submitted-By, address present/absent, br+CRLF assign addresses), fail-closed validation-gate rules, adversarial and prompt-injection cases that must route to ai_fallback or parse without corrupting other fields, the issue #23 comment_id idempotency invariants, and the Bedrock-fallback dispatch plus EMF-metric emission with a mocked invoke_model. Extend pytest.ini testpaths to discover the co-located suite, and update tests/conftest.load_handler to put a handler's own directory on sys.path so the WO handler's new `from template_parser import ...` resolves under the existing shared handler tests. Point test_local.py at the new template-first + Bedrock flow. Refs: #23
143 lines
4.7 KiB
Python
143 lines
4.7 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,
|
|
)
|
|
|
|
# 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
|