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"


# --- Advisory A3: multi-paragraph comments must not be truncated ---


def test_multi_paragraph_comment_is_captured_in_full():
    """A blank line inside the New Comment block is a paragraph break, not the
    end of the comment -- the block ends at the next label/separator."""
    email_data = {
        "subject": "AMAZON UPDATE WO DETAILS 11144580730",
        "sender": "noreply@hxgnsmartcloud.com",
        "to": "apm@int.seahaven.com",
        "cc": "",
        "date": "Mon, 27 Apr 2026 23:57:49 +0000 (UTC)",
        "body": (
            "New Comment:\n"
            "first paragraph line one\n"
            "first paragraph line two\n"
            "\n"
            "second paragraph after a blank line\n"
            "\n"
            "Creation Time(UTC): 2026-04-27 23:51:48\n"
            "Submitted By: jdoe\n"
            "\n"
            "_________________\n"
            "\n"
            "Work Order:  11144580730 - fix dock door\n"
            "\n"
            "Building: WCO0.\n"
            "\n"
            "\n"
            "Please review it in Amazon Portal\n"
        ),
    }
    parsed, method, template_id, reason = try_deterministic_parse(email_data)
    assert method == "template"
    assert template_id == "update_plaintext"
    assert reason == "ok"
    assert parsed["comment_text"] == (
        "first paragraph line one\n"
        "first paragraph line two\n"
        "\n"
        "second paragraph after a blank line"
    )
    # Fixed-position blocks still stop at blank lines: the portal footer after
    # Building must not bleed into the address.
    assert parsed["address"] is None
    assert parsed["comment_time"] == "2026-04-27T23:51:48"


def test_huge_blank_run_in_comment_parses_in_linear_time():
    """A New Comment block padded with a massive blank-line run must not go
    quadratic in the leading/trailing-blank trim (CWE-407 guard)."""
    import time

    from template_parser import _T1_LABELS, _capture_block

    lines = ["New Comment:"] + [""] * 500_000 + ["x"] + [""] * 500_000
    start = time.monotonic()
    out = _capture_block(lines, 0, _T1_LABELS, stop_on_blank=False)
    elapsed = time.monotonic() - start
    assert out == ["x"]
    assert elapsed < 2.0  # quadratic trim took ~20s+ at this size


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