Refactored filename processor for new DB statuses and error handling

Author: dlzhry2nhs

filenameprocessor/src/audit_table.py

@@ -1,4 +1,5 @@
 """Add the filename to the audit table and check for duplicates."""
+from typing import Optional
 from clients import dynamodb_client, logger
 from errors import UnhandledAuditTableError
 from constants import AUDIT_TABLE_NAME, AuditTableKeys
@@ -10,23 +11,29 @@ def upsert_audit_table(
     created_at_formatted_str: str,
     expiry_timestamp: int,
     queue_name: str,
-    file_status: str
+    file_status: str,
+    error_details: Optional[str] = None
 ) -> None:
     """
     Updates the audit table with the file details
     """
+    audit_item = {
+        AuditTableKeys.MESSAGE_ID: {"S": message_id},
+        AuditTableKeys.FILENAME: {"S": file_key},
+        AuditTableKeys.QUEUE_NAME: {"S": queue_name},
+        AuditTableKeys.STATUS: {"S": file_status},
+        AuditTableKeys.TIMESTAMP: {"S": created_at_formatted_str},
+        AuditTableKeys.EXPIRES_AT: {"N": str(expiry_timestamp)}
+    }
+
+    if error_details is not None:
+        audit_item[AuditTableKeys.ERROR_DETAILS] = {"S": error_details}
+
     try:
         # Add to the audit table (regardless of whether it is a duplicate)
         dynamodb_client.put_item(
             TableName=AUDIT_TABLE_NAME,
-            Item={
-                AuditTableKeys.MESSAGE_ID: {"S": message_id},
-                AuditTableKeys.FILENAME: {"S": file_key},
-                AuditTableKeys.QUEUE_NAME: {"S": queue_name},
-                AuditTableKeys.STATUS: {"S": file_status},
-                AuditTableKeys.TIMESTAMP: {"S": created_at_formatted_str},
-                AuditTableKeys.EXPIRES_AT: {"N": str(expiry_timestamp)},
-            },
+            Item=audit_item,
             ConditionExpression="attribute_not_exists(message_id)",  # Prevents accidental overwrites
         )
         logger.info("%s file, with message id %s, successfully added to audit table", file_key, message_id)

filenameprocessor/src/constants.py

@@ -6,10 +6,9 @@
 from errors import (
     VaccineTypePermissionsError,
     InvalidFileKeyError,
-    InvalidSupplierError,
     UnhandledAuditTableError,
-    DuplicateFileError,
-    UnhandledSqsError, EmptyFileError,
+    EmptyFileError,
+    UnhandledSqsError
 )
 
 SOURCE_BUCKET_NAME = os.getenv("SOURCE_BUCKET_NAME")
@@ -25,9 +24,7 @@
     VaccineTypePermissionsError: 403,
     InvalidFileKeyError: 400,  # Includes invalid ODS code, therefore unable to identify supplier
     EmptyFileError: 400,
-    InvalidSupplierError: 500,  # Only raised if supplier variable is not correctly set
     UnhandledAuditTableError: 500,
-    DuplicateFileError: 422,
     UnhandledSqsError: 500,
     Exception: 500,
 }
@@ -42,8 +39,15 @@ class FileStatus(StrEnum):
     QUEUED = "Queued"
     PROCESSING = "Processing"
     PROCESSED = "Processed"
-    DUPLICATE = "Not processed - duplicate"
-    EMPTY = "Not processed - empty file"
+    NOT_PROCESSED = "Not processed"
+    FAILED = "Failed"
+
+
+class FileNotProcessedReason(StrEnum):
+    """Reasons why a file was not processed"""
+    EMPTY = "Empty file"
+    UNAUTHORISED = "Unauthorised"
+    INVALID_FILENAME = "Invalid filename"
 
 
 class AuditTableKeys(StrEnum):
@@ -55,3 +59,4 @@ class AuditTableKeys(StrEnum):
     STATUS = "status"
     TIMESTAMP = "timestamp"
     EXPIRES_AT = "expires_at"
+    ERROR_DETAILS = "error_details"

filenameprocessor/src/errors.py

@@ -1,10 +1,6 @@
 """Custom exceptions for the Filename Processor."""
 
 
-class DuplicateFileError(Exception):
-    """A custom exception for when it is identified that the file is a duplicate."""
-
-
 class EmptyFileError(Exception):
     """A custom exception for when the batch file contains only the header row or is completely empty"""
 
@@ -21,9 +17,5 @@ class InvalidFileKeyError(Exception):
     """A custom exception for when the file key is invalid."""
 
 
-class InvalidSupplierError(Exception):
-    """A custom exception for when the supplier has not been correctly identified."""
-
-
 class UnhandledSqsError(Exception):
     """A custom exception for when an unexpected error occurs whilst sending a message to SQS."""

