VED-757 Handle empty batch files (#801)

Author: dlzhry2nhs

filenameprocessor/src/constants.py

@@ -9,7 +9,7 @@
     InvalidSupplierError,
     UnhandledAuditTableError,
     DuplicateFileError,
-    UnhandledSqsError,
+    UnhandledSqsError, EmptyFileError,
 )
 
 SOURCE_BUCKET_NAME = os.getenv("SOURCE_BUCKET_NAME")
@@ -27,13 +27,17 @@
 ERROR_TYPE_TO_STATUS_CODE_MAP = {
     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,
 }
 
+# The size in bytes of an empty batch file containing only the headers row
+EMPTY_BATCH_FILE_SIZE_IN_BYTES = 614
+
 
 class FileStatus(StrEnum):
     """File status constants"""
@@ -42,6 +46,7 @@ class FileStatus(StrEnum):
     PROCESSING = "Processing"
     PROCESSED = "Processed"
     DUPLICATE = "Not processed - duplicate"
+    EMPTY = "Not processed - empty file"
 
 
 class AuditTableKeys(StrEnum):

filenameprocessor/src/errors.py

@@ -5,6 +5,10 @@ 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"""
+
+
 class UnhandledAuditTableError(Exception):
     """A custom exception for when an unexpected error occurs whilst adding the file to the audit table."""
 

filenameprocessor/src/file_name_processor.py

@@ -9,11 +9,11 @@
 import argparse
 from uuid import uuid4
 from utils_for_filenameprocessor import get_created_at_formatted_string, move_file, invoke_filename_lambda
-from file_key_validation import validate_file_key, is_file_in_directory_root
+from file_validation import validate_file_key, is_file_in_directory_root, validate_file_not_empty
 from send_sqs_message import make_and_send_sqs_message
 from make_and_upload_ack_file import make_and_upload_the_ack_file
 from audit_table import upsert_audit_table, get_next_queued_file_details, ensure_file_is_not_a_duplicate
-from clients import logger
+from clients import logger, s3_client
 from logging_decorator import logging_decorator
 from supplier_permissions import validate_vaccine_type_permissions
 from errors import (
@@ -22,7 +22,7 @@
     InvalidSupplierError,
     UnhandledAuditTableError,
     DuplicateFileError,
-    UnhandledSqsError,
+    UnhandledSqsError, EmptyFileError,
 )
 from constants import FileStatus, DATA_SOURCES_BUCKET_SUFFIX, ERROR_TYPE_TO_STATUS_CODE_MAP
 
@@ -67,11 +67,15 @@ def handle_record(record) -> dict:
 
             # Get message_id if the file is not new, else assign one
             message_id = record.get("message_id", str(uuid4()))
+            s3_response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
 
-            created_at_formatted_string = get_created_at_formatted_string(bucket_name, file_key)
+            created_at_formatted_string = get_created_at_formatted_string(s3_response)
 
             vaccine_type, supplier = validate_file_key(file_key)
+            # VED-757: Known issue with suppliers sometimes sending empty files
+            validate_file_not_empty(s3_response)
             permissions = validate_vaccine_type_permissions(vaccine_type=vaccine_type, supplier=supplier)
+
             if not is_existing_file:
                 ensure_file_is_not_a_duplicate(file_key, created_at_formatted_string)
 
@@ -106,12 +110,18 @@ def handle_record(record) -> dict:
             InvalidSupplierError,
             UnhandledAuditTableError,
             DuplicateFileError,
+            EmptyFileError,
             UnhandledSqsError,
             Exception,
         ) as error:
             logger.error("Error processing file '%s': %s", file_key, str(error))
 
-            file_status = FileStatus.DUPLICATE if isinstance(error, DuplicateFileError) else FileStatus.PROCESSED
+            file_status = FileStatus.PROCESSED
+            if isinstance(error, DuplicateFileError):
+                file_status = FileStatus.DUPLICATE
+            elif isinstance(error, EmptyFileError):
+                file_status = FileStatus.EMPTY
+
             queue_name = f"{supplier}_{vaccine_type}"
             upsert_audit_table(
                 message_id, file_key, created_at_formatted_string, queue_name, file_status, is_existing_file

…nameprocessor/src/file_key_validation.py filenameprocessor/src/file_validation.pyfilenameprocessor/src/file_key_validation.py renamed to filenameprocessor/src/file_validation.py filenameprocessor/src/file_key_validation.py renamed to filenameprocessor/src/file_validation.py

@@ -2,9 +2,9 @@
 
 from re import match
 from datetime import datetime
-from constants import VALID_VERSIONS
+from constants import VALID_VERSIONS, EMPTY_BATCH_FILE_SIZE_IN_BYTES
 from elasticache import get_valid_vaccine_types_from_cache, get_supplier_system_from_cache
-from errors import InvalidFileKeyError
+from errors import InvalidFileKeyError, EmptyFileError
 
 
 def is_file_in_directory_root(file_key: str) -> bool:
@@ -68,3 +68,9 @@ def validate_file_key(file_key: str) -> tuple[str, str]:
         raise InvalidFileKeyError("Initial file validation failed: invalid file key")
 
     return vaccine_type, supplier
+
+
+def validate_file_not_empty(s3_response: dict) -> None:
+    """Checks that the batch file from S3 is not empty or containing only the header row"""
+    if s3_response.get("ContentLength", 0) <= EMPTY_BATCH_FILE_SIZE_IN_BYTES:
+        raise EmptyFileError("Initial file validation failed: batch file was empty")

filenameprocessor/src/utils_for_filenameprocessor.py

@@ -5,10 +5,9 @@
 from clients import s3_client, logger, lambda_client
 
 
-def get_created_at_formatted_string(bucket_name: str, file_key: str) -> str:
+def get_created_at_formatted_string(s3_response: dict) -> str:
     """Get the created_at_formatted_string from the response"""
-    response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
-    return response["LastModified"].strftime("%Y%m%dT%H%M%S00")
+    return s3_response["LastModified"].strftime("%Y%m%dT%H%M%S00")
 
 
 def move_file(bucket_name: str, source_file_key: str, destination_file_key: str) -> None:

filenameprocessor/tests/test_file_key_validation.py

@@ -9,7 +9,7 @@
 
 # Ensure environment variables are mocked before importing from src files
 with patch.dict("os.environ", MOCK_ENVIRONMENT_DICT):
-    from file_key_validation import is_file_in_directory_root, is_valid_datetime, validate_file_key
+    from file_validation import is_file_in_directory_root, is_valid_datetime, validate_file_key
     from errors import InvalidFileKeyError
 
 VALID_FLU_EMIS_FILE_KEY = MockFileDetails.emis_flu.file_key

filenameprocessor/tests/test_lambda_handler.py

@@ -18,7 +18,8 @@
     MOCK_ODS_CODE_TO_SUPPLIER
 )
 from tests.utils_for_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames, Sqs
-from tests.utils_for_tests.values_for_tests import MOCK_CREATED_AT_FORMATTED_STRING, MockFileDetails
+from tests.utils_for_tests.values_for_tests import MOCK_CREATED_AT_FORMATTED_STRING, MockFileDetails, \
+    MOCK_BATCH_FILE_CONTENT, MOCK_FILE_HEADERS
 
 # Ensure environment variables are mocked before importing from src files
 with patch.dict("os.environ", MOCK_ENVIRONMENT_DICT):
@@ -92,7 +93,7 @@ def make_record(file_key: str):
     @staticmethod
     def make_record_with_message_id(file_key: str, message_id: str):
         """
