From 23a8c395f23ba30e5a2dd58fc2f94ed40d7a36a3 Mon Sep 17 00:00:00 2001 From: Wassilios Lytras Date: Mon, 18 Mar 2024 20:25:21 +0100 Subject: [PATCH] Async Tests --- pyas2lib/tests/test_advanced.py | 65 ----------------------- pyas2lib/tests/test_async.py | 92 +++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 65 deletions(-) create mode 100644 pyas2lib/tests/test_async.py diff --git a/pyas2lib/tests/test_advanced.py b/pyas2lib/tests/test_advanced.py index 490b74a..0bc6db7 100644 --- a/pyas2lib/tests/test_advanced.py +++ b/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): diff --git a/pyas2lib/tests/test_async.py b/pyas2lib/tests/test_async.py new file mode 100644 index 0000000..57acb66 --- /dev/null +++ b/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"