email-connect/tests/test_parser.py

from __future__ import annotations

import unittest
from pathlib import Path

from email_connect.models import MessageClass
from email_connect.parser import parse_message_file


FIXTURES = Path(__file__).parent / "fixtures" / "mailbox"


class ParserTests(unittest.TestCase):
    def test_hard_bounce_maps_to_permanent_rejection(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "hard_bounce.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.HARD_BOUNCE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "notification.endpoint.rejected_permanent")
        self.assertEqual(candidate.assessment_subclass, "fail.hard_bounce")

    def test_soft_bounce_maps_to_temporary_rejection(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "soft_bounce.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.SOFT_BOUNCE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "notification.endpoint.rejected_temporary")

    def test_out_of_office_stays_undef(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "out_of_office.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.OUT_OF_OFFICE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.assessment_category.value, "undef")
        self.assertEqual(candidate.event_type, "interaction.out_of_office_received")

    def test_human_reply_is_email_channel_success_only(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "human_reply.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.HUMAN_REPLY)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "interaction.reply_received")
        self.assertEqual(candidate.assessment_subclass, "success.reply_received")

    def test_delayed_delivery_notice_stays_deferred(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "delayed_delivery.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.DELAYED_DELIVERY_NOTICE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "notification.endpoint.deferred")
        self.assertEqual(candidate.assessment_subclass, "undef.deferred")

    def test_final_failure_maps_to_expired_without_delivery(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "final_failure.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.FINAL_DELIVERY_FAILURE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "notification.endpoint.rejected_permanent")
        self.assertEqual(candidate.assessment_subclass, "fail.expired_without_delivery")

    def test_complaint_maps_to_channel_failure(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "complaint.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.COMPLAINT_OR_ABUSE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "notification.channel.complaint_received")
        self.assertEqual(candidate.assessment_subclass, "fail.complaint_received")

    def test_unsubscribe_maps_to_opt_out(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "unsubscribe.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.UNSUBSCRIBE_OR_OPT_OUT)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "notification.channel.unsubscribe_received")
        self.assertEqual(candidate.assessment_subclass, "fail.unsubscribed")

    def test_unknown_return_message_is_preserved(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "unknown_return.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.UNKNOWN_RETURN_MESSAGE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "notification.endpoint.unknown")

    def test_challenge_response_stays_identity_uncertain(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "challenge_response.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.CHALLENGE_RESPONSE)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "interaction.unverified_actor_interaction")
        self.assertEqual(candidate.assessment_subclass, "undef.identity_uncertain")

    def test_parse_failure_is_reportable_diagnostic(self) -> None:
        _inbound, parsed, candidate = parse_message_file(FIXTURES / "parse_failed.eml", mailbox_id="test")
        self.assertEqual(parsed.message_class, MessageClass.PARSE_FAILED)
        self.assertIsNotNone(candidate)
        self.assertEqual(candidate.event_type, "diagnostic.message.parse_failed")
        self.assertEqual(candidate.assessment_subclass, "undef.parse_failed")

    def test_dsn_detail_fields_are_preserved_as_notes(self) -> None:
        _inbound, parsed, _candidate = parse_message_file(FIXTURES / "hard_bounce.eml", mailbox_id="test")
        self.assertIn("final_recipient=rfc822; missing@example.com", parsed.notes)
        self.assertIn("action=failed", parsed.notes)
        self.assertIn("diagnostic_code=smtp; 550 5.1.1 User unknown", parsed.notes)

    def test_mta_received_headers_do_not_make_a_message_a_reply(self) -> None:
        """Regression: the reply heuristic used to match the Received: trace.

        Every message that traverses an MTA carries `Received:` headers, and the
        word "received" was enough to classify ordinary mail as a human reply
        with a success.reply_received assessment. Hand-written fixtures have no
        Received headers, so only real scanned mail exposed it.
        """

        transit = Path(__file__).parent / "fixtures" / "mailbox_transit" / "ordinary_transit.eml"
        _inbound, parsed, candidate = parse_message_file(transit, mailbox_id="test")
        self.assertNotEqual(parsed.message_class, MessageClass.HUMAN_REPLY)
        if candidate is not None:
            self.assertNotEqual(candidate.event_type, "interaction.reply_received")


if __name__ == "__main__":
    unittest.main()