VED-58 Batch processor error messages improved

Included supplier and vaccine type in error messages.

Author: robertnovac1

filenameprocessor/Makefile

@@ -1,20 +1,20 @@
 build:
 	docker build -t imms-lambda-build -f Dockerfile .
 
-package:build
+package: build
 	mkdir -p build
 	docker run --rm -v $(shell pwd)/build:/build imms-lambda-build
 
 test:
 	python -m unittest
 
-coverage run:
-    coverage run -m unittest discover
+coverage-run:
+	coverage run -m unittest discover
 
-coverage report:
+coverage-report:
 	coverage report -m 
 
-coverage html:
-    coverage html  
-	
-.PHONY: build package test
+coverage-html:
+	coverage html  
+
+.PHONY: build package test coverage-run coverage-report coverage-html

filenameprocessor/src/file_name_processor.py

@@ -6,6 +6,7 @@
 (ODS code has multiple lengths)
 """
 
+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
@@ -44,6 +45,9 @@ def handle_record(record) -> dict:
         logger.error("Error obtaining file_key: %s", error)
         return {"statusCode": 500, "message": "Failed to download file key", "error": str(error)}
 
+    vaccine_type = "unknown"
+    supplier = "unknown"
+
     if "data-sources" in bucket_name:
 
         # The lambda is unintentionally invoked when a file is moved into a different folder in the source bucket.
@@ -56,8 +60,6 @@ def handle_record(record) -> dict:
         # Set default values for file-specific variables
         message_id = "Message id was not created"
         created_at_formatted_string = "created_at_time not identified"
-        vaccine_type = "unknown"
-        supplier = "unknown"
 
         try:
             # If the record contains a message_id, then the lambda has been invoked by a file already in the queue
@@ -134,6 +136,8 @@ def handle_record(record) -> dict:
                 "file_key": file_key,
                 "message_id": message_id,
                 "error": str(error),
+                "vaccine_type": vaccine_type,
+                "supplier": supplier
             }
 
     elif "config" in bucket_name:
@@ -148,9 +152,21 @@ def handle_record(record) -> dict:
             return {"statusCode": 500, "message": message, "file_key": file_key, "error": str(error)}
 
     else:
-        logger.error("Unable to process file %s due to unexpected bucket name %s", file_key, bucket_name)
-        message = f"Failed to process file due to unexpected bucket name {bucket_name}"
-        return {"statusCode": 500, "message": message, "file_key": file_key}
+        try:
+            vaccine_type, supplier = validate_file_key(file_key)
+            logger.error("Unable to process file %s due to unexpected bucket name %s", file_key, bucket_name)
+            message = f"Failed to process file due to unexpected bucket name {bucket_name}"
+
+            return {"statusCode": 500, "message": message, "file_key": file_key,
+                    "vaccine_type": vaccine_type, "supplier": supplier}
+
+        except Exception as error:
+            logger.error("Unable to process file due to unexpected bucket name %s and file key %s",
+                         bucket_name, file_key)
+            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)}
 
 
 def lambda_handler(event: dict, context) -> None:  # pylint: disable=unused-argument
@@ -161,3 +177,27 @@ def lambda_handler(event: dict, context) -> None:  # pylint: disable=unused-argu
         handle_record(record)
 
     logger.info("Filename processor lambda task completed")
+
+
+def run_local():
+    parser = argparse.ArgumentParser("file_name_processor")
+    parser.add_argument("--bucket", required=True, help="Bucket name.", type=str)
+    parser.add_argument("--key", required=True, help="Object key.", type=str)
+    args = parser.parse_args()
+
+    event = {
+        "Records": [
+            {
+                "s3": {
+                    "bucket": {"name": args.bucket},
+                    "object": {"key": args.key}
+                }
+            }
+        ]
+    }
+    print(event)
+    print(lambda_handler(event=event, context={}))
+
+
+if __name__ == "__main__":
+    run_local()

filenameprocessor/tests/test_lambda_handler.py

@@ -1,5 +1,5 @@
 """Tests for lambda_handler"""
-
+import sys
 from unittest.mock import patch
 from unittest import TestCase
 from json import loads as json_loads
@@ -154,6 +154,14 @@ def get_audit_table_items():
         """Return all items in the audit table"""
         return dynamodb_client.scan(TableName=AUDIT_TABLE_NAME).get("Items", [])
 
+    def test_lambda_handler_no_file_key_throws_exception(self):
+        """Tests if exception is thrown when file_key is not provided"""
+
+        broken_record = {"Records": [{"s3": {"bucket": {"name": "test"}}}]}
+        with patch("file_name_processor.logger") as mock_logger:
+            lambda_handler(broken_record, None)
+            mock_logger.error.assert_called_once()
+
     def test_lambda_handler_new_file_success_and_first_in_queue(self):
         """
         Tests that for a new file, which passes validation and is the only file processing for the supplier_vaccineType
