Refactored and updated tests for batch processor

Author: dlzhry2nhs

filenameprocessor/src/file_name_processor.py

@@ -47,7 +47,7 @@ def handle_record(record) -> dict:
     supplier = "unknown"
 
     if bucket_name != SOURCE_BUCKET_NAME:
-        return handle_unexpected_bucket_name(bucket_name, file_key, vaccine_type, supplier)
+        return handle_unexpected_bucket_name(bucket_name, file_key)
 
     # In addition to when a batch file is added to the S3 bucket root for processing, this Lambda is also invoked
     # when the file is moved to the processing/ directory and finally the /archive directory. We want to ignore
@@ -122,7 +122,7 @@ def handle_record(record) -> dict:
         }
 
 
-def handle_unexpected_bucket_name(bucket_name: str, file_key: str, vaccine_type: str, supplier: str) -> dict:
+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"""
     try:
@@ -139,7 +139,7 @@ def handle_unexpected_bucket_name(bucket_name: str, file_key: str, vaccine_type:
         message = f"Failed to process file due to unexpected bucket name {bucket_name} and file key {file_key}"
 
         return {"statusCode": 500, "message": message, "file_key": file_key,
-                "vaccine_type": vaccine_type, "supplier": supplier, "error": str(error)}
+                "vaccine_type": "unknown", "supplier": "unknown", "error": str(error)}
 
 
 def lambda_handler(event: dict, context) -> None:  # pylint: disable=unused-argument

recordprocessor/Makefile

@@ -4,16 +4,17 @@ build:
 package:build
 	mkdir -p build
 	docker run --rm -v $(shell pwd)/build:/build processor-lambda-build
+
 test:
-	python -m unittest	
+	@PYTHONPATH=src:tests python -m unittest
 
-coverage run:
-	coverage run -m unittest discover
+coverage-run:
+	@PYTHONPATH=src:tests coverage run -m unittest discover
 
-coverage report:
-	coverage report -m 
+coverage-report:
+	coverage report -m
 
-coverage html:
+coverage-html:
 	coverage html
 
 .PHONY: build package

recordprocessor/src/audit_table.py

@@ -1,49 +1,30 @@
 """Add the filename to the audit table and check for duplicates."""
 
-from typing import Union
-from boto3.dynamodb.conditions import Key
-from clients import dynamodb_client, dynamodb_resource, logger
+from clients import dynamodb_client, logger
 from errors import UnhandledAuditTableError
-from constants import AUDIT_TABLE_NAME, AUDIT_TABLE_QUEUE_NAME_GSI, AuditTableKeys, FileStatus
+from constants import AUDIT_TABLE_NAME, AuditTableKeys
 
 
-def change_audit_table_status_to_processed(file_key: str, message_id: str) -> None:
-    """Updates the status in the audit table to 'Processed' and returns the queue name."""
+def update_audit_table_status(file_key: str, message_id: str, status: str) -> None:
+    """Updates the status in the audit table to the requested value"""
     try:
         # Update the status in the audit table to "Processed"
         dynamodb_client.update_item(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": message_id}},
             UpdateExpression="SET #status = :status",
             ExpressionAttributeNames={"#status": "status"},
-            ExpressionAttributeValues={":status": {"S": FileStatus.PROCESSED}},
+            ExpressionAttributeValues={":status": {"S": status}},
             ConditionExpression="attribute_exists(message_id)",
         )
 
         logger.info(
             "The status of %s file, with message id %s, was successfully updated to %s in the audit table",
             file_key,
             message_id,
-            FileStatus.PROCESSED,
+            status
         )
 
     except Exception as error:  # pylint: disable = broad-exception-caught
         logger.error(error)
         raise UnhandledAuditTableError(error) from error
-
-
-def get_next_queued_file_details(queue_name: str) -> Union[dict, None]:
-    """
-    Checks for queued files.
-    Returns a dictionary containing the details of the oldest queued file, or returns None if no queued files are found.
-    """
-    queued_files_found_in_audit_table: dict = dynamodb_resource.Table(AUDIT_TABLE_NAME).query(
-        IndexName=AUDIT_TABLE_QUEUE_NAME_GSI,
-        KeyConditionExpression=Key(AuditTableKeys.QUEUE_NAME).eq(queue_name)
-        & Key(AuditTableKeys.STATUS).eq(FileStatus.QUEUED),
-    )
-
-    queued_files_details: list = queued_files_found_in_audit_table["Items"]
-
-    # Return the oldest queued file
-    return sorted(queued_files_details, key=lambda x: x["timestamp"])[0] if queued_files_details else None

recordprocessor/src/batch_processor.py

@@ -3,8 +3,11 @@
 import json
 import os
 import time
+
+from constants import FileStatus
 from process_row import process_row
 from mappings import map_target_disease
+from audit_table import update_audit_table_status
 from send_to_kinesis import send_to_kinesis
 from clients import logger
 from file_level_validation import file_level_validation
@@ -55,6 +58,8 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
 
         logger.info("Total rows processed: %s", row_count)
 
+    update_audit_table_status(file_key, file_id, FileStatus.PREPROCESSED)
+
 
 def main(event: str) -> None:
     """Process each row of the file"""

recordprocessor/src/constants.py

@@ -53,7 +53,8 @@ class FileStatus:
 
     QUEUED = "Queued"
     PROCESSING = "Processing"
-    PROCESSED = "Processed"
+    PREPROCESSED = "Preprocessed"  # All entries in file converted to FHIR and forwarded to Kinesis
+    PROCESSED = "Processed"  # All entries processed and ack file created
     DUPLICATE = "Duplicate"
 
 

recordprocessor/src/file_level_validation.py

@@ -4,11 +4,11 @@
 """
 from clients import logger, s3_client
 from make_and_upload_ack_file import make_and_upload_ack_file
