Tests Pass

Author: nhsdevws

recordprocessor/src/batch_processor.py

@@ -50,10 +50,10 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     logger.info(f"process with encoder {encoder} from row {row_count+1}")
     row_count, err = process_rows(file_id, vaccine, supplier, file_key, allowed_operations,
                                   created_at_formatted_string, csv_reader, target_disease)
+
     if err:
-        logger.warning(f"Error processing: {err}.")
-        # check if it's a decode error
-        if err.reason == "invalid continuation byte":
+        logger.warning(f"Processing Error: {err}.")
+        if isinstance(err, InvalidEncoding):
             new_encoder = "cp1252"
             logger.info(f"Encode error at row {row_count} with {encoder}. Switch to {new_encoder}")
             encoder = new_encoder
@@ -108,7 +108,7 @@ def process_rows(file_id, vaccine, supplier, file_key, allowed_operations, creat
         # if error reason is 'invalid continuation byte', then it's a decode error
         logger.error("Error processing row %s: %s", row_count, error)
         if hasattr(error, 'reason') and error.reason == "invalid continuation byte":
-            return total_rows_processed_count, error
+            return total_rows_processed_count, InvalidEncoding("Invalid continuation byte")
         else:
             raise error
     return total_rows_processed_count, None

recordprocessor/tests/test_batch_processor.py

@@ -0,0 +1,174 @@
+import unittest
+import os
+from io import BytesIO
+from unittest.mock import patch
+from batch_processor import process_csv_to_fhir
+from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import create_patch
+
+
+class TestProcessorEdgeCases(unittest.TestCase):
+
+    def setUp(self):
+        self.mock_logger_info = create_patch("logging.Logger.info")
+        self.mock_logger_warning = create_patch("logging.Logger.warning")
+        self.mock_logger_error = create_patch("logging.Logger.error")
+        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_s3_put_object = create_patch("utils_for_recordprocessor.s3_client.put_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")
+        self.mock_firehose_client = create_patch("logging_decorator.firehose_client")
+
+    def tearDown(self):
+        patch.stopall()
+
+    def expand_test_data(self, data: list[bytes], num_rows: int) -> list[bytes]:
+        n_rows = len(data) - 1  # Exclude header
+
+        if n_rows < num_rows:
+            multiplier = (num_rows // n_rows) + 1
+            header = data[0:1]
+            body = data[1:] * multiplier
+            data = header + body
+            data = data[:num_rows + 1]
+        return data
+
+    def create_test_data_from_file(self, file_name: str) -> list[bytes]:
+        test_csv_path = os.path.join(
+            os.path.dirname(__file__), "test_data", file_name
+        )
+        with open(test_csv_path, "rb") as f:
+            data = f.readlines()
+        return data
+
+    def insert_cp1252_at_end(self, data: list[bytes], new_text: bytes, field: int) -> list[bytes]:
+        for i in reversed(range(len(data))):
+            line = data[i]
+            # Split fields by pipe
+            fields = line.strip().split(b"|")
+            fields[field] = new_text
+            # Reconstruct the line
+            data[i] = b"|".join(fields) + b"\n"
+            break
+        return data
+
+    def test_process_large_file_with_cp1252(self):
+        """ Test processing a large file with cp1252 encoding """
+        n_rows = 500
+        data = self.create_test_data_from_file("test-batch-data.csv")
+        data = self.expand_test_data(data, n_rows)
+        data = self.insert_cp1252_at_end(data, b'D\xe9cembre', 2)
+        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 = "some disease"
+
+        message_body = {
+                    "vaccine_type": "vax-type-1",
+                    "supplier": "test-supplier",
+                }
+        self.mock_map_target_disease.return_value = "some disease"
+
+        n_rows_processed = process_csv_to_fhir(message_body)
+        self.assertEqual(n_rows_processed, n_rows)
+        self.assertEqual(self.mock_send_to_kinesis.call_count, n_rows)
+        # check logger.warning called for decode error
+        self.mock_logger_warning.assert_called()
+        warning_call_args = self.mock_logger_warning.call_args[0][0]
+        self.assertEqual(warning_call_args, "Processing Error: Invalid continuation byte.")
+
+    def test_process_large_file_with_utf8(self):
+        """ Test processing a large file with utf-8 encoding """
+        n_rows = 500
+        data = self.create_test_data_from_file("test-batch-data.csv")
+        data = self.expand_test_data(data, n_rows)
+        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 = "some disease"
+
+        message_body = {
+                    "vaccine_type": "vax-type-1",
+                    "supplier": "test-supplier",
+                }
+        self.mock_map_target_disease.return_value = "some disease"
+
+        n_rows_processed = process_csv_to_fhir(message_body)
+        self.assertEqual(n_rows_processed, n_rows)
+        self.assertEqual(self.mock_send_to_kinesis.call_count, n_rows)
+        self.mock_logger_warning.assert_not_called()
+        self.mock_logger_error.assert_not_called()
+
+    def test_process_cp1252_small_file(self):
+        """ Test processing a small file with cp1252 encoding """
+        data = self.create_test_data_from_file("test-batch-data-cp1252.csv")
+        data = [line if line.endswith(b"\n") else line + b"\n" for line in data]
+        n_rows = len(data) - 1  # Exclude header
+
+        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 = "some disease"
+
+        message_body = {
+                    "vaccine_type": "vax-type-1",
+                    "supplier": "test-supplier",
+                }
+
+        self.mock_map_target_disease.return_value = "some disease"
+
+        n_rows_processed = process_csv_to_fhir(message_body)
+        self.assertEqual(n_rows_processed, n_rows)
+        self.assertEqual(self.mock_send_to_kinesis.call_count, n_rows)
+        self.mock_logger_warning.assert_called()
+        warning_call_args = self.mock_logger_warning.call_args[0][0]
+        self.assertTrue(warning_call_args.startswith("Invalid Encoding detected in process_csv_to_fhir"))
+
+    def test_process_utf8_small_file(self):
+        """ Test processing a small file with utf-8 encoding """
+        data = self.create_test_data_from_file("test-batch-data.csv")
+        data = [line if line.endswith(b"\n") else line + b"\n" for line in data]
+        n_rows = len(data) - 1  # Exclude header
+
+        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 = "some disease"
+
+        message_body = {
+                    "vaccine_type": "vax-type-1",
+                    "supplier": "test-supplier",
+                }
+        self.mock_map_target_disease.return_value = "some disease"
+
+        n_rows_processed = process_csv_to_fhir(message_body)
+        self.assertEqual(n_rows_processed, n_rows)
+        self.assertEqual(self.mock_send_to_kinesis.call_count, n_rows)
+        self.mock_logger_warning.assert_not_called()
+        self.mock_logger_error.assert_not_called()
+
+    def test_fix_cp1252(self):
+        # create a cp1252 string that contains an accented E
+        # this is not a unit test as such but checks/confirms encoding assumptions
+        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(input_dict: dict, encoding: str) -> dict:
+    """
+    Decode all byte strings in a dictionary to UTF-8 strings using the specified encoding.
+    """
+    decoded_dict = {}
+    for key, value in input_dict.items():
+        if isinstance(value, bytes):
+            decoded_dict[key] = value.decode(encoding)
+        else:
+            decoded_dict[key] = value
+    return decoded_dict

recordprocessor/tests/test_recordprocessor_edge_cases.py

@@ -23,6 +23,7 @@
     REGION_NAME,
 )
 from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames, Kinesis
+from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import create_patch
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
     from constants import Diagnostics
@@ -52,6 +53,7 @@ def setUp(self) -> None:
             "code": "55735004",
             "term": "Respiratory syncytial virus infection (disorder)"
         }])
+        self.mock_logger_info = create_patch("logging.Logger.info")
 
     def tearDown(self) -> None:
         GenericTearDown(s3_client, firehose_client, kinesis_client)