VED-734 Handle multiple file events in lambda forwarder (#758)

Author: dlzhry2nhs

.github/workflows/sonarcloud.yml

@@ -60,7 +60,7 @@ jobs:
         continue-on-error: true
         run: |
           poetry install
-          poetry run coverage run -m unittest discover -s "./tests" -p "*batch*.py" || echo "recordforwarder tests failed" >> ../failed_tests.txt
+          poetry run coverage run -m unittest discover -p "*batch*.py" || echo "recordforwarder tests failed" >> ../failed_tests.txt
           poetry run coverage xml -o ../recordforwarder-coverage.xml
 
       - name: Run unittest with coverage-ack-lambda

backend/src/batch/__init__.py

@@ -0,0 +1,31 @@
+import copy
+
+from models.errors import MessageNotSuccessfulError
+
+
+class BatchFilenameToEventsMapper:
+    FILENAME_NOT_PRESENT_ERROR_MSG = "Filename data was not present"
+
+    def __init__(self):
+        self._filename_to_events_map: dict[str, list[dict]] = {}
+
+    def add_event(self, event: dict) -> None:
+        filename_key = self._make_key(event)
+
+        if filename_key not in self._filename_to_events_map:
+            self._filename_to_events_map[filename_key] = [event]
+            return
+
+        self._filename_to_events_map[filename_key].append(event)
+
+    def get_map(self) -> dict[str, list[dict]]:
+        return copy.deepcopy(self._filename_to_events_map)
+
+    def _make_key(self, event: dict) -> str:
+        file_key = event.get("file_key")
+        created_at_string = event.get("created_at_formatted_string")
+
+        if not file_key or not created_at_string:
+            raise MessageNotSuccessfulError(self.FILENAME_NOT_PRESENT_ERROR_MSG)
+
+        return f"{file_key}_{created_at_string}"

backend/src/batch/batch_filename_to_events_mapper.py

@@ -5,6 +5,8 @@
 import base64
 import time
 import logging
+
+from batch.batch_filename_to_events_mapper import BatchFilenameToEventsMapper
 from fhir_batch_repository import create_table
 from fhir_batch_controller import ImmunizationBatchController, make_batch_controller
 from clients import sqs_client
@@ -59,7 +61,7 @@ def forward_lambda_handler(event, _):
     """Forward each row to the Imms API"""
     logger.info("Processing started")
     table = create_table()
-    array_of_messages = []
+    filename_to_events_mapper = BatchFilenameToEventsMapper()
     array_of_identifiers = []
     controller = make_batch_controller()
 
@@ -101,23 +103,20 @@ def forward_lambda_handler(event, _):
                 array_of_identifiers.append(identifier)
 
             imms_id = forward_request_to_dynamo(incoming_message_body, table, identifier_already_present, controller)
-            array_of_messages.append({**base_outgoing_message_body, "imms_id": imms_id})
+            filename_to_events_mapper.add_event({**base_outgoing_message_body, "imms_id": imms_id})
 
         except Exception as error:  # pylint: disable = broad-exception-caught
-            array_of_messages.append(
+            filename_to_events_mapper.add_event(
                 {**base_outgoing_message_body, "diagnostics": create_diagnostics_dictionary(error)}
             )
             logger.error("Error processing message: %s", error)
 
     # Send to SQS
-    sqs_message_body = json.dumps(array_of_messages)
-    message_len = len(sqs_message_body)
-    logger.info(f"total message length:{message_len}")
-    message_group_id = f"{file_key}_{created_at_formatted_string}"
-    if message_len < 256 * 1024:
-        sqs_client.send_message(QueueUrl=QUEUE_URL, MessageBody=sqs_message_body, MessageGroupId=message_group_id)
-    else:
-        logger.info("Message size exceeds 256 KB limit.Sending to sqs failed")
+    for filename_key, events in filename_to_events_mapper.get_map().items():
+        sqs_message_body = json.dumps(events)
+        logger.info(f"total message length:{len(sqs_message_body)}")
+
+        sqs_client.send_message(QueueUrl=QUEUE_URL, MessageBody=sqs_message_body, MessageGroupId=filename_key)
 
 
 if __name__ == "__main__":

backend/src/forwarding_batch_lambda.py

@@ -0,0 +1,99 @@
+import unittest
+
+from batch.batch_filename_to_events_mapper import BatchFilenameToEventsMapper
+from models.errors import MessageNotSuccessfulError
+
+MOCK_SUPPLIER_ONE_RSV_EVENT_ONE = {
+    "row_id": "row-1",
+    "file_key": "supplier_one_rsv_file_key",
+    "created_at_formatted_string": "2025-01-24T12:00:00Z",
+    "supplier": "supplier_one",
+    "vax_type": "RSV",
+    "local_id": "local-1",
+    "operation_requested": "CREATE"
+}
+
+MOCK_SUPPLIER_ONE_RSV_EVENT_TWO = {
+    "row_id": "row-2",
+    "file_key": "supplier_one_rsv_file_key",
+    "created_at_formatted_string": "2025-01-24T12:00:00Z",
+    "supplier": "supplier_one",
+    "vax_type": "RSV",
+    "local_id": "local-2",
+    "operation_requested": "UPDATE"
+}
+
+MOCK_SUPPLIER_TWO_COVID_EVENT_ONE = {
+    "row_id": "row-1",
+    "file_key": "supplier_two_covid19_file_key",
+    "created_at_formatted_string": "2025-01-24T14:00:00Z",
+    "supplier": "supplier_two",
+    "vax_type": "COVID-19",
+    "local_id": "local-1",
+    "operation_requested": "CREATE"
+}
+
+
+class TestBatchFilenameToEventsMapper(unittest.TestCase):
+    expected_key_supplier_one = "supplier_one_rsv_file_key_2025-01-24T12:00:00Z"
+    expected_key_supplier_two = "supplier_two_covid19_file_key_2025-01-24T14:00:00Z"
+
+    def setUp(self) -> None:
+        self.batch_filename_to_events_mapper = BatchFilenameToEventsMapper()
+
+    def test_add_event_creates_new_key(self):
+        self.batch_filename_to_events_mapper.add_event(MOCK_SUPPLIER_ONE_RSV_EVENT_ONE)
+
+        result = self.batch_filename_to_events_mapper.get_map()
+
+        self.assertIn(self.expected_key_supplier_one, result)
+        self.assertEqual(result[self.expected_key_supplier_one], [MOCK_SUPPLIER_ONE_RSV_EVENT_ONE])
+
+    def test_add_event_appends_to_existing_key(self):
+        self.batch_filename_to_events_mapper.add_event(MOCK_SUPPLIER_ONE_RSV_EVENT_ONE)
+        self.batch_filename_to_events_mapper.add_event(MOCK_SUPPLIER_ONE_RSV_EVENT_TWO)
+
+        result = self.batch_filename_to_events_mapper.get_map()
+
+        self.assertIn(self.expected_key_supplier_one, result)
+        self.assertEqual(result[self.expected_key_supplier_one], [
+            MOCK_SUPPLIER_ONE_RSV_EVENT_ONE,
+            MOCK_SUPPLIER_ONE_RSV_EVENT_TWO
+        ])
+
+    def test_mapper_handles_events_from_multiple_files(self):
+        self.batch_filename_to_events_mapper.add_event(MOCK_SUPPLIER_ONE_RSV_EVENT_ONE)
+        self.batch_filename_to_events_mapper.add_event(MOCK_SUPPLIER_ONE_RSV_EVENT_TWO)
+        self.batch_filename_to_events_mapper.add_event(MOCK_SUPPLIER_TWO_COVID_EVENT_ONE)
+
+        result = self.batch_filename_to_events_mapper.get_map()
+
+        self.assertEqual(len(result.keys()), 2)
+        self.assertIn(self.expected_key_supplier_one, result)
+        self.assertEqual(result[self.expected_key_supplier_one], [
+            MOCK_SUPPLIER_ONE_RSV_EVENT_ONE,
+            MOCK_SUPPLIER_ONE_RSV_EVENT_TWO
+        ])
+        self.assertIn(self.expected_key_supplier_two, result)
+        self.assertEqual(result[self.expected_key_supplier_two], [MOCK_SUPPLIER_TWO_COVID_EVENT_ONE])
+
+    def test_event_with_missing_filename_data_raises_error(self):
+        incomplete_event = {
+            "file_key": "file1"
+            # Missing 'created_at_formatted_string'
+        }
+        with self.assertRaises(MessageNotSuccessfulError) as error:
+            self.batch_filename_to_events_mapper.add_event(incomplete_event)
+
+        self.assertEqual(error.exception.message, "Filename data was not present")
+
+    def test_get_map_returns_a_copy_instead_of_exact_references(self):
+        self.batch_filename_to_events_mapper.add_event(MOCK_SUPPLIER_ONE_RSV_EVENT_ONE)
+
+        result = self.batch_filename_to_events_mapper.get_map()
+
+        self.assertNotEqual(id(result), id(self.batch_filename_to_events_mapper._filename_to_events_map))
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/tests/batch/__init__.py

@@ -1,7 +1,8 @@
 import unittest
 import os
+from typing import Optional
 from unittest import TestCase
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch, MagicMock, call, ANY
 from boto3 import resource as boto3_resource
 from moto import mock_aws
 from models.errors import (
@@ -19,7 +20,10 @@
 from utils.test_utils_for_batch import ForwarderValues, MockFhirImmsResources
 
 with patch.dict("os.environ", ForwarderValues.MOCK_ENVIRONMENT_DICT):
-    from forwarding_batch_lambda import forward_lambda_handler, create_diagnostics_dictionary, forward_request_to_dynamo
+    from forwarding_batch_lambda import forward_lambda_handler, create_diagnostics_dictionary, forward_request_to_dynamo, \
+    QUEUE_URL
+
+
 @mock_aws
 @patch.dict(os.environ, ForwarderValues.MOCK_ENVIRONMENT_DICT)
 class TestForwardLambdaHandler(TestCase):
@@ -99,13 +103,14 @@ def generate_details_from_processing(
 
     @staticmethod
     def generate_input(
-        row_id,
-        identifier_value=None,
-        file_key="test_file_key",
-        created_at_formatted_string="2025-01-24T12:00:00Z",
-        include_fhir_json=True,
-        operation_requested="create",
-        diagnostics=None,
+        row_id: int,
+        identifier_value: Optional[str] = None,
+        file_key: str = "test_file_key",
+        created_at_formatted_string: str = "2025-01-24T12:00:00Z",
+        include_fhir_json: bool = True,
+        operation_requested: str = "create",
+        diagnostics: dict = None,
+        supplier: str = "test_supplier"
     ):
         """Generates input rows for test_cases."""
         details_from_processing = TestForwardLambdaHandler.generate_details_from_processing(
@@ -118,7 +123,7 @@ def generate_input(
             "row_id": f"row-{row_id}",
             "file_key": file_key,
             "created_at_formatted_string": created_at_formatted_string,
-            "supplier": "test_supplier",
+            "supplier": supplier,
             "vax_type": "RSV",
             **details_from_processing,
         }
@@ -403,13 +408,71 @@ def test_forward_lambda_handler_multiple_scenarios(self, mock_send_message):
         mock_send_message.reset_mock()
         event = self.generate_event(test_cases)
 
-
         self.mock_redis_client.hget.return_value = "RSV"
         forward_lambda_handler(event, {})
 
         self.assert_dynamo_item(table_item)
         self.assert_values_in_sqs_messages(mock_send_message, test_cases)
 
+    @patch("forwarding_batch_lambda.sqs_client.send_message")
+    def test_forward_lambda_handler_groups_and_sends_events_by_filename(self, mock_send_message):
+        """VED-734 - each batch handled by the Lambda may have events relating to different parent CSV files. This
+        test ensures events are grouped accordingly and sent with the correct SQS"""
+        mock_records = [
+            {
+                "input": self.generate_input(
+                    row_id=1,
+                    identifier_value="supplier_1_system/54321",
+                    operation_requested="CREATE",
+                    file_key="supplier_1_rsv_test_file",
+                    supplier="supplier_1"
+                )
+            },
+            {
+                "input": self.generate_input(
+                    row_id=2,
+                    identifier_value="supplier_2_system/12345",
+                    operation_requested="CREATE",
+                    file_key="supplier_2_rsv_test_file",
+                    supplier="supplier_2"
+                )
+            }
+        ]
+        mock_kinesis_event = self.generate_event(mock_records)
+        self.mock_redis_client.hget.return_value = "RSV"
+
+        forward_lambda_handler(mock_kinesis_event, {})
+
+        sqs_calls = mock_send_message.call_args_list
+        _, first_call_kwargs = sqs_calls[0]
+        _, second_call_kwargs = sqs_calls[1]
+
+        # Separate calls are made for each of the respective groups
+        self.assertEqual(len(sqs_calls), 2)
+        self.assertEqual(first_call_kwargs["MessageGroupId"], "supplier_1_rsv_test_file_2025-01-24T12:00:00Z")
+        self.assertEqual(second_call_kwargs["MessageGroupId"], "supplier_2_rsv_test_file_2025-01-24T12:00:00Z")
+
+        self.assertDictEqual(json.loads(first_call_kwargs["MessageBody"])[0], {
+            "created_at_formatted_string": "2025-01-24T12:00:00Z",
+            "file_key": "supplier_1_rsv_test_file",
+            "imms_id": ANY,
+            "local_id": "local-1",
+            "operation_requested": "CREATE",
+            "row_id": "row-1",
+            "supplier": "supplier_1",
+            "vaccine_type": "RSV"
+        })
+        self.assertDictEqual(json.loads(second_call_kwargs["MessageBody"])[0], {
+            "created_at_formatted_string": "2025-01-24T12:00:00Z",
+            "file_key": "supplier_2_rsv_test_file",
+            "imms_id": ANY,
+            "local_id": "local-2",
+            "operation_requested": "CREATE",
+            "row_id": "row-2",
+            "supplier": "supplier_2",
+            "vaccine_type": "RSV"
+        })
+
     @patch("forwarding_batch_lambda.sqs_client.send_message")
     def test_forward_lambda_handler_update_scenarios(self, mock_send_message):
         """Test forward lambda handler with multiple rows in the event with update and delete operations,
@@ -584,7 +647,7 @@ def test_create_diagnostics_dictionary(self):
     @patch("forwarding_batch_lambda.forward_request_to_dynamo")
     @patch("forwarding_batch_lambda.create_table")
     @patch("forwarding_batch_lambda.make_batch_controller")
-    def test_forward_request_to_dyanamo(
+    def test_forward_request_to_dynamo(
         self, mock_make_controller, mock_create_table, mock_forward_request_to_dynamo, mock_send_message
     ):
         """Test forward lambda handler to assert dynamo db is called,