filenameprocessor/src/file_name_processor.py

@@ -19,12 +19,11 @@
 from errors import (
     VaccineTypePermissionsError,
     InvalidFileKeyError,
-    InvalidSupplierError,
     UnhandledAuditTableError,
     UnhandledSqsError,
     EmptyFileError
 )
-from constants import FileStatus, ERROR_TYPE_TO_STATUS_CODE_MAP, SOURCE_BUCKET_NAME
+from constants import FileNotProcessedReason, FileStatus, ERROR_TYPE_TO_STATUS_CODE_MAP, SOURCE_BUCKET_NAME
 
 
 # NOTE: logging_decorator is applied to handle_record function, rather than lambda_handler, because
@@ -46,6 +45,7 @@ def handle_record(record) -> dict:
 
     vaccine_type = "unknown"
     supplier = "unknown"
+    expiry_timestamp = "not_set"
 
     if bucket_name != SOURCE_BUCKET_NAME:
         return handle_unexpected_bucket_name(bucket_name, file_key)
@@ -73,12 +73,12 @@ def handle_record(record) -> dict:
         permissions = validate_vaccine_type_permissions(vaccine_type=vaccine_type, supplier=supplier)
 
         queue_name = f"{supplier}_{vaccine_type}"
-        upsert_audit_table(
-            message_id, file_key, created_at_formatted_string, expiry_timestamp, queue_name, FileStatus.QUEUED
-        )
         make_and_send_sqs_message(
             file_key, message_id, permissions, vaccine_type, supplier, created_at_formatted_string
         )
+        upsert_audit_table(
+            message_id, file_key, created_at_formatted_string, expiry_timestamp, queue_name, FileStatus.QUEUED
+        )
 
         logger.info("Lambda invocation successful for file '%s'", file_key)
 
@@ -96,18 +96,18 @@ def handle_record(record) -> dict:
         VaccineTypePermissionsError,
         EmptyFileError,
         InvalidFileKeyError,
-        InvalidSupplierError,
         UnhandledAuditTableError,
         UnhandledSqsError,
         Exception,
     ) as error:
         logger.error("Error processing file '%s': %s", file_key, str(error))
 
         queue_name = f"{supplier}_{vaccine_type}"
-        file_status = FileStatus.EMPTY if isinstance(error, EmptyFileError) else FileStatus.PROCESSED
+        file_status = get_file_status_for_error(error)
 
         upsert_audit_table(
-            message_id, file_key, created_at_formatted_string, expiry_timestamp, queue_name, file_status
+            message_id, file_key, created_at_formatted_string, expiry_timestamp, queue_name, file_status,
+            error_details=str(error)
         )
 
         # Create ack file
@@ -129,6 +129,20 @@ def handle_record(record) -> dict:
         }
 
 
+def get_file_status_for_error(error: Exception) -> str:
+    """Creates a file status based on the type of error that was thrown"""
+    if not isinstance(error, (VaccineTypePermissionsError, InvalidFileKeyError, EmptyFileError)):
+        return FileStatus.FAILED
+
+    if isinstance(error, VaccineTypePermissionsError):
+        # TODO - come back and clarify behaviour for client errors
+        return f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.UNAUTHORISED}"
+    elif isinstance(error, EmptyFileError):
+        return f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.EMPTY}"
+
+    return f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.INVALID_FILENAME}"
+
+
 def handle_unexpected_bucket_name(bucket_name: str, file_key: str) -> dict:
     """Handles scenario where Lambda was not invoked by the data-sources bucket. Should not occur due to terraform
     config and overarching design"""

filenameprocessor/src/send_sqs_message.py

@@ -3,17 +3,11 @@
 import os
 from json import dumps as json_dumps
 from clients import sqs_client, logger
-from errors import InvalidSupplierError, UnhandledSqsError
+from errors import UnhandledSqsError
 
 
 def send_to_supplier_queue(message_body: dict, vaccine_type: str, supplier: str) -> None:
     """Sends a message to the supplier queue. Raises an exception if the message is not successfully sent."""
