From 76f60abadec466a2d351a7737293ef65d757e7b5 Mon Sep 17 00:00:00 2001 From: Adam Moussa <166072409+amoussa1229@users.noreply.github.com> Date: Fri, 29 May 2026 14:59:40 -0400 Subject: [PATCH] Add per-bucket and precedence tests for comment intent --- tests/test_comment_intent.py | 284 +++++++++++++++++++++++++++++++++++ 1 file changed, 284 insertions(+) create mode 100644 tests/test_comment_intent.py diff --git a/tests/test_comment_intent.py b/tests/test_comment_intent.py new file mode 100644 index 0000000..45be12f --- /dev/null +++ b/tests/test_comment_intent.py @@ -0,0 +1,284 @@ +"""Per-bucket and precedence tests for classify.comment_intent(). + +Tests operate on *already-stripped* plain text (as comment_intent() expects). +End-to-end precedence is also verified via classify.classify(), which handles +HTML stripping internally. + +Bucket coverage: + - One positive case per ladder bucket that can fire deterministically: + 3rd Escalation, 2nd Escalation, 1st Escalation, SIM Ticket, Vendor No-Show, + Weekly WO Scheduled, Avetta Project Created, Schedule Confirmed, + Completed / Pending Close, Rescheduled, Report / Docs Needed, + Awaiting Report / Invoice, Awaiting Scheduling, Status Inquiry, On Hold, + Acknowledgement / No-op. + - "Other Escalation" has no dedicated regex rule in the ladder and is + intentionally omitted (it is a Haiku-only bucket). + +Precedence / negative traps: + - 3rd attempt + report language -> 3rd Escalation (most-specific wins). + - 1st attempt process for schedule confirmation -> Awaiting Scheduling + (not an escalation; the ladder requires an explicit escalation signal). + - Bare "1st attempt to contact vendor" (no escalation keyword) -> None. + - Empty / whitespace -> None. + - "Copy" / "Copy 5/26." -> Acknowledgement / No-op (fullmatch). +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert( + 0, str(Path(__file__).resolve().parent.parent / "lambdas" / "classifier") +) + +import classify # noqa: E402 + + +# --------------------------------------------------------------------------- +# Axis-1 per-bucket positive cases +# --------------------------------------------------------------------------- + + +class TestCommentIntentBuckets: + def test_3rd_escalation_attempt(self): + assert ( + classify.comment_intent("3rd attempt process for schedule confirmation") + == "3rd Escalation" + ) + + def test_3rd_escalation_explicit(self): + assert ( + classify.comment_intent("3rd escalation sent to vendor") == "3rd Escalation" + ) + + def test_2nd_escalation_attempt(self): + assert ( + classify.comment_intent("2nd attempt process for schedule confirmation") + == "2nd Escalation" + ) + + def test_2nd_escalation_explicit(self): + assert ( + classify.comment_intent("2nd escalation sent to vendor") == "2nd Escalation" + ) + + def test_1st_escalation_explicit(self): + assert ( + classify.comment_intent("1st escalation sent to management") + == "1st Escalation" + ) + + def test_sim_ticket_with_version(self): + assert ( + classify.comment_intent("SIM ticket v1234567 created for this WO") + == "SIM Ticket" + ) + + def test_sim_ticket_tcorp_url(self): + assert ( + classify.comment_intent("SIM TT opened t.corp.amazon.com/issues/123") + == "SIM Ticket" + ) + + def test_vendor_no_show(self): + assert classify.comment_intent("Vendor was a no show today") == "Vendor No-Show" + + def test_weekly_wo_scheduled(self): + assert ( + classify.comment_intent( + "Weekly WO scheduled. Service reports required EOD Friday." + ) + == "Weekly WO Scheduled" + ) + + def test_avetta_project_created(self): + assert ( + classify.comment_intent("Avetta project created for this work") + == "Avetta Project Created" + ) + + def test_avetta_work_request(self): + assert ( + classify.comment_intent("Work request created in Avetta for this job") + == "Avetta Project Created" + ) + + def test_schedule_confirmed_with_vendor(self): + assert ( + classify.comment_intent("Schedule confirmed with vendor for next week") + == "Schedule Confirmed" + ) + + def test_schedule_confirmed_vendor_confirmed(self): + assert classify.comment_intent("Vendor confirmed") == "Schedule Confirmed" + + def test_completed_pending_close_performed_task(self): + assert ( + classify.comment_intent("Vendor arrived and performed task") + == "Completed / Pending Close" + ) + + def test_completed_pending_close_completed_by(self): + assert ( + classify.comment_intent("Completed by contractor on Monday") + == "Completed / Pending Close" + ) + + def test_completed_pending_close_cant_close(self): + assert ( + classify.comment_intent("Can't close the WO") == "Completed / Pending Close" + ) + + def test_completed_pending_close_pending_close(self): + assert classify.comment_intent("Pending close") == "Completed / Pending Close" + + def test_rescheduled(self): + assert ( + classify.comment_intent("WO rescheduled to next Tuesday") == "Rescheduled" + ) + + def test_rescheduled_new_eta(self): + assert classify.comment_intent("New ETA provided by vendor") == "Rescheduled" + + def test_report_docs_needed_service_report(self): + assert ( + classify.comment_intent("Please upload service report") + == "Report / Docs Needed" + ) + + def test_report_docs_needed_completion_report(self): + assert ( + classify.comment_intent("Completion report required") + == "Report / Docs Needed" + ) + + def test_awaiting_report_invoice_awaiting(self): + assert ( + classify.comment_intent("Awaiting the invoice from vendor") + == "Awaiting Report / Invoice" + ) + + def test_awaiting_report_invoice_pending(self): + assert ( + classify.comment_intent("Pending report from contractor") + == "Awaiting Report / Invoice" + ) + + def test_awaiting_scheduling_please_schedule(self): + assert ( + classify.comment_intent( + "Please schedule this WO at your earliest convenience" + ) + == "Awaiting Scheduling" + ) + + def test_status_inquiry_any_update(self): + assert classify.comment_intent("Any update on this WO?") == "Status Inquiry" + + def test_status_inquiry_eta(self): + assert ( + classify.comment_intent("Do you have an ETA on this?") == "Status Inquiry" + ) + + def test_on_hold(self): + assert ( + classify.comment_intent("WO is on hold pending budget approval") + == "On Hold" + ) + + def test_acknowledgement_copy(self): + assert classify.comment_intent("Copy") == "Acknowledgement / No-op" + + def test_acknowledgement_copy_with_date(self): + assert classify.comment_intent("Copy 5/26.") == "Acknowledgement / No-op" + + def test_acknowledgement_noted(self): + assert classify.comment_intent("Noted") == "Acknowledgement / No-op" + + +# --------------------------------------------------------------------------- +# Precedence / negative traps +# --------------------------------------------------------------------------- + + +class TestCommentIntentPrecedence: + def test_3rd_attempt_plus_report_language_resolves_to_3rd_escalation(self): + """A comment that mentions both '3rd attempt' and report language must + resolve to '3rd Escalation' — the most-specific bucket wins.""" + text = ( + "3rd attempt process for schedule confirmation. " + "Please provide service report." + ) + assert classify.comment_intent(text) == "3rd Escalation" + + def test_1st_attempt_for_schedule_confirmation_is_awaiting_scheduling(self): + """'1st attempt process for schedule confirmation' is routine outreach, + not an escalation. The ladder has an explicit Awaiting Scheduling rule + for this canonical phrase.""" + text = "1st attempt process for schedule confirmation" + assert classify.comment_intent(text) == "Awaiting Scheduling" + + def test_bare_1st_attempt_no_escalation_keyword_is_none(self): + """A bare '1st attempt to contact vendor' with no escalation keyword + must not fire any escalation bucket — the ladder requires an explicit + escalation signal beyond the mere ordinal.""" + text = "1st attempt to contact vendor" + assert classify.comment_intent(text) is None + + def test_empty_string_returns_none(self): + assert classify.comment_intent("") is None + + def test_whitespace_only_returns_none(self): + assert classify.comment_intent(" ") is None + + +# --------------------------------------------------------------------------- +# End-to-end precedence via classify() +# (strips HTML, then calls comment_intent, then applies structured-state axis) +# --------------------------------------------------------------------------- + + +class TestClassifyPrecedence: + def test_3rd_attempt_html_with_report_hold_resolves_to_3rd_escalation(self): + """Even when REPORT hold is set, a 3rd-attempt comment must resolve to + '3rd Escalation' because comment intent wins over structured state.""" + cat, mm = classify.classify( + "H", + "REPORT", + "3rd attempt process for schedule confirmation. " + "Please provide service report.", + ) + assert cat == "3rd Escalation" + # 3rd Escalation is not in _DONE_INTENTS, so no mismatch is produced. + assert mm is None + + def test_1st_attempt_schedule_confirmation_via_classify(self): + """End-to-end: HTML-wrapped '1st attempt process for schedule + confirmation' classifies as Awaiting Scheduling.""" + cat, _ = classify.classify( + "IP", + "", + "1st attempt process for schedule confirmation", + ) + assert cat == "Awaiting Scheduling" + + def test_comment_intent_wins_over_structured_state(self): + """When the comment fires a rule, it overrides the Hold Reason axis.""" + # Schedule Confirmed comment wins over SCHEDULING hold. + cat, _ = classify.classify( + "R", + "SCHEDULING", + "WO schedule confirmed with vendor", + ) + assert cat == "Schedule Confirmed" + + def test_structured_state_used_when_no_comment_intent(self): + """When no Axis-1 rule fires (blank comment), structured state decides.""" + cat, _ = classify.classify("IP", "REPORT", "") + assert cat == "Report / Docs Needed" + + def test_other_when_nothing_matches(self): + """Neither comment intent nor structured state → Other.""" + cat, _ = classify.classify("IP", "", "Uplift request submitted") + assert cat == "Other"