@@ -458,6 +466,29 @@ def setUp(self):
     def tearDown(self):
         GenericTearDown(s3_client, firehose_client, sqs_client, dynamodb_client)
 
+    def test_elasticcache_failure_handled(self):
+        "Tests if elastic cache failure is handled when service fails to send message"
+        event = {
+            "s3": {
+                "bucket": {"name": "my-config-bucket"},  # triggers 'config' branch
+                "object": {"key": "testfile.csv"}
+            }
+        }
+
+        with patch("file_name_processor.upload_to_elasticache", side_effect=Exception("Upload failed")), \
+             patch("file_name_processor.logger") as mock_logger:
+
+            result = handle_record(event)
+
+            self.assertEqual(result["statusCode"], 500)
+            self.assertEqual(result["message"], "Failed to upload file content to cache")
+            self.assertEqual(result["file_key"], "testfile.csv")
+            self.assertIn("error", result)
+
+            mock_logger.error.assert_called_once()
+            logged_msg = mock_logger.error.call_args[0][0]
+            self.assertIn("Error uploading to cache", logged_msg)
+
     def test_successful_processing_from_configs(self):
         """Tests that the permissions config file content is uploaded to elasticache successfully"""
         fake_redis = fakeredis.FakeStrictRedis()
@@ -496,7 +527,7 @@ def test_successful_processing_from_configs(self):
             "file_key": ravs_rsv_file_details_1.file_key,
             "message_id": ravs_rsv_file_details_1.message_id,
             "vaccine_type": ravs_rsv_file_details_1.vaccine_type,
-            "supplier": ravs_rsv_file_details_1.supplier,
+            "supplier": ravs_rsv_file_details_1.supplier
         }
         self.assertEqual(result, expected_result)
 
@@ -524,5 +555,104 @@ def test_successful_processing_from_configs(self):
             "file_key": ravs_rsv_file_details_2.file_key,
             "message_id": ravs_rsv_file_details_2.message_id,
             "error": "Initial file validation failed: RAVS does not have permissions for RSV",
+            "vaccine_type": ravs_rsv_file_details_2.vaccine_type,
+            "supplier": ravs_rsv_file_details_2.supplier
         }
         self.assertEqual(result, expected_result)
