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

"""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