-        Makes a record which includes a message_id, with the s3 bucket name set to BucketNames.SOURCE and and
+        Makes a record which includes a message_id, with the s3 bucket name set to BucketNames.SOURCE and
         s3 object key set to the file_key.
         """
         return {"s3": {"bucket": {"name": BucketNames.SOURCE}, "object": {"key": file_key}}, "message_id": message_id}
@@ -180,7 +181,7 @@ def test_lambda_handler_new_file_success_and_first_in_queue(self):
         for file_details in test_cases:
             with self.subTest(file_details.name):
                 # Setup the file in the source bucket
-                s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key)
+                s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key, Body=MOCK_BATCH_FILE_CONTENT)
 
                 with (  # noqa: E999
                     patch("file_name_processor.uuid4", return_value=file_details.message_id),  # noqa: E999
@@ -208,7 +209,7 @@ def test_lambda_handler_new_file_success_and_other_files_in_queue(self):
         file_details = MockFileDetails.ravs_rsv_1
         file_already_processing_details = MockFileDetails.ravs_rsv_2
 
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key)
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key, Body=MOCK_BATCH_FILE_CONTENT)
 
         add_entry_to_table(file_already_processing_details, FileStatus.PROCESSING)
 
@@ -234,6 +235,8 @@ def test_lambda_handler_existing_file(self):
         file_details = MockFileDetails.ravs_rsv_1
         add_entry_to_table(file_details, FileStatus.QUEUED)
 
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key, Body=MOCK_BATCH_FILE_CONTENT)
+
         with (  # noqa: E999
             patch("file_name_processor.uuid4", return_value=file_details.message_id),  # noqa: E999
             patch("file_name_processor.invoke_filename_lambda") as mock_invoke_filename_lambda,  # noqa: E999
@@ -248,6 +251,32 @@ def test_lambda_handler_existing_file(self):
         self.assert_no_ack_file(file_details)
         mock_invoke_filename_lambda.assert_not_called()
 
+    def test_lambda_handler_correctly_flags_empty_file(self):
+        """
+        VED-757 Tests that for an empty batch file:
+        * The file status is updated to 'Not processed - empty file' in the audit table
+        * The message is not sent to SQS
+        * The failure inf_ack file is created
+        """
+        file_details = MockFileDetails.ravs_rsv_1
+        add_entry_to_table(file_details, FileStatus.QUEUED)
+
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key, Body=MOCK_FILE_HEADERS)
+
+        with (  # noqa: E999
+            patch("file_name_processor.uuid4", return_value=file_details.message_id),  # noqa: E999
+            patch("file_name_processor.invoke_filename_lambda") as mock_invoke_filename_lambda,  # noqa: E999
+        ):  # noqa: E999
+            lambda_handler(
+                self.make_event([self.make_record_with_message_id(file_details.file_key, file_details.message_id)]),
+                None,
+            )
+
+        assert_audit_table_entry(file_details, FileStatus.EMPTY)
+        self.assert_no_sqs_message()
+        self.assert_ack_file_contents(file_details)
+        mock_invoke_filename_lambda.assert_not_called()
+
     def test_lambda_handler_non_root_file(self):
         """
         Tests that when the file is not in the root of the source bucket, no action is taken:
