review fixes

Author: JamesW1-NHS

lambdas/ack_backend/src/ack_processor.py

@@ -50,6 +50,9 @@ def lambda_handler(event, _):
                 break
 
             ack_data_rows.append(convert_message_to_ack_row(message, created_at_formatted_string))
+            # Note: we are only currently receiving failure records.
+            # Nevertheless, we leave this here pending rewrite of tests (which currently still incorporate
+            # success records and use "diagnostics" to identify failures).
             if message.get("diagnostics"):
                 increment_records_failed_count(message_id)
 

lambdas/ack_backend/src/audit_table.py

@@ -62,21 +62,21 @@ def set_records_succeeded_count(message_id: str) -> None:
     records_failed = int(records_failed_item) if records_failed_item else 0
     records_succeeded = record_count - records_failed
 
-    counter_name = AuditTableKeys.RECORDS_SUCCEEDED
+    attribute_name = AuditTableKeys.RECORDS_SUCCEEDED
     try:
         response = dynamodb_client.update_item(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": message_id}},
-            UpdateExpression="SET #counter = :value",
-            ExpressionAttributeNames={"#counter": counter_name},
+            UpdateExpression="SET #attribute = :value",
+            ExpressionAttributeNames={"#attribute": attribute_name},
             ExpressionAttributeValues={":value": {"N": str(records_succeeded)}},
             ConditionExpression=CONDITION_EXPRESSION,
             ReturnValues="UPDATED_NEW",
         )
-        result = response.get("Attributes", {}).get(counter_name).get("N")
+        result = response.get("Attributes", {}).get(attribute_name).get("N")
         logger.info(
-            "Counter %s for message id %s set to %s in the audit table",
-            counter_name,
+            "Attribute %s for message id %s set to %s in the audit table",
+            attribute_name,
             message_id,
             result,
         )
@@ -94,43 +94,37 @@ def increment_records_failed_count(message_id: str) -> None:
 
     increment_value = 1
     initial_value = 0
-    counter_name = AuditTableKeys.RECORDS_FAILED
+    attribute_name = AuditTableKeys.RECORDS_FAILED
     try:
-        # Use SET with if_not_exists to safely increment the counter
+        # Use SET with if_not_exists to safely increment the counter attribute
         response = dynamodb_client.update_item(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": message_id}},
-            UpdateExpression="SET #counter = if_not_exists(#counter, :initial) + :increment",
-            ExpressionAttributeNames={"#counter": counter_name},
+            UpdateExpression="SET #attribute = if_not_exists(#attribute, :initial) + :increment",
+            ExpressionAttributeNames={"#attribute": attribute_name},
             ExpressionAttributeValues={":increment": {"N": str(increment_value)}, ":initial": {"N": str(initial_value)}},
             ConditionExpression=CONDITION_EXPRESSION,
             ReturnValues="UPDATED_NEW",
         )
-        result = response.get("Attributes", {}).get(counter_name).get("N")
-        logger.info(
-            "Counter %s for message id %s incremented to %s in the audit table",
-            counter_name,
-            message_id,
-            result,
-        )
+        response.get("Attributes", {}).get(attribute_name).get("N")
 
     except Exception as error:  # pylint: disable = broad-exception-caught
         logger.error(error)
         raise UnhandledAuditTableError(error) from error
 
 
