Merge branch 'master' into dependabot/terraform/infrastructure/account/terraform-minor-patch-7a8f69ee52

Author: dlzhry2nhs

.github/workflows/create-release-tag.yml

@@ -15,7 +15,7 @@ jobs:
           fetch-depth: 0 # This causes all history to be fetched, which is required for calculate-version to function
 
       - name: Install Python 3.9
-        uses: actions/setup-python@v6
+        uses: actions/setup-python@v6.1.0
         with:
           python-version: 3.9
 

.github/workflows/deploy-backend.yml

@@ -63,7 +63,7 @@ jobs:
         uses: actions/checkout@93cb6efe18208431cddfb8368fd83d5badbf9bfd
 
       - name: Connect to AWS
-        uses: aws-actions/configure-aws-credentials@00943011d9042930efac3dcd3a170e4273319bc8
+        uses: aws-actions/configure-aws-credentials@61815dcd50bd041e203e49132bacad1fd04d2708
         with:
           aws-region: eu-west-2
           role-to-assume: arn:aws:iam::${{ vars.AWS_ACCOUNT_ID }}:role/auto-ops
@@ -96,7 +96,7 @@ jobs:
       - name: Checkout
         uses: actions/checkout@93cb6efe18208431cddfb8368fd83d5badbf9bfd
 
-      - uses: aws-actions/configure-aws-credentials@00943011d9042930efac3dcd3a170e4273319bc8
+      - uses: aws-actions/configure-aws-credentials@61815dcd50bd041e203e49132bacad1fd04d2708
         with:
           aws-region: eu-west-2
           role-to-assume: arn:aws:iam::${{ vars.AWS_ACCOUNT_ID }}:role/auto-ops
@@ -126,7 +126,7 @@ jobs:
         if: ${{ inputs.environment == 'dev' && inputs.create_mns_subscription }}
         run: pip install poetry==2.1.4
 
-      - uses: actions/setup-python@v6
+      - uses: actions/setup-python@v6.1.0
         if: ${{ inputs.environment == 'dev' && inputs.create_mns_subscription }}
         with:
           python-version: 3.11

.github/workflows/pr-teardown.yml

@@ -26,7 +26,7 @@ jobs:
 
     steps:
       - name: Connect to AWS
-        uses: aws-actions/configure-aws-credentials@00943011d9042930efac3dcd3a170e4273319bc8
+        uses: aws-actions/configure-aws-credentials@61815dcd50bd041e203e49132bacad1fd04d2708
         with:
           aws-region: eu-west-2
           role-to-assume: arn:aws:iam::${{ vars.AWS_ACCOUNT_ID }}:role/auto-ops
@@ -52,7 +52,7 @@ jobs:
       - name: Install poetry
         run: pip install poetry==2.1.4
 
-      - uses: actions/setup-python@v6
+      - uses: actions/setup-python@v6.1.0
         with:
           python-version: 3.11
           cache: "poetry"

.github/workflows/quality-checks.yml

@@ -43,7 +43,7 @@ jobs:
       - name: Install poetry
         run: pip install poetry==2.1.4
 
-      - uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c
+      - uses: actions/setup-python@83679a892e2d95755f2dac6acb0bfd1e9ac5d548
         with:
           python-version: 3.11
           cache: "poetry"
@@ -84,7 +84,7 @@ jobs:
       - name: Install poetry
         run: pip install poetry==2.1.4
 
-      - uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c
+      - uses: actions/setup-python@83679a892e2d95755f2dac6acb0bfd1e9ac5d548
         with:
           python-version: 3.11
           cache: "poetry"

.github/workflows/run-e2e-tests.yml

@@ -119,7 +119,7 @@ jobs:
 
       - name: Connect to AWS
         if: ${{ env.TF_OUTPUTS_REQUIRED == 'true' }}
-        uses: aws-actions/configure-aws-credentials@00943011d9042930efac3dcd3a170e4273319bc8
+        uses: aws-actions/configure-aws-credentials@61815dcd50bd041e203e49132bacad1fd04d2708
         with:
           aws-region: eu-west-2
           role-to-assume: arn:aws:iam::${{ vars.AWS_ACCOUNT_ID }}:role/auto-ops
@@ -153,7 +153,7 @@ jobs:
       - name: Install poetry
         run: pip install poetry==2.1.4
 
-      - uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c
+      - uses: actions/setup-python@83679a892e2d95755f2dac6acb0bfd1e9ac5d548
         with:
           python-version: 3.11
           cache: "poetry"
