Merge branch 'master' into VED-000-update-devops-user

Author: mfjarvis

filenameprocessor/src/constants.py

@@ -9,7 +9,7 @@
     InvalidSupplierError,
     UnhandledAuditTableError,
     DuplicateFileError,
-    UnhandledSqsError,
+    UnhandledSqsError, EmptyFileError,
 )
 
 SOURCE_BUCKET_NAME = os.getenv("SOURCE_BUCKET_NAME")
@@ -24,13 +24,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"""
@@ -39,6 +43,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_creation_and_expiry_times, move_file
-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
-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,6 +22,7 @@
     InvalidSupplierError,
     UnhandledAuditTableError,
     UnhandledSqsError,
+    EmptyFileError
 )
 from constants import FileStatus, ERROR_TYPE_TO_STATUS_CODE_MAP, SOURCE_BUCKET_NAME
 
@@ -63,9 +64,12 @@ def handle_record(record) -> dict:
 
     try:
         message_id = str(uuid4())
-        created_at_formatted_string, expiry_timestamp = get_creation_and_expiry_times(bucket_name, file_key)
+        s3_response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
+        created_at_formatted_string, expiry_timestamp = get_creation_and_expiry_times(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)
 
         queue_name = f"{supplier}_{vaccine_type}"
@@ -90,6 +94,7 @@ def handle_record(record) -> dict:
 
     except (  # pylint: disable=broad-exception-caught
         VaccineTypePermissionsError,
+        EmptyFileError,
         InvalidFileKeyError,
         InvalidSupplierError,
         UnhandledAuditTableError,
@@ -99,8 +104,10 @@ def handle_record(record) -> dict:
         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
+
         upsert_audit_table(
-            message_id, file_key, created_at_formatted_string, expiry_timestamp, queue_name, FileStatus.PROCESSED
+            message_id, file_key, created_at_formatted_string, expiry_timestamp, queue_name, file_status
         )
 
         # Create ack 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

@@ -4,10 +4,9 @@
 from constants import AUDIT_TABLE_TTL_DAYS
 
 
-def get_creation_and_expiry_times(bucket_name: str, file_key: str) -> (str, int):
+def get_creation_and_expiry_times(s3_response: dict) -> (str, int):
     """Get 'created_at_formatted_string' and 'expires_at' from the response"""
-    response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
-    creation_datetime = response["LastModified"]
+    creation_datetime = s3_response["LastModified"]
     expiry_datetime = creation_datetime + timedelta(days=int(AUDIT_TABLE_TTL_DAYS))
     expiry_timestamp = int(expiry_datetime.timestamp())
     return creation_datetime.strftime("%Y%m%dT%H%M%S00"), expiry_timestamp

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, MOCK_EXPIRES_AT, MockFileDetails
+from tests.utils_for_tests.values_for_tests import MOCK_CREATED_AT_FORMATTED_STRING, MockFileDetails, \
+    MOCK_BATCH_FILE_CONTENT, MOCK_FILE_HEADERS, MOCK_EXPIRES_AT
 
 # Ensure environment variables are mocked before importing from src files
 with patch.dict("os.environ", MOCK_ENVIRONMENT_DICT):
@@ -99,7 +100,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}
@@ -184,7 +185,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):
                 # Set up 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
@@ -195,6 +196,29 @@ def test_lambda_handler_new_file_success_and_first_in_queue(self):
                 self.assert_sqs_message(file_details)
                 self.assert_no_ack_file(file_details)
 
+    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
+
+        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
+        ):  # 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)
+
     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:
@@ -222,7 +246,7 @@ def test_lambda_invalid_file_key_no_other_files_in_queue(self):
         * The failure inf_ack file is created
         """
         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)
@@ -259,7 +283,7 @@ def test_lambda_invalid_permissions_other_files_in_queue(self):
         * 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)
+        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)

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

@@ -35,19 +35,14 @@ def tearDown(self):
 
     def test_get_creation_and_expiry_times(self):
         """Test that get_creation_and_expiry_times 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_created_at = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
-        mock_last_modified = {"LastModified": mock_last_modified_created_at}
+        mock_last_modified_s3_response = {"LastModified": mock_last_modified_created_at}
+
         expected_result_created_at = "20240101T12000000"
         expected_expiry_datetime = mock_last_modified_created_at + timedelta(days=int(AUDIT_TABLE_TTL_DAYS))
         expected_result_expires_at = int(expected_expiry_datetime.timestamp())
 
-        with patch("utils_for_filenameprocessor.s3_client.get_object", return_value=mock_last_modified):
-            created_at_formatted_string, expires_at = get_creation_and_expiry_times(bucket_name, file_key)
+        created_at_formatted_string, expires_at = get_creation_and_expiry_times(mock_last_modified_s3_response)
 
         self.assertEqual(created_at_formatted_string, expected_result_created_at)
         self.assertEqual(expires_at, expected_result_expires_at)

filenameprocessor/tests/utils_for_tests/values_for_tests.py

@@ -78,3 +78,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