-def set_audit_table_ingestion_complete(
+def set_audit_table_ingestion_end_time(
     file_key: str,
     message_id: str,
     complete_time: float,
 ) -> None:
-    """Sets the ingestion_complete in the audit table to the requested time"""
+    """Sets the ingestion_end_time in the audit table to the requested time"""
     # format the time
-    ingestion_complete = time.strftime("%Y%m%dT%H%M%S00", time.gmtime(complete_time))
+    ingestion_end_time = time.strftime("%Y%m%dT%H%M%S00", time.gmtime(complete_time))
 
-    update_expression = f"SET #{AuditTableKeys.INGESTION_COMPLETE} = :{AuditTableKeys.INGESTION_COMPLETE}"
-    expression_attr_names = {f"#{AuditTableKeys.INGESTION_COMPLETE}": AuditTableKeys.INGESTION_COMPLETE}
-    expression_attr_values = {f":{AuditTableKeys.INGESTION_COMPLETE}": {"S": ingestion_complete}}
+    update_expression = f"SET #{AuditTableKeys.INGESTION_END_TIME} = :{AuditTableKeys.INGESTION_END_TIME}"
+    expression_attr_names = {f"#{AuditTableKeys.INGESTION_END_TIME}": AuditTableKeys.INGESTION_END_TIME}
+    expression_attr_values = {f":{AuditTableKeys.INGESTION_END_TIME}": {"S": ingestion_end_time}}
 
     try:
         response = dynamodb_client.update_item(
@@ -142,9 +136,9 @@ def set_audit_table_ingestion_complete(
             ConditionExpression=f"attribute_exists({AuditTableKeys.MESSAGE_ID})",
             ReturnValues="UPDATED_NEW",
         )
-        result = response.get("Attributes", {}).get(AuditTableKeys.INGESTION_COMPLETE).get("S")
+        result = response.get("Attributes", {}).get(AuditTableKeys.INGESTION_END_TIME).get("S")
         logger.info(
-            "ingestion_complete for %s file, with message id %s, was successfully updated to %s in the audit table",
+            "ingestion_end_time for %s file, with message id %s, was successfully updated to %s in the audit table",
             file_key,
             message_id,
             result,

lambdas/ack_backend/src/constants.py

@@ -38,7 +38,7 @@ class AuditTableKeys:
     RECORD_COUNT = "record_count"
     STATUS = "status"
     TIMESTAMP = "timestamp"
-    INGESTION_COMPLETE = "ingestion_complete"
+    INGESTION_END_TIME = "ingestion_end_time"
     RECORDS_SUCCEEDED = "records_succeeded"
     RECORDS_FAILED = "records_failed"
 

lambdas/ack_backend/src/logging_decorators.py

@@ -5,7 +5,6 @@
 from datetime import datetime
 from functools import wraps
 
-from audit_table import set_audit_table_ingestion_complete
 from common.log_decorator import generate_and_send_logs
 
 PREFIX = "ack_processor"
@@ -79,10 +78,9 @@ def wrapper(*args, **kwargs):
             "function_name": f"{PREFIX}_{func.__name__}",
             "date_time": str(datetime.now()),
         }
-        complete_time = time.time()
 
         # NB this doesn't require a try-catch block as the wrapped function never throws an exception
-        result = func(*args, **kwargs)
+        result, ingestion_end_time = func(*args, **kwargs)
         if result is not None:
             message_for_logs = "Record processing complete"
             base_log_data.update(result)
@@ -91,11 +89,8 @@ def wrapper(*args, **kwargs):
                 "statusCode": 200,
                 "message": message_for_logs,
             }
-            file_key = base_log_data.get("filename")
-            message_id = base_log_data.get("message_id")
-            set_audit_table_ingestion_complete(file_key, message_id, complete_time)
+            generate_and_send_logs(STREAM_NAME, ingestion_end_time, base_log_data, additional_log_data)
 
-            generate_and_send_logs(STREAM_NAME, complete_time, base_log_data, additional_log_data)
         return result
 
     return wrapper

lambdas/ack_backend/src/update_ack_file.py

@@ -1,12 +1,14 @@
 """Functions for uploading the data to the ack file"""
 
+import time
 from io import BytesIO, StringIO
 
 from botocore.exceptions import ClientError
 
 from audit_table import (
     change_audit_table_status_to_processed,
     get_record_count_by_message_id,
+    set_audit_table_ingestion_end_time,
     set_records_succeeded_count,
 )
 from common.aws_s3_utils import move_file
@@ -65,7 +67,7 @@ def complete_batch_file_process(
     vaccine_type: str,
     created_at_formatted_string: str,
     file_key: str,
-) -> dict:
+) -> tuple[dict, float]:
     """Mark the batch file as processed. This involves moving the ack and original file to destinations and updating
     the audit table status"""
     ack_filename = f"{file_key.replace('.csv', f'_BusAck_{created_at_formatted_string}.csv')}"