@@ -210,7 +210,7 @@ jobs:
       - uses: actions/checkout@93cb6efe18208431cddfb8368fd83d5badbf9bfd
 
       - name: Connect to AWS
-        uses: aws-actions/configure-aws-credentials@00943011d9042930efac3dcd3a170e4273319bc8
+        uses: aws-actions/configure-aws-credentials@61815dcd50bd041e203e49132bacad1fd04d2708
         with:
           aws-region: eu-west-2
           role-to-assume: arn:aws:iam::${{ vars.AWS_ACCOUNT_ID }}:role/auto-ops
@@ -219,7 +219,7 @@ jobs:
       - name: Install poetry
         run: pip install poetry==2.1.4
 
-      - uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c
+      - uses: actions/setup-python@83679a892e2d95755f2dac6acb0bfd1e9ac5d548
         with:
           python-version: 3.11
           cache: "poetry"

lambdas/ack_backend/src/ack_processor.py

@@ -2,10 +2,10 @@
 
 import json
 
+from common.batch.eof_utils import is_eof_message
 from convert_message_to_ack_row import convert_message_to_ack_row
 from logging_decorators import ack_lambda_handler_logging_decorator
 from update_ack_file import complete_batch_file_process, update_ack_file
-from utils_for_ack_lambda import is_ack_processing_complete
 
 
 @ack_lambda_handler_logging_decorator
@@ -22,9 +22,11 @@ def lambda_handler(event, _):
     file_key = None
     created_at_formatted_string = None
     message_id = None
+    supplier = None
+    vaccine_type = None
 
     ack_data_rows = []
-    total_ack_rows_processed = 0
+    file_processing_complete = False
 
     for i, record in enumerate(event["Records"]):
         try:
@@ -42,18 +44,16 @@ def lambda_handler(event, _):
             created_at_formatted_string = incoming_message_body[0].get("created_at_formatted_string")
 
         for message in incoming_message_body:
+            if is_eof_message(message):
+                file_processing_complete = True
+                break
+
             ack_data_rows.append(convert_message_to_ack_row(message, created_at_formatted_string))
 
     update_ack_file(file_key, created_at_formatted_string, ack_data_rows)
 
