Async Tests

pyas2lib/tests/test_advanced.py

@@ -9,8 +9,6 @@
 from pyas2lib.exceptions import ImproperlyConfigured
 from pyas2lib.tests import Pyas2TestCase, TEST_DIR
 
-import asyncio
-
 
 class TestAdvanced(Pyas2TestCase):
     def setUp(self):
@@ -517,57 +515,6 @@ def test_final_recipient_fallback(self):
 
         self.assertEqual(message_recipient, self.partner.as2_name)
 
-    @pytest.mark.asyncio
-    async def test_duplicate_message_async(self):
-        """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
-        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 = await in_message.parse(
-            raw_out_message,
-            find_org_cb=self.afind_org,
-            find_partner_cb=self.afind_partner,
-            find_message_cb=self.afind_duplicate_message,
-        )
-
-        out_mdn = as2.Mdn()
-        status, detailed_status = await out_mdn.parse(
-            mdn.headers_str + b"\r\n" + mdn.content, find_message_cb=self.afind_message
-        )
-        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
 
@@ -577,18 +524,6 @@ def find_partner(self, headers):
     def find_message(self, message_id, message_recipient):
         return self.out_message
 
-    async def afind_org(self, headers):
-        return self.org
-
-    async def afind_partner(self, headers):
-        return self.partner
-
-    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):

pyas2lib/tests/test_async.py

@@ -0,0 +1,92 @@
+import pytest
+from pyas2lib import as2
+import os
+
+from pyas2lib.tests import TEST_DIR
+
+with open(os.path.join(TEST_DIR, "payload.txt"), "rb") as fp:
+    test_data = fp.read()
+
+with open(os.path.join(TEST_DIR, "cert_test.p12"), "rb") as fp:
+    private_key = fp.read()
+
+with open(os.path.join(TEST_DIR, "cert_test_public.pem"), "rb") as fp:
+    public_key = fp.read()
+
+org = as2.Organization(
+    as2_name="some_organization",
+    sign_key=private_key,
+    sign_key_pass="test",
+    decrypt_key=private_key,
+    decrypt_key_pass="test",
+)
+partner = as2.Partner(
+    as2_name="some_partner",
+    verify_cert=public_key,
+    encrypt_cert=public_key,
+)
+
+
+async def afind_org(headers):
+    return org
+
+
+async def afind_partner(headers):
+    return partner
+
+
+async def afind_duplicate_message(message_id, message_recipient):
+    return True
+
+
+async def afind_org_partner(as2_org, as2_partner):
+    return org, partner
+
+
+@pytest.mark.asyncio
+async def test_duplicate_message_async():
+    """Test case where a duplicate message is sent to the partner using async callbacks"""
+
+    # Build an As2 message to be transmitted to partner
+    partner.sign = True
+    partner.encrypt = True
+    partner.mdn_mode = as2.SYNCHRONOUS_MDN
+    out_message = as2.Message(org, partner)
+    out_message.build(test_data)
+
+    # Parse the generated AS2 message as the partner
+    raw_out_message = out_message.headers_str + b"\r\n" + out_message.content
+    in_message = as2.Message()
+    _, _, mdn = await in_message.aparse(
+        raw_out_message,
+        find_org_cb=afind_org,
+        find_partner_cb=afind_partner,
+        find_message_cb=afind_duplicate_message,
+    )
+
+    out_mdn = as2.Mdn()
+    status, detailed_status = await out_mdn.aparse(
+        mdn.headers_str + b"\r\n" + mdn.content,
+        find_message_cb=lambda x, y: out_message,
+    )
+    assert status == "processed/Warning"
+    assert detailed_status == "duplicate-document"
+
+
+@pytest.mark.asyncio
+async def test_async_partnership():
+    """Test Async Partnership callback"""
+
+    # Build an As2 message to be transmitted to partner
+    out_message = as2.Message(org, partner)
+    out_message.build(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, _, _ = await in_message.aparse(
+        raw_out_message, find_org_partner_cb=afind_org_partner
+    )
+
+    # Compare contents of the input and output messages
+    assert status == "processed"