""" Tests for the Amazon ConsumeConsultation v2 response mapper These tests verify: 1. Identifiers are echoed exactly (API and legacy-SFTP shaped requests) 2. SUCCESS / FAILED payloads follow the v2 schema, including the ZIP pre-signed URL, per-medicine status and rejection reasons 3. Missing MANDATORY inputs raise ConsumeMappingError naming the field instead of shipping a payload Amazon will ERRORED """ import copy import sys import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from app.src.consume_mapper import ( ConsumeMappingError, rx_to_amazon_consume_response, ) PRIMARY = "404-1234567-1234567" SECONDARY = "8a7c6d5e-4f3b-2e1d-9c8b-7a6f5e4d3c2b" TRACKING = "f29e7d5a-b18c-43e2-95a7-6c21deb47f08" PROVIDER = "myrx" FILE_URL = "https://bucket.s3.ap-south-1.amazonaws.com/rx.zip?X-Amz-Signature=abc" SENT_AT = 1699968883231 APPROVED_AT = 1699968800000 ITEM_A = "CILIDIN_CILNIDIPINE#10.0#MILLIGRAMS###1_TABLETS" ITEM_B = "PARACET_PARACETAMOL#500.0#MILLIGRAMS###1_TABLETS" URL_A = "https://www.amazon.in/dp/B07X1ZM63M/" URL_B = "https://www.amazon.in/dp/B08ABCDEFG/" def api_request(**overrides): request = { "messageIdentifiers": { "primaryReferenceId": PRIMARY, "secondaryReferenceId": SECONDARY, "trackingId": "amazon-trk-1", "isTestData": False, }, "data": { "requestMetadata": {"messageId": "msg-1", "messageSentTime": 1699968000000}, "rxDetailForConsultation": [ { "rxHeaderDetailsForConsultation": { "rxId": "rx-1", "patientContactDetails": { "name": "Request Patient", "contactNumber": "+919999999999", }, }, "rxItemDetailForConsultation": [ {"rxItemId": ITEM_A, "orderedMedicine": URL_A}, {"rxItemId": ITEM_B, "orderedMedicine": URL_B}, ], } ], }, } request["messageIdentifiers"].update(overrides) return request def sftp_request(): return { "requestMetadata": { "referenceId": PRIMARY, "consultationId": SECONDARY, "messageId": "msg-1", }, "rxDetailForConsultation": [ { "rxHeaderDetailsForConsultation": {"rxId": "rx-1"}, "rxItemDetailForConsultation": [ {"rxItemId": ITEM_A, "orderedMedicine": URL_A} ], } ], } def medicine(item_id, uuid, **overrides): med = { "id": item_id, "uuid": uuid, "name": "Some medicine", "quantity": 10, "dosage": "1-0-1", "type": "TABLETS", "days": 5, "comments": "after food", } med.update(overrides) return med def approved_response(**overrides): response = { "status": "APPROVED", "pharmacyOrderId": PRIMARY, "shipmentCode": SECONDARY, "approveTime": APPROVED_AT, "orderedMedicines": [ {"uuid": "u-a", "url": URL_A}, {"uuid": "u-b", "url": URL_B}, ], "approvedMedicines": [medicine(ITEM_A, "u-a")], "rejectedMedicines": [], "doctor": { "name": "Dr. A. Sharma", "phone": "+91-9876543210", "registrationNumber": "MCI-12345", "address": "Bangalore, KA", }, "patient": {"name": "John Doe", "gender": "M", "age": 35}, } response.update(overrides) return response def build(rx_response, original_request=None, **kwargs): params = { "tracking_id": TRACKING, "provider_name": PROVIDER, "file_url": FILE_URL, "file_name": f"{SECONDARY}.jpg", "sent_time_millis": SENT_AT, } params.update(kwargs) return rx_to_amazon_consume_response( rx_response, original_request or api_request(), **params ) class TestMessageIdentifiers(unittest.TestCase): def test_echoes_identifiers_from_api_request(self): payload = build(approved_response()) self.assertEqual( payload["messageIdentifiers"], { "primaryReferenceId": PRIMARY, "secondaryReferenceId": SECONDARY, "trackingId": TRACKING, "providerName": PROVIDER, "isTestData": False, }, ) def test_echoes_test_data_flag(self): payload = build(approved_response(), api_request(isTestData=True)) self.assertIs(payload["messageIdentifiers"]["isTestData"], True) def test_omits_secondary_reference_when_amazon_did_not_send_one(self): request = api_request() del request["messageIdentifiers"]["secondaryReferenceId"] payload = build(approved_response(), request) self.assertNotIn("secondaryReferenceId", payload["messageIdentifiers"]) def test_reads_identifiers_from_legacy_sftp_request(self): payload = build(approved_response(), sftp_request()) ids = payload["messageIdentifiers"] self.assertEqual(ids["primaryReferenceId"], PRIMARY) self.assertEqual(ids["secondaryReferenceId"], SECONDARY) self.assertNotIn("isTestData", ids) def test_uses_our_delivery_tracking_id_not_amazons(self): payload = build(approved_response()) self.assertEqual(payload["messageIdentifiers"]["trackingId"], TRACKING) self.assertNotEqual(payload["messageIdentifiers"]["trackingId"], "amazon-trk-1") def test_missing_primary_reference_raises(self): request = api_request() del request["messageIdentifiers"]["primaryReferenceId"] with self.assertRaises(ConsumeMappingError) as ctx: build(approved_response(), request) self.assertEqual(ctx.exception.field, "messageIdentifiers.primaryReferenceId") def test_missing_tracking_id_or_provider_raises(self): with self.assertRaises(ConsumeMappingError): build(approved_response(), tracking_id="") with self.assertRaises(ConsumeMappingError): build(approved_response(), provider_name="") class TestSuccessPayload(unittest.TestCase): def test_metadata_and_consultation_response(self): payload = build(approved_response()) self.assertEqual( payload["data"]["responseMetadata"], {"messageSentTime": SENT_AT, "messageType": "CONSULTATION_COMPLETED"}, ) self.assertEqual( payload["data"]["consultationResponse"], {"status": "SUCCESS", "customerCallAttemptNumber": 1}, ) self.assertNotIn("failureReason", payload["data"]["consultationResponse"]) def test_call_attempt_number_is_passed_through(self): payload = build(approved_response(), call_attempt_number=3) self.assertEqual( payload["data"]["consultationResponse"]["customerCallAttemptNumber"], 3 ) def test_header_details(self): payload = build(approved_response()) detail = payload["data"]["rxDetailFromConsultation"][0] self.assertEqual(detail["rxIds"], ["rx-1"]) header = detail["rxHeaderDetailsFromConsultation"] self.assertEqual(header["rxIssueDate"], APPROVED_AT) self.assertEqual( header["prescriberDetails"], { "name": "Dr. A. Sharma", "contactNumber": "+91-9876543210", "registrationNumber": "MCI-12345", "address": "Bangalore, KA", }, ) self.assertEqual( header["patientDetails"], {"name": "John Doe", "gender": "M", "age": 35} ) def test_issue_date_falls_back_to_sent_time_without_approve_time(self): payload = build(approved_response(approveTime=None)) header = payload["data"]["rxDetailFromConsultation"][0][ "rxHeaderDetailsFromConsultation" ] self.assertEqual(header["rxIssueDate"], SENT_AT) def test_approved_item(self): payload = build(approved_response()) items = payload["data"]["rxDetailFromConsultation"][0][ "rxItemDetailFromConsultation" ] self.assertEqual(len(items), 1) self.assertEqual( items[0], { "rxItemId": ITEM_A, "rxItemIssueDate": APPROVED_AT, "orderedMedicine": URL_A, "medicineName": "Some medicine", "totalPrescribedQty": 10, "dosage": { "dose": 1, "form": "TABLETS", "frequency": "2", "frequencyUnit": "DAILY", "duration": "5", "durationUnit": "DAYS", }, "itemStatus": "APPROVED", "instructions": "after food", "isEligibleForFutureConsultations": True, }, ) self.assertNotIn("numberOfUnitsOrdered", items[0]) def test_prescription_zip_image_details(self): payload = build(approved_response()) images = payload["data"]["rxDetailFromConsultation"][0]["rxImageDetails"] self.assertEqual( images, [ { "fileName": f"{SECONDARY}.jpg", "formatType": "ZIP", "rxIndex": "1", "fileUrl": FILE_URL, } ], ) def test_mixed_approved_and_rejected_medicines(self): response = approved_response( rejectedMedicines=[ medicine(ITEM_B, "u-b", rejectionReason="Physical visit required") ] ) payload = build(response) items = payload["data"]["rxDetailFromConsultation"][0][ "rxItemDetailFromConsultation" ] self.assertEqual([i["itemStatus"] for i in items], ["APPROVED", "REJECTED"]) rejected = items[1] self.assertEqual(rejected["rxItemId"], ITEM_B) self.assertEqual(rejected["orderedMedicine"], URL_B) self.assertEqual(rejected["rejectionReason"], ["MANUAL_VISIT_REQUIRED"]) self.assertIs(rejected["isEligibleForFutureConsultations"], False) self.assertNotIn("rejectionReason", items[0]) def test_unmapped_rejection_reason_becomes_others_and_stays_eligible(self): response = approved_response( rejectedMedicines=[medicine(ITEM_B, "u-b", rejectionReason="Out of stock")] ) payload = build(response) rejected = payload["data"]["rxDetailFromConsultation"][0][ "rxItemDetailFromConsultation" ][1] self.assertEqual(rejected["rejectionReason"], ["OTHERS"]) self.assertIs(rejected["isEligibleForFutureConsultations"], True) def test_whole_order_rejected_without_medicine_entries(self): response = approved_response( status="REJECTED", orderCancelReason="Medicine not permitted", approvedMedicines=[], rejectedMedicines=[], orderedMedicines=[medicine(ITEM_A, "u-a", url=URL_A)], ) payload = build(response, file_url=None) self.assertEqual(payload["data"]["consultationResponse"]["status"], "SUCCESS") detail = payload["data"]["rxDetailFromConsultation"][0] self.assertEqual(detail["rxImageDetails"], []) items = detail["rxItemDetailFromConsultation"] self.assertEqual(len(items), 1) self.assertEqual(items[0]["itemStatus"], "REJECTED") self.assertEqual(items[0]["rejectionReason"], ["SUSPECTED_ABUSE"]) self.assertIs(items[0]["isEligibleForFutureConsultations"], False) def test_ordered_medicine_url_falls_back_to_amazon_request(self): response = approved_response(orderedMedicines=[]) response["approvedMedicines"][0].pop("uuid") payload = build(response) item = payload["data"]["rxDetailFromConsultation"][0][ "rxItemDetailFromConsultation" ][0] self.assertEqual(item["orderedMedicine"], URL_A) def test_patient_name_falls_back_to_amazon_request(self): payload = build(approved_response(patient={"age": 40})) header = payload["data"]["rxDetailFromConsultation"][0][ "rxHeaderDetailsFromConsultation" ] self.assertEqual(header["patientDetails"], {"name": "Request Patient", "age": 40}) def test_optional_medicine_fields_omitted_when_absent(self): response = approved_response( approvedMedicines=[medicine(ITEM_A, "u-a", name=None, comments=None)] ) payload = build(response) item = payload["data"]["rxDetailFromConsultation"][0][ "rxItemDetailFromConsultation" ][0] self.assertNotIn("medicineName", item) self.assertNotIn("instructions", item) class TestFailedPayload(unittest.TestCase): def failed_response(self, reason="Call not answered", status="CANCELLED"): return approved_response( status=status, orderCancelReason=reason, approvedMedicines=[], doctor={}, patient={}, ) def test_failed_consultation_has_reason_and_no_rx_details(self): payload = build(self.failed_response(), file_url=None) self.assertEqual( payload["data"]["consultationResponse"], { "status": "FAILED", "customerCallAttemptNumber": 1, "failureReason": "USER_NOT_ANSWERING_CALL", }, ) self.assertNotIn("rxDetailFromConsultation", payload["data"]) def test_failure_reason_mapping(self): cases = { "Patient requested to call later": "USER_ASKED_TO_CALL_LATER", "Doctor is out of network": "NETWORK_CONNECTIVITY_ISSUES", "Language barrier during consultation": "UNABLE_TO_UNDERSTAND_USER", "Patient wants to cancel the order": "USER_ASKED_TO_CANCEL_ORDER", "Wrong number provided": "INCORRECT_NUMBER", "Something new": "USER_NOT_REACHABLE", None: "USER_NOT_REACHABLE", } for reason, expected in cases.items(): with self.subTest(reason=reason): payload = build(self.failed_response(reason=reason), file_url=None) self.assertEqual( payload["data"]["consultationResponse"]["failureReason"], expected ) def test_failed_does_not_need_doctor_or_file(self): payload = build(self.failed_response(status="FAILED"), file_url=None) self.assertEqual(payload["data"]["consultationResponse"]["status"], "FAILED") class TestMandatoryFieldEnforcement(unittest.TestCase): def assert_missing(self, rx_response, field, **kwargs): with self.assertRaises(ConsumeMappingError) as ctx: build(rx_response, **kwargs) self.assertEqual(ctx.exception.field, field) def test_each_prescriber_field_is_mandatory(self): for key, field in ( ("name", "prescriberDetails.name"), ("phone", "prescriberDetails.contactNumber"), ("registrationNumber", "prescriberDetails.registrationNumber"), ("address", "prescriberDetails.address"), ): with self.subTest(field=field): response = approved_response() response["doctor"][key] = " " self.assert_missing(response, field) def test_missing_doctor_object(self): self.assert_missing(approved_response(doctor=None), "prescriberDetails.name") def test_patient_name_mandatory_when_neither_side_has_it(self): request = api_request() del request["data"]["rxDetailForConsultation"][0][ "rxHeaderDetailsForConsultation" ]["patientContactDetails"] with self.assertRaises(ConsumeMappingError) as ctx: build(approved_response(patient={}), request) self.assertEqual(ctx.exception.field, "patientDetails.name") def test_file_url_mandatory_when_approved(self): self.assert_missing(approved_response(), "rxImageDetails.fileUrl", file_url=None) def test_rx_ids_mandatory(self): request = api_request() del request["data"]["rxDetailForConsultation"][0][ "rxHeaderDetailsForConsultation" ]["rxId"] with self.assertRaises(ConsumeMappingError) as ctx: build(approved_response(), request) self.assertEqual(ctx.exception.field, "rxDetailFromConsultation.rxIds") def test_medicine_without_id(self): response = approved_response(approvedMedicines=[medicine(None, "u-a")]) self.assert_missing(response, "rxItemDetailFromConsultation.rxItemId") def test_medicine_without_any_url_source(self): response = approved_response( orderedMedicines=[], approvedMedicines=[medicine("UNKNOWN_ITEM", "u-x")], ) self.assert_missing( response, "rxItemDetailFromConsultation[UNKNOWN_ITEM].orderedMedicine" ) def test_medicine_without_quantity(self): response = approved_response(approvedMedicines=[medicine(ITEM_A, "u-a", quantity=None)]) self.assert_missing( response, f"rxItemDetailFromConsultation[{ITEM_A}].totalPrescribedQty" ) def test_success_without_any_medicines(self): response = approved_response(approvedMedicines=[], rejectedMedicines=[]) self.assert_missing(response, "rxItemDetailFromConsultation") def test_error_message_names_the_field(self): response = approved_response() response["doctor"]["address"] = None with self.assertRaises(ConsumeMappingError) as ctx: build(response) self.assertIn("prescriberDetails.address", str(ctx.exception)) def test_inputs_are_not_mutated(self): response = approved_response() request = api_request() response_copy, request_copy = copy.deepcopy(response), copy.deepcopy(request) build(response, request) self.assertEqual(response, response_copy) self.assertEqual(request, request_copy) if __name__ == "__main__": unittest.main()