-    # Get the row count of the final processed record
-    # Format of the row id is {batch_message_id}^{row_number}
-    total_ack_rows_processed = int(incoming_message_body[-1].get("row_id", "").split("^")[1])
-
-    if is_ack_processing_complete(message_id, total_ack_rows_processed):
-        complete_batch_file_process(
-            message_id, supplier, vaccine_type, created_at_formatted_string, file_key, total_ack_rows_processed
-        )
+    if file_processing_complete:
+        complete_batch_file_process(message_id, supplier, vaccine_type, created_at_formatted_string, file_key)
 
     return {
         "statusCode": 200,

lambdas/ack_backend/src/update_ack_file.py

@@ -4,7 +4,7 @@
 
 from botocore.exceptions import ClientError
 
-from audit_table import change_audit_table_status_to_processed
+from audit_table import change_audit_table_status_to_processed, get_record_count_by_message_id
 from common.aws_s3_utils import move_file
 from common.clients import get_s3_client, logger
 from constants import (
@@ -61,7 +61,6 @@ def complete_batch_file_process(
     vaccine_type: str,
     created_at_formatted_string: str,
     file_key: str,
-    total_ack_rows_processed: int,
 ) -> dict:
     """Mark the batch file as processed. This involves moving the ack and original file to destinations and updating
     the audit table status"""
@@ -72,6 +71,7 @@ def complete_batch_file_process(
         get_source_bucket_name(), f"{BATCH_FILE_PROCESSING_DIR}/{file_key}", f"{BATCH_FILE_ARCHIVE_DIR}/{file_key}"
     )
 
+    total_ack_rows_processed = get_record_count_by_message_id(message_id)
     change_audit_table_status_to_processed(file_key, message_id)
 
     return {
@@ -111,7 +111,7 @@ def update_ack_file(
     """Updates the ack file with the new data row based on the given arguments"""
     ack_filename = f"{file_key.replace('.csv', f'_BusAck_{created_at_formatted_string}.csv')}"
     temp_ack_file_key = f"{TEMP_ACK_DIR}/{ack_filename}"
-    archive_ack_file_key = f"{COMPLETED_ACK_DIR}/{ack_filename}"
+    completed_ack_file_key = f"{COMPLETED_ACK_DIR}/{ack_filename}"
     accumulated_csv_content = obtain_current_ack_content(temp_ack_file_key)
 
     for row in ack_data_rows:
@@ -123,4 +123,4 @@ def update_ack_file(
     ack_bucket_name = get_ack_bucket_name()
 
     get_s3_client().upload_fileobj(csv_file_like_object, ack_bucket_name, temp_ack_file_key)
-    logger.info("Ack file updated to %s: %s", ack_bucket_name, archive_ack_file_key)
+    logger.info("Ack file updated to %s: %s", ack_bucket_name, completed_ack_file_key)

lambdas/ack_backend/src/utils_for_ack_lambda.py

@@ -3,8 +3,9 @@
 import json
 import os
 import unittest
+from copy import deepcopy
 from io import StringIO
-from unittest.mock import Mock, patch
+from unittest.mock import patch
 
 from boto3 import client as boto3_client
 from moto import mock_aws
@@ -30,7 +31,6 @@
     DiagnosticsDictionaries,
     ValidValues,
 )
-from utils_for_ack_lambda import _BATCH_EVENT_ID_TO_RECORD_COUNT_MAP
 
 with patch.dict("os.environ", MOCK_ENVIRONMENT_DICT):
     from ack_processor import lambda_handler
@@ -269,45 +269,6 @@ def test_lambda_handler_updates_ack_file_but_does_not_mark_complete_when_records
         )
         self.assert_audit_entry_status_equals(mock_batch_message_id, "Preprocessed")
 
-    @patch("utils_for_ack_lambda.get_record_count_by_message_id", return_value=500)
-    def test_lambda_handler_uses_message_id_to_record_count_cache_to_reduce_ddb_calls(self, mock_get_record_count: Mock):
-        """The DynamoDB Audit table is used to store the total record count for each source file. To reduce calls each
-        time - this test checks that we cache the value as this lambda is called many times for large files"""
-        mock_batch_message_id = "622cdeea-461e-4a83-acb5-7871d47ddbcd"
-
-        # Original source file had 500 records
-        add_audit_entry_to_table(self.dynamodb_client, mock_batch_message_id, record_count=500)
-
-        message_one = [
-            {**BASE_SUCCESS_MESSAGE, "row_id": f"{mock_batch_message_id}^1", "imms_id": "imms_1", "local_id": "local^1"}
-        ]
-        message_two = [
-            {**BASE_SUCCESS_MESSAGE, "row_id": f"{mock_batch_message_id}^2", "imms_id": "imms_2", "local_id": "local^2"}
-        ]
-        test_event_one = {"Records": [{"body": json.dumps(message_one)}]}
-        test_event_two = {"Records": [{"body": json.dumps(message_two)}]}
-
-        response = lambda_handler(event=test_event_one, context={})
-        self.assertEqual(response, EXPECTED_ACK_LAMBDA_RESPONSE_FOR_SUCCESS)
-        second_invocation_response = lambda_handler(event=test_event_two, context={})
-        self.assertEqual(second_invocation_response, EXPECTED_ACK_LAMBDA_RESPONSE_FOR_SUCCESS)
-
-        # Assert that the DDB call is only performed once on the first invocation
-        mock_get_record_count.assert_called_once_with(mock_batch_message_id)
-        validate_ack_file_content(
-            self.s3_client,
-            [*message_one, *message_two],
-            existing_file_content=ValidValues.ack_headers,
-        )
-        self.assert_ack_and_source_file_locations_correct(
-            MOCK_MESSAGE_DETAILS.file_key,
-            MOCK_MESSAGE_DETAILS.temp_ack_file_key,
-            MOCK_MESSAGE_DETAILS.archive_ack_file_key,
-            is_complete=False,
-        )
-        self.assertEqual(_BATCH_EVENT_ID_TO_RECORD_COUNT_MAP[mock_batch_message_id], 500)
-        self.assert_audit_entry_status_equals(mock_batch_message_id, "Preprocessed")
-
     def test_lambda_handler_updates_ack_file_and_marks_complete_when_all_records_processed(self):
         """
         Test that the batch file process is marked as complete when all records have been processed.
@@ -339,7 +300,11 @@ def test_lambda_handler_updates_ack_file_and_marks_complete_when_all_records_pro
             }
             for i in range(50, 101)
         ]
-        test_event = {"Records": [{"body": json.dumps(array_of_success_messages)}]}
+
+        # Include the EoF message in the event
+        all_messages_plus_eof = deepcopy(array_of_success_messages)
+        all_messages_plus_eof.append(MOCK_MESSAGE_DETAILS.eof_message)
+        test_event = {"Records": [{"body": json.dumps(all_messages_plus_eof)}]}
 
         response = lambda_handler(event=test_event, context={})
 

lambdas/ack_backend/tests/test_ack_processor.py

@@ -9,13 +9,13 @@
 from boto3 import client as boto3_client
 from moto import mock_aws
 
-from tests.utils.generic_setup_and_teardown_for_ack_backend import (
+from utils.generic_setup_and_teardown_for_ack_backend import (
     GenericSetUp,
     GenericTearDown,
 )
-from tests.utils.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames
-from tests.utils.utils_for_ack_backend_tests import generate_event
-from tests.utils.values_for_ack_backend_tests import (
+from utils.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames
+from utils.utils_for_ack_backend_tests import generate_event
+from utils.values_for_ack_backend_tests import (
     EXPECTED_ACK_LAMBDA_RESPONSE_FOR_SUCCESS,
     DiagnosticsDictionaries,
     InvalidValues,
@@ -120,7 +120,6 @@ def test_splunk_logging_successful_rows(self):
         for operation in ["CREATE", "UPDATE", "DELETE"]:
             with (  # noqa: E999
                 patch("common.log_decorator.send_log_to_firehose") as mock_send_log_to_firehose,  # noqa: E999
-                patch("ack_processor.is_ack_processing_complete", return_value=False),
                 patch("common.log_decorator.logger") as mock_logger,  # noqa: E999
             ):  # noqa: E999
                 result = lambda_handler(
@@ -224,7 +223,6 @@ def test_splunk_logging_statuscode_diagnostics(
         for test_case in test_cases:
             with (  # noqa: E999
                 patch("common.log_decorator.send_log_to_firehose") as mock_send_log_to_firehose,  # noqa: E999
-                patch("ack_processor.is_ack_processing_complete", return_value=False),
                 patch("common.log_decorator.logger") as mock_logger,  # noqa: E999
             ):  # noqa: E999
                 result = lambda_handler(
@@ -262,7 +260,6 @@ def test_splunk_logging_multiple_rows(self):
 
         with (  # noqa: E999
             patch("common.log_decorator.send_log_to_firehose") as mock_send_log_to_firehose,  # noqa: E999
-            patch("ack_processor.is_ack_processing_complete", return_value=False),
             patch("common.log_decorator.logger") as mock_logger,  # noqa: E999
         ):  # noqa: E999
             result = lambda_handler(generate_event(messages), context={})
@@ -313,7 +310,6 @@ def test_splunk_logging_multiple_with_diagnostics(
 
         with (  # noqa: E999
             patch("common.log_decorator.send_log_to_firehose") as mock_send_log_to_firehose,  # noqa: E999
-            patch("ack_processor.is_ack_processing_complete", return_value=False),
             patch("common.log_decorator.logger") as mock_logger,  # noqa: E999
         ):  # noqa: E999
             result = lambda_handler(generate_event(messages), context={})
@@ -369,7 +365,6 @@ def test_splunk_logging_multiple_with_diagnostics(
         )
 
     def test_splunk_update_ack_file_not_logged(self):
-        self.maxDiff = None
         """Tests that update_ack_file is not logged if we have sent acks for less than the whole file"""
         # send 98 messages
         messages = []
@@ -380,7 +375,6 @@ def test_splunk_update_ack_file_not_logged(self):
         with (  # noqa: E999
             patch("common.log_decorator.send_log_to_firehose") as mock_send_log_to_firehose,  # noqa: E999
             patch("common.log_decorator.logger") as mock_logger,  # noqa: E999
-            patch("ack_processor.is_ack_processing_complete", return_value=False),
             patch(
                 "update_ack_file.change_audit_table_status_to_processed"
             ) as mock_change_audit_table_status_to_processed,  # noqa: E999
@@ -421,12 +415,12 @@ def test_splunk_update_ack_file_logged(self):
         with (  # noqa: E999
             patch("common.log_decorator.send_log_to_firehose") as mock_send_log_to_firehose,  # noqa: E999
             patch("common.log_decorator.logger") as mock_logger,  # noqa: E999
-            patch("ack_processor.is_ack_processing_complete", return_value=True),
+            patch("update_ack_file.get_record_count_by_message_id", return_value=99),
             patch(
                 "update_ack_file.change_audit_table_status_to_processed"
             ) as mock_change_audit_table_status_to_processed,  # noqa: E999
         ):  # noqa: E999
-            result = lambda_handler(generate_event(messages), context={})
+            result = lambda_handler(generate_event(messages, include_eof_message=True), context={})
 
         self.assertEqual(result, EXPECTED_ACK_LAMBDA_RESPONSE_FOR_SUCCESS)