-    # Check the supplier has been identified (this should already have been validated by initial file validation)
-    if not supplier or not vaccine_type:
-        error_message = "Message not sent to supplier queue as unable to identify supplier and/ or vaccine type"
-        logger.error(error_message)
-        raise InvalidSupplierError(error_message)
-
     try:
         queue_url = os.getenv("QUEUE_URL")
         sqs_client.send_message(

filenameprocessor/tests/test_lambda_handler.py

@@ -10,9 +10,9 @@
 from boto3 import client as boto3_client
 from moto import mock_s3, mock_sqs, mock_firehose, mock_dynamodb
 
+from errors import UnhandledSqsError
 from tests.utils_for_tests.generic_setup_and_teardown import GenericSetUp, GenericTearDown
 from tests.utils_for_tests.utils_for_filenameprocessor_tests import (
-    add_entry_to_table,
     assert_audit_table_entry,
     create_mock_hget,
     MOCK_ODS_CODE_TO_SUPPLIER
@@ -215,7 +215,18 @@ def test_lambda_handler_correctly_flags_empty_file(self):
                 None,
             )
 
-        assert_audit_table_entry(file_details, FileStatus.EMPTY)
+        expected_table_items = [
+            {
+                "message_id": {"S": file_details.message_id},
+                "filename": {"S": file_details.file_key},
+                "queue_name": {"S": "RAVS_RSV"},
+                "status": {"S": "Not processed - Empty file"},
+                "error_details": {"S": "Initial file validation failed: batch file was empty"},
+                "timestamp": {"S": file_details.created_at_formatted_string},
+                "expires_at": {"N": str(file_details.expires_at)},
+            }
+        ]
+        self.assertEqual(self.get_audit_table_items(), expected_table_items)
         self.assert_no_sqs_message()
         self.assert_ack_file_contents(file_details)
 
@@ -240,8 +251,8 @@ def test_lambda_handler_non_root_file(self):
 
     def test_lambda_invalid_file_key_no_other_files_in_queue(self):
         """
-        Tests that when the file_key is invalid, and there are no other files in the supplier_vaccineType queue:
-        * The file is added to the audit table with a status of 'Processed'
+        Tests that when the file_key is invalid:
+        * The file is added to the audit table with a status of 'Not processed - Invalid filename'
         * The message is not sent to SQS
         * The failure inf_ack file is created
         """
@@ -265,7 +276,8 @@ def test_lambda_invalid_file_key_no_other_files_in_queue(self):
                 "message_id": {"S": file_details.message_id},
                 "filename": {"S": file_details.file_key},
                 "queue_name": {"S": "unknown_unknown"},
-                "status": {"S": "Processed"},
+                "status": {"S": "Not processed - Invalid filename"},
+                "error_details": {"S": "Initial file validation failed: invalid file key"},
                 "timestamp": {"S": file_details.created_at_formatted_string},
                 "expires_at": {"N": str(file_details.expires_at)},
             }
@@ -275,30 +287,72 @@ def test_lambda_invalid_file_key_no_other_files_in_queue(self):
         self.assert_ack_file_contents(file_details)
         self.assert_no_sqs_message()
 
-    def test_lambda_invalid_permissions_other_files_in_queue(self):
+    def test_lambda_invalid_permissions(self):
         """
-        Tests that when the file permissions are invalid, and there are other files in the supplier_vaccineType queue:
-        * The file is added to the audit table with a status of 'Processed'
+        Tests that when the file permissions are invalid:
+        * The file is added to the audit table with a status of 'Not processed - Unauthorised'
         * The message is not sent to SQS
         * The failure inf_ack file is created
         """
         file_details = MockFileDetails.ravs_rsv_1
         s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key, Body=MOCK_BATCH_FILE_CONTENT)
 
-        queued_file_details = MockFileDetails.ravs_rsv_2
-        add_entry_to_table(queued_file_details, FileStatus.QUEUED)
-
         # Mock the supplier permissions with a value which doesn't include the requested Flu permissions
         mock_hget = create_mock_hget({"X8E5B": "RAVS"}, {})
         with (  # noqa: E999
             patch("file_name_processor.uuid4", return_value=file_details.message_id),  # noqa: E999
             patch("elasticache.redis_client.hget", side_effect=mock_hget),  # noqa: E999
         ):  # noqa: E999
             lambda_handler(self.make_event([self.make_record(file_details.file_key)]), None)
-        assert_audit_table_entry(file_details, FileStatus.PROCESSED)
+
+        expected_table_items = [
+            {
+                "message_id": {"S": file_details.message_id},
+                "filename": {"S": file_details.file_key},
+                "queue_name": {"S": "RAVS_RSV"},
+                "status": {"S": "Not processed - Unauthorised"},
+                "error_details": {"S": "Initial file validation failed: RAVS does not have permissions for RSV"},
+                "timestamp": {"S": file_details.created_at_formatted_string},
+                "expires_at": {"N": str(file_details.expires_at)}
+            }
+        ]
+        self.assertEqual(self.get_audit_table_items(), expected_table_items)
         self.assert_no_sqs_message()
         self.assert_ack_file_contents(file_details)
 
