VED-902: Extended Attributes (#1011)

* Refactor extended attributes

* check extended attributes and move file

* refactor extended attribute journeys

* rolling back test

* add test for file validation

* refactor filenameprocessor file path into two

* add test for move outside bucket

* add upserts for EA files

* ruff

* fix upsert

* test for no condition_expression

* ACCOUNT_ID in tf

* pathname

* pathname II

* cherry-pick #1013

* resolve e2e batch

* unit tests

* fix to e2e_batch

* cleanup trace

* change buckeowner and destination bucket name

* duplicate batch fix

* duplicate batch test fix

* duplicate batch fix

* change bucket name to dps destination

* revert error message

* bugfix - delete_file()

* bugfix - delete_file() II

* comment

* resolve comments

* fix shared utils

* remove prefix

* remove file validation patching

* remove file validation patching

* ruff

* EXTENDED_ATTRIBUTES_VACC_TYPE

* EXPECTED_BUCKET_OWNER_ACCOUNT

* formatting

* test_unexpected_bucket_name_with_extended_attributes_file

* test_lambda_handler_extended_attributes_failure

* remove copy check, some comments

* remove copy check, some comments II

* improved test_unexpected_bucket_name_and_filename_validation_fails

* expiry_timestamp: int

* ruff

* test_lambda_handler_extended_attributes_failure

---------

Co-authored-by: Akol125 <[email protected]>

Author: JamesW1-NHS

lambdas/filenameprocessor/src/audit_table.py

@@ -15,6 +15,7 @@ def upsert_audit_table(
     queue_name: str,
     file_status: str,
     error_details: Optional[str] = None,
+    condition_expression: Optional[str] = None,
 ) -> None:
     """
     Updates the audit table with the file details
@@ -33,11 +34,17 @@ def upsert_audit_table(
 
     try:
         # Add to the audit table (regardless of whether it is a duplicate)
-        dynamodb_client.put_item(
-            TableName=AUDIT_TABLE_NAME,
-            Item=audit_item,
-            ConditionExpression="attribute_not_exists(message_id)",  # Prevents accidental overwrites
-        )
+        if not condition_expression:
+            dynamodb_client.put_item(
+                TableName=AUDIT_TABLE_NAME,
+                Item=audit_item,
+            )
+        else:
+            dynamodb_client.put_item(
+                TableName=AUDIT_TABLE_NAME,
+                Item=audit_item,
+                ConditionExpression=condition_expression,
+            )
         logger.info(
             "%s file, with message id %s, successfully added to audit table",
             file_key,

lambdas/filenameprocessor/src/constants.py

@@ -12,7 +12,7 @@
 
 SOURCE_BUCKET_NAME = os.getenv("SOURCE_BUCKET_NAME")
 
-# We have used an internal temporary bucket here and an acutal dps bucket will replace this
+# We have used an internal temporary bucket here and an actual dps bucket will replace this
 DPS_DESTINATION_BUCKET_NAME = os.getenv("ACK_BUCKET_NAME")
 EXPECTED_BUCKET_OWNER_ACCOUNT = os.getenv("ACCOUNT_ID")
 AUDIT_TABLE_NAME = os.getenv("AUDIT_TABLE_NAME")
@@ -22,6 +22,7 @@
 VACCINE_TYPE_TO_DISEASES_HASH_KEY = "vacc_to_diseases"
 ODS_CODE_TO_SUPPLIER_SYSTEM_HASH_KEY = "ods_code_to_supplier"
 EXTENDED_ATTRIBUTES_FILE_PREFIX = "Vaccination_Extended_Attributes"
+EXTENDED_ATTRIBUTES_VACC_TYPE = "COVID"
 DPS_DESTINATION_PREFIX = "dps_destination/"
 
 ERROR_TYPE_TO_STATUS_CODE_MAP = {

lambdas/filenameprocessor/src/file_name_processor.py

@@ -10,15 +10,20 @@
 from uuid import uuid4
 
 from audit_table import upsert_audit_table
-from common.aws_s3_utils import move_file, move_file_to_external_bucket
+from common.aws_s3_utils import (
+    copy_file_to_external_bucket,
+    delete_file,
+    move_file,
+)
 from common.clients import STREAM_NAME, get_s3_client, logger
 from common.log_decorator import logging_decorator
 from common.models.errors import UnhandledAuditTableError
 from constants import (
     DPS_DESTINATION_BUCKET_NAME,
-    DPS_DESTINATION_PREFIX,
     ERROR_TYPE_TO_STATUS_CODE_MAP,
+    EXPECTED_BUCKET_OWNER_ACCOUNT,
     EXTENDED_ATTRIBUTES_FILE_PREFIX,
+    EXTENDED_ATTRIBUTES_VACC_TYPE,
     SOURCE_BUCKET_NAME,
     FileNotProcessedReason,
     FileStatus,
@@ -56,8 +61,6 @@ def handle_record(record) -> dict:
             "error": str(error),
         }
 
-    expiry_timestamp = "unknown"
-
     if bucket_name != SOURCE_BUCKET_NAME:
         return handle_unexpected_bucket_name(bucket_name, file_key)
 
@@ -69,10 +72,6 @@ def handle_record(record) -> dict:
         message = "Processing not required. Event was for a file moved to /archive or /processing"
         return {"statusCode": 200, "message": message, "file_key": file_key}
 
-    # Set default values for file-specific variables
-    message_id = "Message id was not created"
-    created_at_formatted_string = "created_at_time not identified"
-
     message_id = str(uuid4())
     s3_response = get_s3_client().get_object(Bucket=bucket_name, Key=file_key)
     created_at_formatted_string, expiry_timestamp = get_creation_and_expiry_times(s3_response)
@@ -108,7 +107,8 @@ def handle_unexpected_bucket_name(bucket_name: str, file_key: str) -> dict:
     config and overarching design"""
     try:
         if file_key.startswith(EXTENDED_ATTRIBUTES_FILE_PREFIX):
-            extended_attribute_identifier = validate_extended_attributes_file_key(file_key)
+            organization_code = validate_extended_attributes_file_key(file_key)
+            extended_attribute_identifier = f"{organization_code}_{EXTENDED_ATTRIBUTES_VACC_TYPE}"
             logger.error(
                 "Unable to process file %s due to unexpected bucket name %s",
                 file_key,
@@ -157,7 +157,7 @@ def handle_unexpected_bucket_name(bucket_name: str, file_key: str) -> dict:
 
 
 def handle_batch_file(
-    file_key: str, bucket_name: str, message_id: str, created_at_formatted_string: str, expiry_timestamp: str
+    file_key: str, bucket_name: str, message_id: str, created_at_formatted_string: str, expiry_timestamp: int
 ) -> dict:
     """
     Processes a single record for batch file.
@@ -177,6 +177,7 @@ def handle_batch_file(
             expiry_timestamp,
             queue_name,
             FileStatus.QUEUED,
+            condition_expression="attribute_not_exists(message_id)",  # Prevents accidental overwrites
         )
         make_and_send_sqs_message(
             file_key,
@@ -239,17 +240,17 @@ def handle_batch_file(
 
 
 def handle_extended_attributes_file(
-    file_key: str, bucket_name: str, message_id: str, created_at_formatted_string: str, expiry_timestamp: str
+    file_key: str, bucket_name: str, message_id: str, created_at_formatted_string: str, expiry_timestamp: int
 ) -> dict:
     """
     Processes a single record for extended attributes file.
     Returns a dictionary containing information to be included in the logs.
     """
+
+    extended_attribute_identifier = None
     try:
-        extended_attribute_identifier = validate_extended_attributes_file_key(file_key)
-        move_file_to_external_bucket(
-            bucket_name, file_key, DPS_DESTINATION_BUCKET_NAME, f"{DPS_DESTINATION_PREFIX}{file_key}"
-        )
+        organization_code = validate_extended_attributes_file_key(file_key)
+        extended_attribute_identifier = f"{organization_code}_{EXTENDED_ATTRIBUTES_VACC_TYPE}"
 
         upsert_audit_table(
             message_id,
@@ -259,6 +260,28 @@ def handle_extended_attributes_file(
             extended_attribute_identifier,
             FileStatus.PROCESSING,
         )
+
+        # TODO: agree the prefix with DPS
+        dest_file_key = f"dps_destination/{file_key}"
+        copy_file_to_external_bucket(
+            bucket_name,
+            file_key,
+            DPS_DESTINATION_BUCKET_NAME,
+            dest_file_key,
+            EXPECTED_BUCKET_OWNER_ACCOUNT,
+            EXPECTED_BUCKET_OWNER_ACCOUNT,
+        )
+        delete_file(bucket_name, dest_file_key, EXPECTED_BUCKET_OWNER_ACCOUNT)
+
+        upsert_audit_table(
+            message_id,
+            file_key,
+            created_at_formatted_string,
+            expiry_timestamp,
+            extended_attribute_identifier,
+            FileStatus.PROCESSED,
+        )
+
         return {
             "statusCode": 200,
             "message": "Extended Attributes file successfully processed",
@@ -276,7 +299,13 @@ def handle_extended_attributes_file(
         logger.error("Error processing file '%s': %s", file_key, str(error))
 
         file_status = get_file_status_for_error(error)
-        extended_attribute_identifier = validate_extended_attributes_file_key(file_key)
+
+        # NB if we got InvalidFileKeyError we won't have a valid queue name
+        if not extended_attribute_identifier:
+            extended_attribute_identifier = "unknown"
+
+        # Move file to archive
+        move_file(bucket_name, file_key, f"archive/{file_key}")
 
         upsert_audit_table(
             message_id,

lambdas/filenameprocessor/src/file_validation.py

@@ -40,14 +40,14 @@ def is_valid_datetime(timestamp: str) -> bool:
 def validate_extended_attributes_file_key(file_key: str) -> str:
     """
     Checks that all elements of the file key are valid, raises an exception otherwise.
-    Returns a string containing the organization code and COVID vaccine type needed in the audit table.
+    Returns a string containing the organization code.
     """
     if not match(r"^[^_.]*_[^_.]*_[^_.]*_[^_.]*_[^_.]*_[^_.]*_[^_.]*", file_key):
         raise InvalidFileKeyError("Initial file validation failed: invalid extended attributes file key format")
 
     file_key_parts_without_extension, _ = split_file_key(file_key)
     organization_code = file_key_parts_without_extension[5]
-    return f"{organization_code}_COVID"
+    return organization_code
 
 
 def validate_batch_file_key(file_key: str) -> tuple[str, str]:

lambdas/filenameprocessor/tests/test_audit_table.py

@@ -73,6 +73,7 @@ def test_upsert_audit_table_with_duplicate_message_id_raises_exception(self):
             queue_name=ravs_rsv_test_file.queue_name,
             file_status=FileStatus.PROCESSED,
             expiry_timestamp=ravs_rsv_test_file.expires_at,
+            condition_expression="attribute_not_exists(message_id)",
         )
 
         assert_audit_table_entry(ravs_rsv_test_file, FileStatus.PROCESSED)
@@ -85,4 +86,32 @@ def test_upsert_audit_table_with_duplicate_message_id_raises_exception(self):
                 queue_name=ravs_rsv_test_file.queue_name,
                 file_status=FileStatus.PROCESSED,
                 expiry_timestamp=ravs_rsv_test_file.expires_at,
+                condition_expression="attribute_not_exists(message_id)",
             )
+
+    def test_upsert_audit_table_with_duplicate_message_id_no_condition(self):
+        """Test that attempting to create an entry with a message_id that already exists causes no exception
+        if the condition_expression is not set"""
+        ravs_rsv_test_file = FileDetails("RAVS", "RSV", "YGM41", file_number=1)
+
+        upsert_audit_table(
+            message_id=ravs_rsv_test_file.message_id,
+            file_key=ravs_rsv_test_file.file_key,
+            created_at_formatted_str=ravs_rsv_test_file.created_at_formatted_string,
+            queue_name=ravs_rsv_test_file.queue_name,
+            file_status=FileStatus.PROCESSING,
+            expiry_timestamp=ravs_rsv_test_file.expires_at,
+        )
+
+        assert_audit_table_entry(ravs_rsv_test_file, FileStatus.PROCESSING)
+
+        upsert_audit_table(
+            message_id=ravs_rsv_test_file.message_id,
+            file_key=ravs_rsv_test_file.file_key,
+            created_at_formatted_str=ravs_rsv_test_file.created_at_formatted_string,
+            queue_name=ravs_rsv_test_file.queue_name,
+            file_status=FileStatus.PROCESSED,
+            expiry_timestamp=ravs_rsv_test_file.expires_at,
+        )
+
+        assert_audit_table_entry(ravs_rsv_test_file, FileStatus.PROCESSED)

lambdas/filenameprocessor/tests/test_file_key_validation.py

@@ -114,18 +114,18 @@ def test_split_file_key(self, _):
                 self.assertEqual(split_file_key(file_key), expected)
 
     def test_validate_extended_attributes_file_key(self, _):
-        """Tests that validate_extended_attributes_file_key returns organization code and COVID vaccine type if all
+        """Tests that validate_extended_attributes_file_key returns organization code if all
         elements pass validation, and raises an exception otherwise"""
         test_cases_for_success_scenarios = [
             # Valid extended attributes file key
             (
                 "Vaccination_Extended_Attributes_v1_5_X8E5B_20000101T00000001.csv",
-                "X8E5B_COVID",
+                "X8E5B",
             ),
             # Valid extended attributes file key with different organization code
             (
                 "Vaccination_Extended_Attributes_v1_5_YGM41_20221231T23595999.csv",
-                "YGM41_COVID",
+                "YGM41",
             ),
         ]