-from utils_for_recordprocessor import get_csv_content_dict_reader, invoke_filename_lambda
+from utils_for_recordprocessor import get_csv_content_dict_reader
 from errors import InvalidHeaders, NoOperationPermissions
 from logging_decorator import file_level_validation_logging_decorator
-from audit_table import change_audit_table_status_to_processed, get_next_queued_file_details
-from constants import SOURCE_BUCKET_NAME, EXPECTED_CSV_HEADERS, permission_to_operation_map, Permission
+from audit_table import update_audit_table_status
+from constants import SOURCE_BUCKET_NAME, EXPECTED_CSV_HEADERS, permission_to_operation_map, FileStatus, Permission
 
 
 def validate_content_headers(csv_content_reader) -> None:
@@ -64,7 +64,7 @@ def move_file(bucket_name: str, source_file_key: str, destination_file_key: str)
 def file_level_validation(incoming_message_body: dict) -> dict:
     """
     Validates that the csv headers are correct and that the supplier has permission to perform at least one of
-    the requested operations. Uploades the inf ack file and moves the source file to the processing folder.
+    the requested operations. Uploads the inf ack file and moves the source file to the processing folder.
     Returns an interim message body for row level processing.
     NOTE: If file level validation fails the source file is moved to the archive folder, the audit table is updated
     to reflect the file has been processed and the filename lambda is invoked with the next file in the queue.
@@ -113,10 +113,6 @@ def file_level_validation(incoming_message_body: dict) -> dict:
         except Exception as move_file_error:
             logger.error("Failed to move file to archive: %s", move_file_error)
 
-        # Update the audit table and invoke the filename lambda with next file in the queue (if one exists)
-        change_audit_table_status_to_processed(file_key, message_id)
-        queue_name = f"{supplier}_{vaccine}"
-        next_queued_file_details = get_next_queued_file_details(queue_name)
-        if next_queued_file_details:
-            invoke_filename_lambda(next_queued_file_details["filename"], next_queued_file_details["message_id"])
+        # Update the audit table
+        update_audit_table_status(file_key, message_id, FileStatus.PROCESSED)
         raise

recordprocessor/src/utils_for_recordprocessor.py

@@ -1,11 +1,9 @@
 """Utils for filenameprocessor lambda"""
 
 import os
-import json
 from csv import DictReader
 from io import TextIOWrapper
-from clients import s3_client, lambda_client, logger
-from constants import SOURCE_BUCKET_NAME, FILE_NAME_PROC_LAMBDA_NAME
+from clients import s3_client
 
 
 def get_environment() -> str:
@@ -26,19 +24,3 @@ def get_csv_content_dict_reader(file_key: str) -> DictReader:
 def create_diagnostics_dictionary(error_type, status_code, error_message) -> dict:
     """Returns a dictionary containing the error_type, statusCode, and error_message"""
     return {"error_type": error_type, "statusCode": status_code, "error_message": error_message}
-
-
-def invoke_filename_lambda(file_key: str, message_id: str) -> None:
-    """Invokes the filenameprocessor lambda with the given file key and message id"""
-    try:
-        lambda_payload = {
-            "Records": [
-                {"s3": {"bucket": {"name": SOURCE_BUCKET_NAME}, "object": {"key": file_key}}, "message_id": message_id}
-            ]
-        }
-        lambda_client.invoke(
-            FunctionName=FILE_NAME_PROC_LAMBDA_NAME, InvocationType="Event", Payload=json.dumps(lambda_payload)
-        )
-    except Exception as error:
-        logger.error("Error invoking filename lambda: %s", error)
-        raise

recordprocessor/tests/test_audit_table.py

@@ -1,5 +1,5 @@
 """Tests for audit_table functions"""
-
+import unittest
 from unittest import TestCase
 from unittest.mock import patch
 from boto3 import client as boto3_client
@@ -10,7 +10,6 @@
 from tests.utils_for_recordprocessor_tests.generic_setup_and_teardown import GenericSetUp, GenericTearDown
 from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import MockFileDetails, FileDetails
 from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import (
-    deserialize_dynamodb_types,
     add_entry_to_table,
 )
 
@@ -21,7 +20,7 @@
         FileStatus,
     )
 
-    from audit_table import get_next_queued_file_details, change_audit_table_status_to_processed
+    from audit_table import update_audit_table_status
     from clients import REGION_NAME
 
 
@@ -49,60 +48,32 @@ def get_table_items() -> list:
 
         return dynamodb_client.scan(TableName=AUDIT_TABLE_NAME).get("Items", [])
 
-    def test_get_next_queued_file_details(self):
-        """Test that the get_next_queued_file_details function returns the correct file details"""
-        # NOTE: Throughout this test the assertions will be checking for the next queued RAVS_RSV file.
-        queue_to_check = "RAVS_RSV"
-
-        # Test case 1: no files in audit table
-        self.assertIsNone(get_next_queued_file_details(queue_to_check))
-
-        # Test case 2: files in audit table, but none of the files are in the RAVS_RSV queue
-        add_entry_to_table(MockFileDetails.flu_emis, file_status=FileStatus.QUEUED)  # different queue
-        add_entry_to_table(MockFileDetails.rsv_emis, file_status=FileStatus.QUEUED)  # different queue
-        add_entry_to_table(MockFileDetails.ravs_flu, file_status=FileStatus.QUEUED)  # different queue
-        add_entry_to_table(MockFileDetails.ravs_rsv_1, FileStatus.PROCESSED)  # same queue but already processed
-        self.assertIsNone(get_next_queued_file_details(queue_to_check))
-
-        # Test case 3: one queued file in the ravs_rsv queue
-        add_entry_to_table(MockFileDetails.ravs_rsv_2, file_status=FileStatus.QUEUED)
-        expected_table_entry = {**MockFileDetails.ravs_rsv_2.audit_table_entry, "status": {"S": FileStatus.QUEUED}}
-        self.assertEqual(get_next_queued_file_details(queue_to_check), deserialize_dynamodb_types(expected_table_entry))
-
-        # # Test case 4: multiple queued files in the RAVS_RSV queue
-        # Note that ravs_rsv files 3 and 4 have later timestamps than file 2, so file 2 remains the first in the queue
-        add_entry_to_table(MockFileDetails.ravs_rsv_3, file_status=FileStatus.QUEUED)
-        add_entry_to_table(MockFileDetails.ravs_rsv_4, file_status=FileStatus.QUEUED)
-        self.assertEqual(get_next_queued_file_details(queue_to_check), deserialize_dynamodb_types(expected_table_entry))
-
-    def test_change_audit_table_status_to_processed(self):
-        """Checks audit table correctly updates a record as processed"""
+    def test_update_audit_table_status(self):
+        """Checks audit table correctly updates a record to the requested status"""
         # Test case 1: file should be updated with status of 'Processed'.
 
-        add_entry_to_table(MockFileDetails.rsv_ravs, file_status=FileStatus.QUEUED)
+        add_entry_to_table(MockFileDetails.rsv_ravs, file_status=FileStatus.PROCESSING)
         add_entry_to_table(MockFileDetails.flu_emis, file_status=FileStatus.QUEUED)
-        table_items = dynamodb_client.scan(TableName=AUDIT_TABLE_NAME).get("Items", [])
 
-        expected_table_entry = {**MockFileDetails.rsv_ravs.audit_table_entry, "status": {"S": FileStatus.PROCESSED}}
+        expected_table_entry = {**MockFileDetails.rsv_ravs.audit_table_entry, "status": {"S": FileStatus.PREPROCESSED}}
         ravs_rsv_test_file = FileDetails("RSV", "RAVS", "X26")
         file_key = ravs_rsv_test_file.file_key
-        message_id = ravs_rsv_test_file.message_id_order
+        message_id = ravs_rsv_test_file.message_id
 
-        change_audit_table_status_to_processed(file_key, message_id)
+        update_audit_table_status(file_key, message_id, FileStatus.PREPROCESSED)
         table_items = dynamodb_client.scan(TableName=AUDIT_TABLE_NAME).get("Items", [])
 
         self.assertIn(expected_table_entry, table_items)
 
-        # Test case 2: # Audit table status should not be updated. Error should be raised.
+    def test_update_audit_table_status_throws_exception_with_invalid_id(self):
         emis_flu_test_file_2 = FileDetails("FLU", "EMIS", "YGM41")
 
         message_id = emis_flu_test_file_2.message_id
         file_key = (emis_flu_test_file_2.file_key,)
+
         with self.assertRaises(UnhandledAuditTableError):
-            change_audit_table_status_to_processed(file_key, message_id)
+            update_audit_table_status(file_key, message_id, FileStatus.PROCESSED)
 
-        # Test case 3: # Audit table status should updated to processed for all values.
-        message_id = emis_flu_test_file_2.message_id_order
-        file_key = emis_flu_test_file_2.file_key
-        change_audit_table_status_to_processed(file_key, message_id)
-        table_items = dynamodb_client.scan(TableName=AUDIT_TABLE_NAME).get("Items", [])
+
+if __name__ == "__main__":
+    unittest.main()

recordprocessor/tests/test_logging_decorator.py

@@ -61,8 +61,7 @@ def run(self, result=None):
         # Set up common patches to be applied to all tests in the class.
         # These patches can be overridden in individual tests.
         common_patches = [
-            patch("file_level_validation.change_audit_table_status_to_processed"),
-            patch("file_level_validation.get_next_queued_file_details", return_value=None),
+            patch("file_level_validation.update_audit_table_status"),
         ]
 
         with ExitStack() as stack: