Needs firehost count test

Author: nhsdevws

recordprocessor/src/batch_processor.py

@@ -7,8 +7,8 @@
 from mappings import map_target_disease
 from send_to_kinesis import send_to_kinesis
 from clients import logger
-from file_level_validation import file_level_validation
-from errors import NoOperationPermissions, InvalidHeaders
+from file_level_validation import file_level_validation, validate_content_headers
+from errors import NoOperationPermissions, InvalidHeaders, InvalidEncoding
 from utils_for_recordprocessor import get_csv_content_dict_reader
 
 
@@ -17,11 +17,21 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     For each row of the csv, attempts to transform into FHIR format, sends a message to kinesis,
     and documents the outcome for each row in the ack file.
     """
+    encoder = "utf-8"  # default encoding
     try:
-        interim_message_body = file_level_validation(incoming_message_body=incoming_message_body)
+        interim_message_body = file_level_validation(incoming_message_body=incoming_message_body, encoder=encoder)
+    except InvalidEncoding as error:
+        logger.warning("Invalid Encoding detected in process_csv_to_fhir: %s", error)
+        # retry with cp1252 encoding
+        encoder = "cp1252"
+        try:
+            interim_message_body = file_level_validation(incoming_message_body=incoming_message_body, encoder=encoder)
+        except Exception as error:
+            logger.error(f"Error in file_level_validation with {encoder} encoding: %s", error)
+            return 0
     except (InvalidHeaders, NoOperationPermissions, Exception):  # pylint: disable=broad-exception-caught
         # If the file is invalid, processing should cease immediately
-        return
+        return 0
 
     file_id = interim_message_body.get("message_id")
     vaccine = interim_message_body.get("vaccine")
@@ -32,29 +42,28 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     csv_reader = interim_message_body.get("csv_dict_reader")
 
     target_disease = map_target_disease(vaccine)
-    print("process csv to fhir")
     row_count = 0
-    encoder = "utf-8"  # default encoding
-    try:
-        row_count = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
-                                 created_at_formatted_string, csv_reader, target_disease)
-    except Exception as error:  # pylint: disable=broad-exception-caught
-        new_encoder = "cp1252"
-        print(f"Error processing: {error}.")
+    row_count, err = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
+                                  created_at_formatted_string, csv_reader, target_disease)
+    if err:
+        print(f"Error processing: {err}.")
         # check if it's a decode error, ie error.args[0] begins with "'utf-8' codec can't decode byte"
-        if error.reason == "invalid continuation byte":
+        if err.reason == "invalid continuation byte":
+            new_encoder = "cp1252"
             print(f"Encode error at row {row_count} with {encoder}. Switch to {new_encoder}")
             # print(f"Detected decode error: {error.reason}")
             encoder = new_encoder
             # if we are here, re-read the file with alternative encoding and skip processed rows
-            row_count = process_rows_retry(file_id, vaccine, supplier, file_key,
-                                           allowed_operations, created_at_formatted_string,
-                                           encoder, row_count)
+            csv_reader = get_csv_content_dict_reader(file_key, encoder=encoder)
+            validate_content_headers(csv_reader)
+            row_count = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
+                                     created_at_formatted_string, csv_reader, target_disease, row_count)
         else:
-            logger.error(f"Non-decode error: {error}. Cannot retry. Call someone.")
-            raise error from error
+            logger.error(f"Non-decode error: {err}. Cannot retry. Call someone.")
+            raise err
 
     logger.info("Total rows processed: %s", row_count)
+    return row_count
 
 
 def process_rows_retry(file_id, vaccine, supplier, file_key, allowed_operations,
@@ -81,35 +90,39 @@ def process_rows(file_id, vaccine, supplier, file_key, allowed_operations, creat
     print("process_rows...")
     row_count = 0
     start_row = total_rows_processed_count
-    for row in csv_reader:
-
-        row_count += 1
-        if row_count > start_row:
-            row_id = f"{file_id}^{row_count}"
-            logger.info("MESSAGE ID : %s", row_id)
-
-            # convert dict to string and print first 20 chars
-            if (total_rows_processed_count % 1000 == 0):
-                print(f"Process: {total_rows_processed_count}")
-            if (total_rows_processed_count > 19995):
-                print(f"Process: {total_rows_processed_count} - {row['PERSON_SURNAME']}")
-
-            # Process the row to obtain the details needed for the message_body and ack file
-            details_from_processing = process_row(target_disease, allowed_operations, row)
-
-            # Create the message body for sending
-            outgoing_message_body = {
-                "row_id": row_id,
-                "file_key": file_key,
-                "supplier": supplier,
-                "vax_type": vaccine,
-                "created_at_formatted_string": created_at_formatted_string,
-                **details_from_processing,
-            }
-
-            send_to_kinesis(supplier, outgoing_message_body, vaccine)
-            total_rows_processed_count += 1
-            logger.info("Total rows processed: %s", total_rows_processed_count)
+    try:
+        for row in csv_reader:
+
+            row_count += 1
+            if row_count > start_row:
+                row_id = f"{file_id}^{row_count}"
+                logger.info("MESSAGE ID : %s", row_id)
+
+                # convert dict to string and print first 20 chars
+                if (total_rows_processed_count % 1000 == 0):
+                    print(f"Process: {total_rows_processed_count}")
+                if (total_rows_processed_count > 19995):
+                    print(f"Process: {total_rows_processed_count} - {row['PERSON_SURNAME']}")
+
+                # Process the row to obtain the details needed for the message_body and ack file
+                details_from_processing = process_row(target_disease, allowed_operations, row)
+
+                # Create the message body for sending
+                outgoing_message_body = {
+                    "row_id": row_id,
+                    "file_key": file_key,
+                    "supplier": supplier,
+                    "vax_type": vaccine,
+                    "created_at_formatted_string": created_at_formatted_string,
+                    **details_from_processing,
+                }
+
+                send_to_kinesis(supplier, outgoing_message_body, vaccine)
+                total_rows_processed_count += 1
+                logger.info("Total rows processed: %s", total_rows_processed_count)
+    except Exception as error:  # pylint: disable=broad-exception-caught
+        logger.error("Error processing row %s: %s", row_count, error)
+        return total_rows_processed_count, error
     return total_rows_processed_count
 
 

recordprocessor/src/errors.py

@@ -9,5 +9,9 @@ class InvalidHeaders(Exception):
     """A custom exception for when the file headers are invalid."""
 
 
+class InvalidEncoding(Exception):
+    """A custom exception for when the file encoding is invalid."""
+
+
 class UnhandledAuditTableError(Exception):
     """A custom exception for when an unexpected error occurs whilst adding the file to the audit table."""

recordprocessor/src/file_level_validation.py

@@ -5,7 +5,7 @@
 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 errors import InvalidHeaders, NoOperationPermissions
+from errors import InvalidHeaders, NoOperationPermissions, InvalidEncoding
 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
@@ -61,7 +61,7 @@ def move_file(bucket_name: str, source_file_key: str, destination_file_key: str)
 
 
 @file_level_validation_logging_decorator
-def file_level_validation(incoming_message_body: dict) -> dict:
+def file_level_validation(incoming_message_body: dict, encoder: str) -> 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.
@@ -78,7 +78,7 @@ def file_level_validation(incoming_message_body: dict) -> dict:
         created_at_formatted_string = incoming_message_body.get("created_at_formatted_string")
 
         # Fetch the data
-        csv_reader = get_csv_content_dict_reader(file_key)
+        csv_reader = get_csv_content_dict_reader(file_key, encoder=encoder)
 
         validate_content_headers(csv_reader)
 
@@ -98,8 +98,10 @@ def file_level_validation(incoming_message_body: dict) -> dict:
             "created_at_formatted_string": created_at_formatted_string,
             "csv_dict_reader": csv_reader,
         }
-
     except (InvalidHeaders, NoOperationPermissions, Exception) as error:
+        if error.reason == "invalid continuation byte" and encoder == "utf-8":
+            # propagate the error to trigger a retry with cp1252 encoding
+            raise InvalidEncoding(f"Error File encoding {encoder} is invalid.")
         logger.error("Error in file_level_validation: %s", error)
 
         # NOTE: The Exception may occur before the file_id, file_key and created_at_formatted_string are assigned

recordprocessor/tests/test_batch_processor.py

@@ -2,36 +2,26 @@
 import os
 from io import BytesIO
 from unittest.mock import patch
-# from utils_for_recordprocessor import dict_decode
 
 with patch("logging_decorator.file_level_validation_logging_decorator", lambda f: f):
-    # from file_level_validation import file_level_validation
     from batch_processor import process_csv_to_fhir
 
 
+def create_patch(target: str):
+    patcher = patch(target)
+    return patcher.start()
+
+
 class TestProcessCsvToFhir(unittest.TestCase):
 
     def setUp(self):
-        self.logger_info_patcher = patch("logging.Logger.info")
-        self.mock_logger_info = self.logger_info_patcher.start()
-        # self.update_audit_table_status_patcher = patch("batch_processor.update_audit_table_status")
-        # self.mock_update_audit_table_status = self.update_audit_table_status_patcher.start()
-        self.send_to_kinesis_patcher = patch("batch_processor.send_to_kinesis")
-        self.mock_send_to_kinesis = self.send_to_kinesis_patcher.start()
-        self.map_target_disease_patcher = patch("batch_processor.map_target_disease")
-        self.mock_map_target_disease = self.map_target_disease_patcher.start()
-        self.s3_get_object_patcher = patch("utils_for_recordprocessor.s3_client.get_object")
-        self.mock_s3_get_object = self.s3_get_object_patcher.start()
-        self.make_and_move_patcher = patch("file_level_validation.make_and_upload_ack_file")
-        self.mock_make_and_move = self.make_and_move_patcher.start()
-        self.make_and_move_patcher = patch("file_level_validation.move_file")
-        self.mock_move_file = self.make_and_move_patcher.start()
-        # get_permitted_operations
-        self.get_permitted_operations_patcher = patch("file_level_validation.get_permitted_operations")
-        self.mock_get_permitted_operations = self.get_permitted_operations_patcher.start()
-
-        # self.validate_content_headers_patcher = patch("file_level_validation.validate_content_headers")
-        # self.mock_validate_content_headers = self.validate_content_headers_patcher.start()
+        self.mock_logger_info = create_patch("logging.Logger.info")
+        self.mock_send_to_kinesis = create_patch("batch_processor.send_to_kinesis")
+        self.mock_map_target_disease = create_patch("batch_processor.map_target_disease")
+        self.mock_s3_get_object = create_patch("utils_for_recordprocessor.s3_client.get_object")
+        self.mock_make_and_move = create_patch("file_level_validation.make_and_upload_ack_file")
+        self.mock_move_file = create_patch("file_level_validation.move_file")
+        self.mock_get_permitted_operations = create_patch("file_level_validation.get_permitted_operations")
 
     def tearDown(self):
         patch.stopall()
@@ -64,24 +54,48 @@ def insert_cp1252_at_end(self, data: list[bytes], new_text: bytes, field: int) -
             print(f"replace field: {fields[field]}")
             fields[field] = new_text
             print(f"replaced field: {fields[field]}")
-
             # Reconstruct the line
             data[i] = b"|".join(fields) + b"\n"
             break
         return data
 
-    def test_process_csv_to_fhir_success(self):
-        # Setup mocks
-        print("test_process_csv_to_fhir_success")
+    def test_process_large_file_with_cp1252(self):
+        """ Test processing a large file with cp1252 encoding """
         try:
+            n_rows = 20000
             data = self.create_test_data_from_file("test-batch-data.csv")
-            data = self.expand_test_data(data, 20_000)
+            data = self.expand_test_data(data, n_rows)
             data = self.insert_cp1252_at_end(data, b'D\xe9cembre', 2)
+            self.mock_s3_get_object.side_effect = [{"Body": BytesIO(b"".join(data))},
+                                                   {"Body": BytesIO(b"".join(data))}]
+            self.mock_map_target_disease.return_value = "RSV"
+
+            message_body = {
+                        "message_id": "file123",
+                        "vaccine_type": "covid",
+                        "supplier": "test-supplier",
+                        "filename": "file-key-1",
+                        "permission": ["COVID.R", "COVID.U", "COVID.D"],
+                        "allowed_operations": ["CREATE", "UPDATE", "DELETE"],
+                        "created_at_formatted_string": "2024-09-05T12:00:00Z"
+                    }
+            self.mock_get_permitted_operations.return_value = {"CREATE", "UPDATE", "DELETE"}
 
-            # Read CSV from test_csv_path as utf-8
-            ret1 = {"Body": BytesIO(b"".join(data))}
-            ret2 = {"Body": BytesIO(b"".join(data))}
-            self.mock_s3_get_object.side_effect = [ret1, ret2]
+            self.mock_map_target_disease.return_value = "RSV"
+
+            n_rows_processed = process_csv_to_fhir(message_body)
+            self.assertEqual(n_rows_processed, n_rows)
+        except Exception as e:
+            print(f"Exception during test: {e}")
+
+    def test_process_small_file_with_cp1252(self):
+        """ Test processing a small file with cp1252 encoding """
+        try:
+            data = self.create_test_data_from_file("test-batch-data-cp1252.csv")
+            n_rows = len(data) - 1  # Exclude header
+            # data = self.insert_cp1252_at_end(data, b'D\xe9cembre', 2)
+            self.mock_s3_get_object.side_effect = [{"Body": BytesIO(b"".join(data))},
+                                                   {"Body": BytesIO(b"".join(data))}]
             self.mock_map_target_disease.return_value = "RSV"
 
             message_body = {
@@ -93,26 +107,18 @@ def test_process_csv_to_fhir_success(self):
                         "allowed_operations": ["CREATE", "UPDATE", "DELETE"],
                         "created_at_formatted_string": "2024-09-05T12:00:00Z"
                     }
-            # self.mock_file_level_validation.return_value = message_body
             self.mock_get_permitted_operations.return_value = {"CREATE", "UPDATE", "DELETE"}
 
             self.mock_map_target_disease.return_value = "RSV"
 
-            process_csv_to_fhir(message_body)
+            n_rows_processed = process_csv_to_fhir(message_body)
+            self.assertEqual(n_rows_processed, n_rows)
         except Exception as e:
             print(f"Exception during test: {e}")
 
     def test_fix_cp1252(self):
         # create a cp1252 string that contains an accented E
-        source_text = b'D\xe9cembre'
-        test_dict = {
-            "date": source_text,
-            "name": "Test Name"}
-        utf8_dict = dict_decode(test_dict, "cp1252")
-        self.assertEqual(utf8_dict["date"], "Décembre")
-        self.assertEqual(utf8_dict["name"], "Test Name")
-
-    def dict_decode(self):
+        # this is a unit test as such but checks our assumptions about encoding
         source_text = b'D\xe9cembre'
         test_dict = {
             "date": source_text,
@@ -133,3 +139,5 @@ def dict_decode(input_dict: dict, encoding: str) -> dict:
         else:
             decoded_dict[key] = value
     return decoded_dict
+
+# @TODO TEST to check correct number of messages sent to firehose with encode error