apm-wo-analysis/tests/test_comment_intent.py

"""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",
            "<html>3rd attempt process for schedule confirmation. "
            "Please provide service report.</html>",
        )
        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",
            "",
            "<html>1st attempt process for schedule confirmation</html>",
        )
        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",
            "<html>WO schedule confirmed with vendor</html>",
        )
        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", "<html></html>")
        assert cat == "Report / Docs Needed"

    def test_other_when_nothing_matches(self):
        """Neither comment intent nor structured state → Other."""
        cat, _ = classify.classify("IP", "", "<html>Uplift request submitted</html>")
        assert cat == "Other"