@@ -280,7 +309,7 @@ def test_lambda_handler_duplicate_file_other_files_in_queue(self):
         * The invoke_filename_lambda method is not called
         """
         file_details = MockFileDetails.ravs_rsv_1
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key)
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key, Body=MOCK_BATCH_FILE_CONTENT)
 
         duplicate_already_in_table = deepcopy(file_details)
         duplicate_already_in_table.message_id = "duplicate_id"
@@ -312,7 +341,7 @@ def test_lambda_invalid_file_key_no_other_files_in_queue(self):
         * The invoke_filename_lambda method is not called
         """
         invalid_file_key = "InvalidVaccineType_Vaccinations_v5_YGM41_20240708T12130100.csv"
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=invalid_file_key)
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=invalid_file_key, Body=MOCK_BATCH_FILE_CONTENT)
         file_details = deepcopy(MockFileDetails.ravs_rsv_1)
         file_details.file_key = invalid_file_key
         file_details.ack_file_key = self.get_ack_file_key(invalid_file_key)
@@ -351,7 +380,7 @@ def test_lambda_invalid_permissions_other_files_in_queue(self):
         * The invoke_filename_lambda method is called with queued file details
         """
         file_details = MockFileDetails.ravs_rsv_1
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=file_details.file_key)
+        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)
@@ -405,9 +434,10 @@ def test_lambda_handler_multiple_records_for_same_queue(self):
         invalid_file_details_3.audit_table_entry[AuditTableKeys.QUEUE_NAME] = {"S": invalid_file_details_3.queue_name}
         invalid_file_details_3.sqs_message_body["filename"] = invalid_file_details_3.file_key
 
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=valid_file_details_1.file_key)
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=valid_file_details_2.file_key)
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=invalid_file_details_3.file_key)
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=valid_file_details_1.file_key, Body=MOCK_BATCH_FILE_CONTENT)
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=valid_file_details_2.file_key, Body=MOCK_BATCH_FILE_CONTENT)
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=invalid_file_details_3.file_key,
+                             Body=MOCK_BATCH_FILE_CONTENT)
 
         message_ids = [
             valid_file_details_1.message_id,

filenameprocessor/tests/test_logging_decorator.py

@@ -10,7 +10,7 @@
 
 from tests.utils_for_tests.generic_setup_and_teardown import GenericSetUp, GenericTearDown
 from tests.utils_for_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames, Firehose
-from tests.utils_for_tests.values_for_tests import MockFileDetails, fixed_datetime
+from tests.utils_for_tests.values_for_tests import MockFileDetails, fixed_datetime, MOCK_BATCH_FILE_CONTENT
 from tests.utils_for_tests.utils_for_filenameprocessor_tests import create_mock_hget
 
 # Ensure environment variables are mocked before importing from src files
@@ -41,7 +41,7 @@ class TestLoggingDecorator(unittest.TestCase):
     def setUp(self):
         """Set up the mock AWS environment and upload a valid FLU/EMIS file example"""
         GenericSetUp(s3_client, firehose_client, sqs_client, dynamodb_client)
-        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=FILE_DETAILS.file_key)
+        s3_client.put_object(Bucket=BucketNames.SOURCE, Key=FILE_DETAILS.file_key, Body=MOCK_BATCH_FILE_CONTENT)
 
     def tearDown(self):
         """Clean the mock AWS environment"""

filenameprocessor/tests/test_utils_for_filenameprocessor.py

@@ -37,16 +37,10 @@ def tearDown(self):
 
     def test_get_created_at_formatted_string(self):
         """Test that get_created_at_formatted_string can correctly get the created_at_formatted_string"""
-        bucket_name = BucketNames.SOURCE
-        file_key = "test_file_key"
-
-        s3_client.put_object(Bucket=bucket_name, Key=file_key)
-
-        mock_last_modified = {"LastModified": datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)}
+        mock_last_modified_s3_response = {"LastModified": datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)}
         expected_result = "20240101T12000000"
 
-        with patch("utils_for_filenameprocessor.s3_client.get_object", return_value=mock_last_modified):
-            created_at_formatted_string = get_created_at_formatted_string(bucket_name, file_key)
+        created_at_formatted_string = get_created_at_formatted_string(mock_last_modified_s3_response)
 
         self.assertEqual(created_at_formatted_string, expected_result)
 

filenameprocessor/tests/utils_for_tests/values_for_tests.py

@@ -75,3 +75,29 @@ class MockFileDetails:
     emis_flu = FileDetails("EMIS", "FLU", "YGM41")
     emis_rsv = FileDetails("EMIS", "RSV", "YGM41")
     ravs_flu = FileDetails("RAVS", "FLU", "X8E5B")
+
+
+MOCK_FILE_HEADERS = (
+    "NHS_NUMBER|PERSON_FORENAME|PERSON_SURNAME|PERSON_DOB|PERSON_GENDER_CODE|PERSON_POSTCODE|"
+    "DATE_AND_TIME|SITE_CODE|SITE_CODE_TYPE_URI|UNIQUE_ID|UNIQUE_ID_URI|ACTION_FLAG|"
+    "PERFORMING_PROFESSIONAL_FORENAME|PERFORMING_PROFESSIONAL_SURNAME|RECORDED_DATE|"
+    "PRIMARY_SOURCE|VACCINATION_PROCEDURE_CODE|VACCINATION_PROCEDURE_TERM|DOSE_SEQUENCE|"
+    "VACCINE_PRODUCT_CODE|VACCINE_PRODUCT_TERM|VACCINE_MANUFACTURER|BATCH_NUMBER|EXPIRY_DATE|"
+    "SITE_OF_VACCINATION_CODE|SITE_OF_VACCINATION_TERM|ROUTE_OF_VACCINATION_CODE|"
+    "ROUTE_OF_VACCINATION_TERM|DOSE_AMOUNT|DOSE_UNIT_CODE|DOSE_UNIT_TERM|INDICATION_CODE|"
+    "LOCATION_CODE|LOCATION_CODE_TYPE_URI"
+) + "\n"
+
+# The content does not matter for the filename processor Lambda. It only needs to ensure the file is not empty.
+# Validation of the headers and contents happens later in the batch flow
+MOCK_FILE_DATA = (
+    '9674963871|"ALAN"|"PARTRIDGE"|"20190131"|"2"|"AA14 6AA"|"20240610T183325"|"J82067"|'
+    '"https://fhir.nhs.uk/Id/ods-organization-code"|"1234"|"a_system"|"new"|"Ellena"|"O\'Reilly"|"20240101"|"TRUE"|'
+    '"1303503001"|"Administration of vaccine product containing only Human orthopneumovirus antigen (procedure)"|'
+    '1|"42605811000001109"|"Abrysvo vaccine powder and solvent for solution for injection 0.5ml vials (Pfizer Ltd) '
+    '(product)"|"Pfizer"|"RSVTEST"|"20241231"|"368208006"|"Left upper arm structure (body structure)"|'
+    '"78421000"|"Intramuscular route (qualifier value)"|"0.5"|"258773002"|"Milliliter (qualifier value)"|"Test"|'
+    '"J82067"|"https://fhir.nhs.uk/Id/ods-organization-code"'
+)
+
+MOCK_BATCH_FILE_CONTENT = MOCK_FILE_HEADERS + MOCK_FILE_DATA