Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature/message not found #69

Merged
merged 10 commits into from
May 3, 2024
5 changes: 5 additions & 0 deletions pyas2lib/as2.py
Original file line number Diff line number Diff line change
Expand Up @@ -943,6 +943,11 @@ def parse(self, raw_content, find_message_cb):
# Call the find message callback which should return a Message instance
orig_message = find_message_cb(self.orig_message_id, orig_recipient)

if not orig_message:
status = "failed/Failure"
details_status = "original-message-not-found"
return status, details_status

# Extract the headers and save it
mdn_headers = {}
for k, v in self.payload.items():
Expand Down
27 changes: 27 additions & 0 deletions pyas2lib/tests/test_advanced.py
Original file line number Diff line number Diff line change
Expand Up @@ -396,6 +396,33 @@ def test_mdn_not_found(self):
self.assertEqual(status, "failed/Failure")
self.assertEqual(detailed_status, "mdn-not-found")

def test_mdn_original_message_not_found(self):
"""Test that the MDN parser raises MDN not found when a non MDN message is passed."""
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,
find_message_cb=lambda x, y: False,
)

# Parse the MDN
out_mdn = as2.Mdn()
status, detailed_status = out_mdn.parse(
mdn.headers_str + b"\r\n" + mdn.content, find_message_cb=lambda x, y: False
)

self.assertEqual(status, "failed/Failure")
self.assertEqual(detailed_status, "original-message-not-found")

def test_unsigned_mdn_sent_error(self):
"""Test the case where a signed mdn was expected but unsigned mdn was returned."""
self.partner.mdn_mode = as2.SYNCHRONOUS_MDN
Expand Down
Loading