feat: option to pass custom domain for message-id generation, unit te…

…sts. (#51)

pyas2lib/as2.py

@@ -74,6 +74,10 @@ class Organization:
 
     :param mdn_url: The URL where the receiver is expected to post
         asynchronous MDNs.
+
+    :param domain:
+        Optional domain if given provides the portion of the message id
+        after the '@'.  It defaults to the locally defined hostname.
     """
 
     as2_name: str
@@ -83,6 +87,7 @@ class Organization:
     decrypt_key_pass: str = None
     mdn_url: str = None
     mdn_confirm_text: str = MDN_CONFIRM_TEXT
+    domain: str = None
 
     def __post_init__(self):
         """Run the post initialisation checks for this class."""
@@ -346,7 +351,6 @@ def build(
         :param disposition_notification_to:
             Email address for disposition-notification-to header entry.
             (default "[email protected]")
-
         """
 
         # Validations
@@ -366,7 +370,9 @@ def build(
             )
 
         # Generate message id using UUID 1 as it uses both hostname and time
-        self.message_id = email_utils.make_msgid().lstrip("<").rstrip(">")
+        self.message_id = (
+            email_utils.make_msgid(domain=self.sender.domain).lstrip("<").rstrip(">")
+        )
 
         # Set up the message headers
         as2_headers = {
@@ -749,7 +755,8 @@ def build(
         """
 
         # Generate message id using UUID 1 as it uses both hostname and time
-        self.message_id = email_utils.make_msgid().lstrip("<").rstrip(">")
+        domain = message.receiver.domain if message.receiver else None
+        self.message_id = email_utils.make_msgid(domain=domain).lstrip("<").rstrip(">")
         self.orig_message_id = message.message_id
 
         # Set up the message headers

pyas2lib/tests/test_basic.py

@@ -1,5 +1,5 @@
 """Module for testing the basic features of pyas2."""
-
+import socket
 from pyas2lib import as2
 from . import Pyas2TestCase
 
@@ -183,6 +183,23 @@ def test_encrypted_signed_compressed_message(self):
         self.assertEqual(out_message.mic, in_message.mic)
         self.assertEqual(self.test_data.splitlines(), in_message.content.splitlines())
 
+    def test_plain_message_with_domain(self):
+        """Test Message building with an org domain"""
+
+        # Build an As2 message to be transmitted to partner
+        self.org.domain = "example.com"
+        out_message = as2.Message(self.org, self.partner)
+        out_message.build(self.test_data)
+        self.assertEqual(out_message.message_id.split("@")[1], self.org.domain)
+
+    def test_plain_message_without_domain(self):
+        """Test Message building without an org domain"""
+
+        # Build an As2 message to be transmitted to partner
+        out_message = as2.Message(self.org, self.partner)
+        out_message.build(self.test_data)
+        self.assertEqual(out_message.message_id.split("@")[1], socket.getfqdn())
+
     def find_org(self, as2_id):
         return self.org
 

pyas2lib/tests/test_mdn.py

@@ -1,4 +1,5 @@
 """Module for testing the MDN related features of pyas2lib"""
+import socket
 from pyas2lib import as2
 from . import Pyas2TestCase
 
@@ -110,6 +111,63 @@ def test_failed_mdn_parse(self):
             "Message Digest does not match.",
         )
 
+    def test_mdn_with_domain(self):
+        """Test MDN generation with an org domain"""
+        self.org.domain = "example.com"
+
+        # Build an As2 message to be transmitted to partner
+        self.partner.sign = True
+        self.partner.encrypt = True
+        self.partner.mdn_mode = as2.SYNCHRONOUS_MDN
+        self.out_message = as2.Message(self.org, self.partner)
+        self.out_message.build(self.test_data)
+
+        # Parse the generated AS2 message as the partner
+        raw_out_message = (
+            self.out_message.headers_str + b"\r\n" + self.out_message.content
+        )
+        in_message = as2.Message()
+        _, _, mdn = in_message.parse(
+            raw_out_message,
+            find_org_cb=self.find_org,
+            find_partner_cb=self.find_partner,
+        )
+
+        out_mdn = as2.Mdn()
+        status, detailed_status = out_mdn.parse(
+            mdn.headers_str + b"\r\n" + mdn.content, find_message_cb=self.find_message
+        )
+
+        self.assertEqual(out_mdn.message_id.split("@")[1], self.org.domain)
+
+    def test_mdn_without_domain(self):
+        """Test MDN generation without an org domain"""
+
+        # Build an As2 message to be transmitted to partner
+        self.partner.sign = True
+        self.partner.encrypt = True
+        self.partner.mdn_mode = as2.SYNCHRONOUS_MDN
+        self.out_message = as2.Message(self.org, self.partner)
+        self.out_message.build(self.test_data)
+
+        # Parse the generated AS2 message as the partner
+        raw_out_message = (
+            self.out_message.headers_str + b"\r\n" + self.out_message.content
+        )
+        in_message = as2.Message()
+        _, _, mdn = in_message.parse(
+            raw_out_message,
+            find_org_cb=self.find_org,
+            find_partner_cb=self.find_partner,
+        )
+
+        out_mdn = as2.Mdn()
+        status, detailed_status = out_mdn.parse(
+            mdn.headers_str + b"\r\n" + mdn.content, find_message_cb=self.find_message
+        )
+
+        self.assertEqual(out_mdn.message_id.split("@")[1], socket.getfqdn())
+
     def find_org(self, as2_id):
         return self.org