+
+
+@patch.dict("os.environ", MOCK_ENVIRONMENT_DICT)
+@mock_s3
+@mock_dynamodb
+@mock_sqs
+@mock_firehose
+class TestUnexpectedBucket(TestCase):
+    """Tests for lambda_handler when an unexpected bucket name is used"""
+
+    def setUp(self):
+        GenericSetUp(s3_client, firehose_client, sqs_client, dynamodb_client)
+
+    def tearDown(self):
+        GenericTearDown(s3_client, firehose_client, sqs_client, dynamodb_client)
+
+    def test_unexpected_bucket_name(self):
+        """Tests if unkown bucket name is handled in lambda_handler"""
+        ravs_record = MockFileDetails.ravs_rsv_1
+        record = {
+            "s3": {
+                "bucket": {"name": "unknown-bucket"},
+                "object": {"key": ravs_record.file_key}
+            }
+        }
+
+        with patch("file_name_processor.logger") as mock_logger:
+            result = handle_record(record)
+
+            self.assertEqual(result["statusCode"], 500)
+            self.assertIn("unexpected bucket name", result["message"])
+            self.assertEqual(result["file_key"], ravs_record.file_key)
+            self.assertEqual(result["vaccine_type"], ravs_record.vaccine_type)
+            self.assertEqual(result["supplier"], ravs_record.supplier)
+
+            mock_logger.error.assert_called_once()
+            args = mock_logger.error.call_args[0]
+            self.assertIn("Unable to process file", args[0])
+            self.assertIn(ravs_record.file_key, args)
+            self.assertIn("unknown-bucket", args)
+
+    def test_unexpected_bucket_name_and_filename_validation_fails(self):
+        """Tests if filename validation error is handled when bucket name is incorrect"""
+        invalid_file_key = "InvalidVaccineType_Vaccinations_v5_YGM41_20240708T12130100.csv"
+        record = {
+            "s3": {
+                "bucket": {"name": "unknown-bucket"},
+                "object": {"key": invalid_file_key}
+            }
+        }
+
+        with patch("file_name_processor.logger") as mock_logger:
+            result = handle_record(record)
+
+            self.assertEqual(result["statusCode"], 500)
+            self.assertIn("unexpected bucket name", result["message"])
+            self.assertEqual(result["file_key"], invalid_file_key)
+            self.assertEqual(result["vaccine_type"], "unknown")
+            self.assertEqual(result["supplier"], "unknown")
+
+            mock_logger.error.assert_called_once()
+            args = mock_logger.error.call_args[0]
+            self.assertIn("Unable to process file", args[0])
+            self.assertIn(invalid_file_key, args)
+            self.assertIn("unknown-bucket", args)
+
+
+class TestMainEntryPoint(TestCase):
+
+    def test_run_local_constructs_event_and_calls_lambda_handler(self):
+        test_args = [
+            "file_name_processor.py",
+            "--bucket", "test-bucket",
+            "--key", "some/path/file.csv"
+        ]
+
+        expected_event = {
+            "Records": [
+                {
+                    "s3": {
+                        "bucket": {"name": "test-bucket"},
+                        "object": {"key": "some/path/file.csv"}
+                    }
+                }
+            ]
+        }
+
+        with (
+            patch.object(sys, "argv", test_args),
+            patch("file_name_processor.lambda_handler") as mock_lambda_handler,
+            patch("file_name_processor.print") as mock_print
+        ):
+            import file_name_processor
+            file_name_processor.run_local()
+
+            mock_lambda_handler.assert_called_once_with(event=expected_event, context={})
+            mock_print.assert_called()

filenameprocessor/tests/test_logging_decorator.py

@@ -6,6 +6,7 @@
 from copy import deepcopy
 from contextlib import ExitStack
 from boto3 import client as boto3_client
+from botocore.exceptions import ClientError
 from moto import mock_s3, mock_firehose, mock_sqs, mock_dynamodb
 
 from tests.utils_for_tests.generic_setup_and_teardown import GenericSetUp, GenericTearDown
@@ -192,13 +193,43 @@ def test_logging_failed_validation(self):
             "file_key": FILE_DETAILS.file_key,
             "message_id": FILE_DETAILS.message_id,
             "error": "Initial file validation failed: EMIS does not have permissions for FLU",
+            "vaccine_type": "FLU",
+            "supplier": "EMIS"
         }
 
         log_data = json.loads(mock_logger.info.call_args[0][0])
         self.assertEqual(log_data, expected_log_data)
 
         mock_send_log_to_firehose.assert_called_once_with(log_data)
 
+    def test_logging_throws_exception(self):
+        """Tests that exception is caught when failing to send message to Firehose"""
+        permissions_config_content = generate_permissions_config_content({"EMIS": ["COVID19_FULL"]})
+
+        firehose_exception = ClientError(
+            error_response={"Error": {"Code": "ServiceUnavailable", "Message": "Service down"}},
+            operation_name="PutRecord"
+        )
+
+        with (
+            patch("file_name_processor.uuid4", return_value=FILE_DETAILS.message_id),
+            patch("elasticache.redis_client.get", return_value=permissions_config_content),
+            patch("logging_decorator.firehose_client.put_record", side_effect=firehose_exception),
+            patch("logging_decorator.logger") as mock_logger,
+        ):
+            lambda_handler(MOCK_VACCINATION_EVENT, context=None)
+
+        # Assert logger.exception was called once
+        mock_logger.exception.assert_called_once()
+
+        # Extract the call arguments
+        exception_message = mock_logger.exception.call_args[0][0]
+        exception_obj = mock_logger.exception.call_args[0][1]
+
+        # Check that the message format is correct
+        self.assertIn("Error sending log to Firehose", exception_message)
+        self.assertEqual(exception_obj, firehose_exception)
+
     def test_logging_successful_config_upload(self):
         """
         Tests that the correct logs are sent to cloudwatch and splunk when the config cache is successfully updated