Adding Async Support for Callbacks

pyas2lib/tests/test_advanced.py

@@ -518,7 +518,7 @@ def test_final_recipient_fallback(self):
 
     @pytest.mark.asyncio
     async def test_duplicate_message_async(self):
-        """Test case where a duplicate message is sent to the partner asynchronously"""
+        """Test case where a duplicate message is sent to the partner using async callbacks"""
 
         # Build an As2 message to be transmitted to partner
         self.partner.sign = True
@@ -546,6 +546,28 @@ async def test_duplicate_message_async(self):
         self.assertEqual(status, "processed/Warning")
         self.assertEqual(detailed_status, "duplicate-document")
 
+    @pytest.mark.asyncio
+    async def test_async_partnership(self):
+        """Test Async Partnership callback with Unencrypted Unsigned Uncompressed Message"""
+
+        # Build an As2 message to be transmitted to partner
+        out_message = as2.Message(self.org, self.partner)
+        out_message.build(self.test_data)
+        raw_out_message = out_message.headers_str + b"\r\n" + out_message.content
+
+        # Parse the generated AS2 message as the partner
+        in_message = as2.Message()
+        status, _, _ = in_message.parse(
+            raw_out_message,
+            find_org_cb=self.find_org,
+            find_partner_cb=self.afind_org_partner,
+        )
+
+        # Compare contents of the input and output messages
+        self.assertEqual(status, "processed")
+        self.assertEqual(self.test_data, in_message.content)
+
+
     def find_org(self, headers):
         return self.org
 
@@ -564,7 +586,8 @@ async def afind_partner(self, headers):
     async def afind_duplicate_message(self, message_id, message_recipient):
         return True
 
-
+    async def afind_org_partner(self, as2_org, as2_partner):
+        return self.org, self.partner
 class SterlingIntegratorTest(Pyas2TestCase):
     def setUp(self):
         self.org = as2.Organization(

pyas2lib/tests/test_basic.py

@@ -183,6 +183,30 @@ 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_encrypted_signed_message_partnership(self):
+        """Test Encrypted Signed Uncompressed Message with Partnership"""
+
+        # Build an As2 message to be transmitted to partner
+        self.partner.sign = True
+        self.partner.encrypt = True
+        out_message = as2.Message(self.org, self.partner)
+        out_message.build(self.test_data)
+        raw_out_message = out_message.headers_str + b"\r\n" + out_message.content
+
+        # Parse the generated AS2 message as the partner
+        in_message = as2.Message()
+        status, _, _ = in_message.parse(
+            raw_out_message,
+            find_org_partner_cb=self.find_org_partner,
+        )
+
+        # Compare the mic contents of the input and output messages
+        self.assertEqual(status, "processed")
+        self.assertTrue(in_message.signed)
+        self.assertTrue(in_message.encrypted)
+        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"""
 
@@ -205,3 +229,6 @@ def find_org(self, as2_id):
 
     def find_partner(self, as2_id):
         return self.partner
+
+    def find_org_partner(self, as2_org, as2_partner):
+        return self.org, self.partner