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
110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""Golden-file tests for the deterministic template parser.
|
|
|
|
Positives: every scrubbed real sample must parse via the deterministic path and
|
|
match its checked-in golden exactly. Negatives: every adversarial / drift sample
|
|
must fail closed to the AI fallback (parsed is None).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from _wo_parser_support import (
|
|
ADVERSARIAL_STEMS,
|
|
AI_FALLBACK_STEMS,
|
|
ASSIGN_STEMS,
|
|
UPDATE_STEMS,
|
|
load_email,
|
|
load_golden,
|
|
)
|
|
from template_parser import try_deterministic_parse
|
|
|
|
|
|
@pytest.mark.parametrize("stem", UPDATE_STEMS)
|
|
def test_update_plaintext_matches_golden(stem):
|
|
email_data = load_email("update-plaintext", stem)
|
|
parsed, method, template_id, reason = try_deterministic_parse(email_data)
|
|
assert method == "template"
|
|
assert template_id == "update_plaintext"
|
|
assert reason == "ok"
|
|
assert parsed == load_golden(stem)
|
|
# Contract invariants for the comment shape.
|
|
assert set(parsed.keys()) == set(load_golden(stem).keys())
|
|
assert parsed["email_type"] == "comment"
|
|
assert parsed["status"] is None # never clobber a real wo_status on upsert
|
|
|
|
|
|
@pytest.mark.parametrize("stem", ASSIGN_STEMS)
|
|
def test_assign_html_matches_golden(stem):
|
|
email_data = load_email("assign-html", stem)
|
|
parsed, method, template_id, reason = try_deterministic_parse(email_data)
|
|
assert method == "template"
|
|
assert template_id == "assign_html"
|
|
assert reason == "ok"
|
|
assert parsed == load_golden(stem)
|
|
assert parsed["email_type"] == "new_work_order"
|
|
assert parsed["status"] == "assigned"
|
|
|
|
|
|
@pytest.mark.parametrize("stem", AI_FALLBACK_STEMS)
|
|
def test_ai_fallback_samples_return_none(stem):
|
|
email_data = load_email("ai-fallback", stem)
|
|
parsed, method, template_id, reason = try_deterministic_parse(email_data)
|
|
assert parsed is None, f"{stem} should have failed closed, got {parsed}"
|
|
assert method == "ai_fallback"
|
|
assert reason != "ok"
|
|
|
|
|
|
def test_positive_result_matches_ai_contract_keys():
|
|
# The deterministic result must carry EXACTLY the keys the AI fallback
|
|
# produces (the EXTRACTION_PROMPT JSON contract) -- no more, no less.
|
|
from template_parser import CONTRACT_KEYS
|
|
|
|
parsed, *_ = try_deterministic_parse(
|
|
load_email("update-plaintext", UPDATE_STEMS[0])
|
|
)
|
|
assert set(parsed.keys()) == set(CONTRACT_KEYS)
|
|
|
|
|
|
def test_extractor_exception_fails_closed(monkeypatch):
|
|
# Any exception inside the extractor must yield None, never a partial parse.
|
|
import template_parser
|
|
|
|
monkeypatch.setattr(
|
|
template_parser,
|
|
"extract_update_plaintext",
|
|
lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("boom")),
|
|
)
|
|
parsed, method, _tid, reason = template_parser.try_deterministic_parse(
|
|
load_email("update-plaintext", UPDATE_STEMS[0])
|
|
)
|
|
assert parsed is None
|
|
assert method == "ai_fallback"
|
|
assert reason == "extractor_raised"
|
|
|
|
|
|
# --- Adversarial: prompt injection / near-miss subjects ---
|
|
|
|
|
|
def test_prompt_injection_in_comment_never_corrupts_other_fields():
|
|
"""Injection text in the comment body must either fall back OR parse safely
|
|
with the real work_order_id / site_code intact and the payload confined to
|
|
comment_text (it must never steer the structured fields)."""
|
|
email_data = load_email("adversarial", "prompt-injection-comment")
|
|
parsed, method, _tid, _reason = try_deterministic_parse(email_data)
|
|
if parsed is None:
|
|
assert method == "ai_fallback"
|
|
return
|
|
# Parsed safely: the subject-derived id wins, injected id/site are ignored.
|
|
assert parsed["work_order_id"] == "11144580730"
|
|
assert parsed["site_code"] == "WCO0"
|
|
assert parsed["email_type"] == "comment"
|
|
assert "00000000000" not in (parsed["work_order_id"] or "")
|
|
assert parsed["site_code"] != "HACK99"
|
|
# The injection payload is confined to the free-text comment field.
|
|
assert "Ignore all previous instructions" in parsed["comment_text"]
|
|
|
|
|
|
@pytest.mark.parametrize("stem", ADVERSARIAL_STEMS)
|
|
def test_adversarial_samples_do_not_raise(stem):
|
|
# The parser must be total: adversarial input returns a tuple, never raises.
|
|
result = try_deterministic_parse(load_email("adversarial", stem))
|
|
assert isinstance(result, tuple) and len(result) == 4
|