+    def test_lambda_adds_event_to_audit_table_as_failed_when_unexpected_exception_is_caught(self):
+        """
+        Tests that when an unexpected error occurs e.g. sending to SQS (maybe in case of bad deployment):
+        * The file is added to the audit table with a status of 'Failed' and the reason
+        * The message is not sent to SQS
+        * The failure inf_ack file is created
+        """
+        test_file_details = MockFileDetails.emis_flu
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=test_file_details.file_key, Body=MOCK_BATCH_FILE_CONTENT)
+
+        with (  # noqa: E999
+            patch("file_name_processor.uuid4", return_value=test_file_details.message_id),  # noqa: E999
+            patch("file_name_processor.make_and_send_sqs_message", side_effect=UnhandledSqsError(
+                "Some client error with SQS"
+            ))
+        ):  # noqa: E999
+            lambda_handler(self.make_event([self.make_record(test_file_details.file_key)]), None)
+
+        expected_table_items = [
+            {
+                "message_id": {"S": test_file_details.message_id},
+                "filename": {"S": test_file_details.file_key},
+                "queue_name": {"S": "EMIS_FLU"},
+                "status": {"S": "Failed"},
+                "error_details": {"S": "Some client error with SQS"},
+                "timestamp": {"S": test_file_details.created_at_formatted_string},
+                "expires_at": {"N": str(test_file_details.expires_at)},
+            }
+        ]
+        self.assertEqual(self.get_audit_table_items(), expected_table_items)
+        self.assert_ack_file_contents(test_file_details)
+        self.assert_no_sqs_message()
+
 
 @patch.dict("os.environ", MOCK_ENVIRONMENT_DICT)
 @mock_s3

filenameprocessor/tests/test_send_sqs_message.py

@@ -1,7 +1,7 @@
 """Tests for send_sqs_message functions"""
 
 from unittest import TestCase
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch
 from json import loads as json_loads
 from copy import deepcopy
 from moto import mock_sqs
@@ -13,7 +13,7 @@
 # Ensure environment variables are mocked before importing from src files
 with patch.dict("os.environ", MOCK_ENVIRONMENT_DICT):
     from send_sqs_message import send_to_supplier_queue, make_and_send_sqs_message
-    from errors import UnhandledSqsError, InvalidSupplierError
+    from errors import UnhandledSqsError
     from clients import REGION_NAME
 
 sqs_client = boto3_client("sqs", region_name=REGION_NAME)
@@ -76,27 +76,6 @@ def test_send_to_supplier_queue_failure_due_to_queue_does_not_exist(self):
             )
         self.assertEqual(NON_EXISTENT_QUEUE_ERROR_MESSAGE, str(context.exception))
 
-    def test_send_to_supplier_queue_failure_due_to_absent_supplier_or_vaccine_type(self):
-        """Test send_to_supplier_queue function for a failed message send"""
-        # Set up the sqs_queue
-        sqs_client.create_queue(QueueName=Sqs.QUEUE_NAME, Attributes=Sqs.ATTRIBUTES)
-        expected_error_message = (
-            "Message not sent to supplier queue as unable to identify supplier and/ or vaccine type"
-        )
-
-        keys_to_set_to_empty = ["supplier", "vaccine_type"]
-        for key_to_set_to_empty in keys_to_set_to_empty:
-            with self.subTest(f"{key_to_set_to_empty} set to empty string"):
-                mock_sqs_client = MagicMock()
-                with patch("send_sqs_message.sqs_client", mock_sqs_client):
-                    with self.assertRaises(InvalidSupplierError) as context:
-                        message_body = {**FLU_EMIS_FILE_DETAILS.sqs_message_body, key_to_set_to_empty: ""}
-                        vaccine_type = message_body["vaccine_type"]
-                        supplier = message_body["supplier"]
-                        send_to_supplier_queue(message_body, supplier, vaccine_type)
-                self.assertEqual(str(context.exception), expected_error_message)
-                mock_sqs_client.send_message.assert_not_called()
-
     def test_make_and_send_sqs_message_success(self):
         """Test make_and_send_sqs_message function for a successful message send"""
         # Create a mock SQS queue

filenameprocessor/tests/utils_for_tests/utils_for_filenameprocessor_tests.py

@@ -39,7 +39,7 @@ def add_entry_to_table(file_details: MockFileDetails, file_status: FileStatus) -
     dynamodb_client.put_item(TableName=AUDIT_TABLE_NAME, Item=audit_table_entry)
 
 
-def assert_audit_table_entry(file_details: FileDetails, expected_status: FileStatus) -> None:
+def assert_audit_table_entry(file_details: FileDetails, expected_status: str) -> None:
     """Assert that the file details are in the audit table"""
     table_entry = dynamodb_client.get_item(
         TableName=AUDIT_TABLE_NAME, Key={AuditTableKeys.MESSAGE_ID: {"S": file_details.message_id}}