@@ -79,13 +81,17 @@ def complete_batch_file_process(
     change_audit_table_status_to_processed(file_key, message_id)
     set_records_succeeded_count(message_id)
 
-    return {
+    ingestion_end_time = time.time()
+    set_audit_table_ingestion_end_time(file_key, message_id, ingestion_end_time)
+
+    result = {
         "message_id": message_id,
         "file_key": file_key,
         "supplier": supplier,
         "vaccine_type": vaccine_type,
         "row_count": total_ack_rows_processed,
     }
+    return result, ingestion_end_time
 
 
 def obtain_current_ack_content(temp_ack_file_key: str) -> StringIO:

lambdas/ack_backend/tests/test_audit_table.py

@@ -65,8 +65,8 @@ def test_set_records_succeeded_count(self):
         self.mock_dynamodb_client.update_item.assert_called_once_with(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": test_message_id}},
-            UpdateExpression="SET #counter = :value",
-            ExpressionAttributeNames={"#counter": AuditTableKeys.RECORDS_SUCCEEDED},
+            UpdateExpression="SET #attribute = :value",
+            ExpressionAttributeNames={"#attribute": AuditTableKeys.RECORDS_SUCCEEDED},
             ExpressionAttributeValues={":value": {"N": "958"}},
             ConditionExpression="attribute_exists(message_id)",
             ReturnValues="UPDATED_NEW",
@@ -83,8 +83,8 @@ def test_set_records_succeeded_count_no_failures(self):
         self.mock_dynamodb_client.update_item.assert_called_once_with(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": test_message_id}},
-            UpdateExpression="SET #counter = :value",
-            ExpressionAttributeNames={"#counter": AuditTableKeys.RECORDS_SUCCEEDED},
+            UpdateExpression="SET #attribute = :value",
+            ExpressionAttributeNames={"#attribute": AuditTableKeys.RECORDS_SUCCEEDED},
             ExpressionAttributeValues={":value": {"N": "1000"}},
             ConditionExpression="attribute_exists(message_id)",
             ReturnValues="UPDATED_NEW",
@@ -99,8 +99,8 @@ def test_set_records_succeeded_count_no_records(self):
         self.mock_dynamodb_client.update_item.assert_called_once_with(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": test_message_id}},
-            UpdateExpression="SET #counter = :value",
-            ExpressionAttributeNames={"#counter": AuditTableKeys.RECORDS_SUCCEEDED},
+            UpdateExpression="SET #attribute = :value",
+            ExpressionAttributeNames={"#attribute": AuditTableKeys.RECORDS_SUCCEEDED},
             ExpressionAttributeValues={":value": {"N": "0"}},
             ConditionExpression="attribute_exists(message_id)",
             ReturnValues="UPDATED_NEW",
@@ -121,13 +121,12 @@ def test_increment_records_failed_count(self):
         self.mock_dynamodb_client.update_item.assert_called_once_with(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": test_message_id}},
-            UpdateExpression="SET #counter = if_not_exists(#counter, :initial) + :increment",
-            ExpressionAttributeNames={"#counter": AuditTableKeys.RECORDS_FAILED},
+            UpdateExpression="SET #attribute = if_not_exists(#attribute, :initial) + :increment",
+            ExpressionAttributeNames={"#attribute": AuditTableKeys.RECORDS_FAILED},
             ExpressionAttributeValues={":increment": {"N": "1"}, ":initial": {"N": "0"}},
             ConditionExpression="attribute_exists(message_id)",
             ReturnValues="UPDATED_NEW",
         )
-        self.mock_logger.info.assert_called_once()
 
     def test_increment_records_failed_count_raises(self):
         self.mock_dynamodb_client.update_item.side_effect = Exception("fail!")
@@ -136,29 +135,29 @@ def test_increment_records_failed_count_raises(self):
         self.assertIn("fail!", str(ctx.exception))
         self.mock_logger.error.assert_called_once()
 
-    def test_set_audit_table_ingestion_complete(self):
-        """Checks audit table correctly sets ingestion_complete to the requested value"""
+    def test_set_audit_table_ingestion_end_time(self):
+        """Checks audit table correctly sets ingestion_end_time to the requested value"""
         test_file_key = "RSV_Vaccinations_v5_X26_20210730T12000000.csv"
         test_message_id = "1234"
-        test_start_time = 1627647000
-        audit_table.set_audit_table_ingestion_complete(test_file_key, test_message_id, test_start_time)
+        test_end_time = 1627647000
+        audit_table.set_audit_table_ingestion_end_time(test_file_key, test_message_id, test_end_time)
         self.mock_dynamodb_client.update_item.assert_called_once_with(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": test_message_id}},
-            UpdateExpression=f"SET #{AuditTableKeys.INGESTION_COMPLETE} = :{AuditTableKeys.INGESTION_COMPLETE}",
-            ExpressionAttributeNames={f"#{AuditTableKeys.INGESTION_COMPLETE}": AuditTableKeys.INGESTION_COMPLETE},
-            ExpressionAttributeValues={f":{AuditTableKeys.INGESTION_COMPLETE}": {"S": "20210730T12100000"}},
+            UpdateExpression=f"SET #{AuditTableKeys.INGESTION_END_TIME} = :{AuditTableKeys.INGESTION_END_TIME}",
+            ExpressionAttributeNames={f"#{AuditTableKeys.INGESTION_END_TIME}": AuditTableKeys.INGESTION_END_TIME},
+            ExpressionAttributeValues={f":{AuditTableKeys.INGESTION_END_TIME}": {"S": "20210730T12100000"}},
             ConditionExpression="attribute_exists(message_id)",
             ReturnValues="UPDATED_NEW",
         )
         self.mock_logger.info.assert_called_once()
 
-    def test_set_audit_table_ingestion_complete_throws_exception_with_invalid_id(self):
+    def test_set_audit_table_ingestion_end_time_throws_exception_with_invalid_id(self):
         test_file_key = "RSV_Vaccinations_v5_X26_20210730T12000000.csv"
         test_message_id = "1234"
-        test_start_time = 1627647000
+        test_end_time = 1627647000
         self.mock_dynamodb_client.update_item.side_effect = Exception("fail!")
         with self.assertRaises(UnhandledAuditTableError) as ctx:
-            audit_table.set_audit_table_ingestion_complete(test_file_key, test_message_id, test_start_time)
+            audit_table.set_audit_table_ingestion_end_time(test_file_key, test_message_id, test_end_time)
         self.assertIn("fail!", str(ctx.exception))
         self.mock_logger.error.assert_called_once()

lambdas/ack_backend/tests/test_splunk_logging.py

@@ -66,7 +66,7 @@ def run(self, result=None):
             # Time is incremented by 1.0 for each call to time.time for ease of testing.
             # Range is set to a large number (300) due to many calls being made to time.time for some tests.
             patch(
-                "logging_decorators.time.time",
+                "update_ack_file.time.time",
                 side_effect=[0.0 + i for i in range(300)],
             ),
         ]
@@ -121,6 +121,7 @@ def test_splunk_logging_successful_rows(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.increment_records_failed_count"),  # noqa: E999
             ):  # noqa: E999
                 result = lambda_handler(
                     event=generate_event([{"operation_requested": operation}]),
@@ -161,6 +162,7 @@ def test_splunk_logging_missing_data(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.increment_records_failed_count"),  # noqa: E999
         ):  # noqa: E999
             with self.assertRaises(AttributeError):
                 lambda_handler(event={"Records": [{"body": json.dumps([{"": "456", "row_id": "test^1"}])}]}, context={})
@@ -263,6 +265,7 @@ 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("common.log_decorator.logger") as mock_logger,  # noqa: E999
+            patch("ack_processor.increment_records_failed_count"),  # noqa: E999
         ):  # noqa: E999
             result = lambda_handler(generate_event(messages), context={})
 
@@ -382,6 +385,7 @@ def test_splunk_update_ack_file_not_logged(self):
             patch(
                 "update_ack_file.change_audit_table_status_to_processed"
             ) as mock_change_audit_table_status_to_processed,  # noqa: E999
+            patch("ack_processor.increment_records_failed_count"),  # noqa: E999
         ):  # noqa: E999
             result = lambda_handler(generate_event(messages), context={})
 
@@ -424,7 +428,8 @@ def test_splunk_update_ack_file_logged(self):
                 "update_ack_file.change_audit_table_status_to_processed"
             ) as mock_change_audit_table_status_to_processed,  # noqa: E999
             patch("update_ack_file.set_records_succeeded_count") as mock_set_records_succeeded_count,  # noqa: E999
-            patch("logging_decorators.set_audit_table_ingestion_complete"),  # noqa: E999
+            patch("ack_processor.increment_records_failed_count"),  # noqa: E999
+            patch("update_ack_file.set_audit_table_ingestion_end_time"),  # noqa: E999
         ):  # noqa: E999
             result = lambda_handler(generate_event(messages, include_eof_message=True), context={})
 

lambdas/ack_backend/tests/test_update_ack_file_flow.py

@@ -41,15 +41,15 @@ def setUp(self):
         self.mock_get_record_count = self.get_record_count_patcher.start()
         self.set_records_succeeded_count_patcher = patch("update_ack_file.set_records_succeeded_count")
         self.mock_set_records_succeeded_count = self.set_records_succeeded_count_patcher.start()
-        self.set_audit_table_ingestion_complete_patcher = patch("logging_decorators.set_audit_table_ingestion_complete")
-        self.mock_set_audit_table_ingestion_complete = self.set_audit_table_ingestion_complete_patcher.start()
+        self.set_audit_table_ingestion_end_time_patcher = patch("update_ack_file.set_audit_table_ingestion_end_time")
+        self.mock_set_audit_table_ingestion_end_time = self.set_audit_table_ingestion_end_time_patcher.start()
 
     def tearDown(self):
         self.logger_patcher.stop()
         self.change_audit_status_patcher.stop()
         self.get_record_count_patcher.stop()
         self.set_records_succeeded_count_patcher.stop()
-        self.set_audit_table_ingestion_complete_patcher.stop()
+        self.set_audit_table_ingestion_end_time_patcher.stop()
 
     def test_audit_table_updated_correctly_when_ack_process_complete(self):
         """VED-167 - Test that the audit table has been updated correctly"""