updated unit tests for recordprocessor

amitkr-22 · amitkr-22 · commit 54b66757a3f4 · 2025-03-04T14:50:01.000Z
diff --git a/recordprocessor/src/send_to_kinesis.py b/recordprocessor/src/send_to_kinesis.py
@@ -15,6 +15,7 @@ def send_to_kinesis(supplier: str, message_body: dict, vaccine_type: str) -> boo
             Data=json.dumps(message_body, ensure_ascii=False),
             PartitionKey=f"{supplier}_{vaccine_type}",
         )
+        return True
     except ClientError as error:
         logger.error("Error sending message to Kinesis: %s", error)
         raise
diff --git a/recordprocessor/tests/test_make_and_upload_ack_file.py b/recordprocessor/tests/test_make_and_upload_ack_file.py
@@ -2,18 +2,31 @@
 
 import unittest
 from unittest.mock import patch
-from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT
+from copy import deepcopy
+from boto3 import client as boto3_client
+from moto import mock_s3
+from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames
+from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import get_csv_file_dict_reader
+from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import MockFileDetails
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
-    from make_and_upload_ack_file import make_ack_data
+    from src.make_and_upload_ack_file import make_ack_data, upload_ack_file, make_and_upload_ack_file
+    from clients import REGION_NAME
 
+from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import GenericSetUp, GenericTearDown
 
+s3_client = boto3_client("s3", region_name=REGION_NAME)
+
+FILE_DETAILS = MockFileDetails.flu_emis
+
+@mock_s3
+@patch.dict("os.environ", MOCK_ENVIRONMENT_DICT)
 class TestMakeAndUploadAckFile(unittest.TestCase):
     "Tests for make_and_upload_ack_file functions"
 
     def setUp(self) -> None:
         self.message_id = "test_id"
-        self.created_at_formatted_string = "20200101T12345600"
+        self.created_at_formatted_string = "20211120T12000000"
         self.ack_data_validation_passed_and_message_delivered = {
             "MESSAGE_HEADER_ID": self.message_id,
             "HEADER_RESPONSE_CODE": "Success",
@@ -57,6 +70,11 @@ def setUp(self) -> None:
             "MESSAGE_DELIVERY": False,
         }
 
+        GenericSetUp(s3_client)
+
+    def tearDown(self):
+        GenericTearDown(s3_client)
+
     def test_make_ack_data(self):
         "Tests make_ack_data makes correct ack data based on the input args"
         # Test case tuples are stuctured as (validation_passed, message_delivered, expected_result)
@@ -73,8 +91,69 @@ def test_make_ack_data(self):
                     make_ack_data(
                         self.message_id, validation_passed, message_delivered, self.created_at_formatted_string
                     ),
-                    expected_result,
-                )
+                    expected_result,                   
+                )           
+
+    def test_upload_ack_file_success(self):
+        """Test that upload_ack_file successfully uploads the ack file"""        
+        
+        upload_ack_file(
+            file_key=FILE_DETAILS.file_key,
+            ack_data=deepcopy(self.ack_data_validation_passed_and_message_delivered),
+            created_at_formatted_string=FILE_DETAILS.created_at_formatted_string,
+        )
+        expected_result = [deepcopy(self.ack_data_validation_passed_and_message_delivered)]        
+        # Note that the data downloaded from the CSV will contain the bool as a string
+        expected_result[0]["MESSAGE_DELIVERY"] = "True"
+        csv_dict_reader = get_csv_file_dict_reader(s3_client, BucketNames.DESTINATION, FILE_DETAILS.inf_ack_file_key)
+        self.assertEqual(list(csv_dict_reader), expected_result)
+
+    def test_upload_ack_file_failure(self):
+        """Test that upload_ack_file failed to upload the ack file"""        
+        
+        upload_ack_file(
+            file_key=FILE_DETAILS.file_key,
+            ack_data=deepcopy(self.ack_data_validation_passed_and_message_not_delivered),
+            created_at_formatted_string=FILE_DETAILS.created_at_formatted_string,
+        )
+        expected_result = [deepcopy(self.ack_data_validation_passed_and_message_not_delivered)]        
+        # Note that the data downloaded from the CSV will contain the bool as a string
+        expected_result[0]["MESSAGE_DELIVERY"] = "False"
+        csv_dict_reader = get_csv_file_dict_reader(s3_client, BucketNames.DESTINATION, FILE_DETAILS.inf_ack_file_key)
+        self.assertEqual(list(csv_dict_reader), expected_result)
+
+    def test_make_and_upload_ack_file_success(self):
+        """Test that make_and_upload_ack_file uploads an ack file containing the correct values"""
+        make_and_upload_ack_file(
+            message_id=self.message_id,
+            file_key=FILE_DETAILS.file_key,
+            validation_passed=True,
+            message_delivered=True,
+            created_at_formatted_string=FILE_DETAILS.created_at_formatted_string,
+        )
+
+        expected_result = [deepcopy(self.ack_data_validation_passed_and_message_delivered)]
+        # Note that the data downloaded from the CSV will contain the bool as a string
+        expected_result[0]["MESSAGE_DELIVERY"] = "True"
+        csv_dict_reader = get_csv_file_dict_reader(s3_client, BucketNames.DESTINATION, FILE_DETAILS.inf_ack_file_key)
+        self.assertEqual(list(csv_dict_reader), expected_result)
+
+
+    def test_make_and_upload_ack_file_failure(self):
+        """Test that make_and_upload_ack_file failed to upload an ack file containing the correct values"""
+        make_and_upload_ack_file(
+            message_id=self.message_id,
+            file_key=FILE_DETAILS.file_key,
+            validation_passed=True,
+            message_delivered=False,
+            created_at_formatted_string=FILE_DETAILS.created_at_formatted_string,
+        )
+
+        expected_result = [deepcopy(self.ack_data_validation_passed_and_message_not_delivered)]
+        # Note that the data downloaded from the CSV will contain the bool as a string
+        expected_result[0]["MESSAGE_DELIVERY"] = "False"
+        csv_dict_reader = get_csv_file_dict_reader(s3_client, BucketNames.DESTINATION, FILE_DETAILS.inf_ack_file_key)
+        self.assertEqual(list(csv_dict_reader), expected_result)
 
 
 if __name__ == "__main__":
diff --git a/recordprocessor/tests/test_process_row.py b/recordprocessor/tests/test_process_row.py
@@ -0,0 +1,127 @@
+"""Tests for the process_row module"""
+
+import unittest
+from unittest.mock import patch
+from copy import deepcopy
+from boto3 import client as boto3_client
+from moto import mock_s3
+from decimal import Decimal
+
+
+from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import (    
+    MockFieldDictionaries,
+    MockFileRows,  
+)
+
+from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import (
+    GenericSetUp,
+    GenericTearDown,
+)
+from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT
+
+with patch("os.environ", MOCK_ENVIRONMENT_DICT):
+    # Do not attempt 'from src.mappings import Vaccine' as this imports a different instance of Vaccine
+    # and tests will break
+    from mappings import Vaccine
+    from clients import REGION_NAME
+    from src.process_row import process_row
+
+s3_client = boto3_client("s3", region_name=REGION_NAME)
+ROW_DETAILS = MockFieldDictionaries.all_fields
+Allowed_Operations = {'CREATE', 'UPDATE', 'DELETE'}
+
+@mock_s3
+@patch.dict("os.environ", MOCK_ENVIRONMENT_DICT)
+class TestProcessRow(unittest.TestCase):
+    """Tests for process_row"""
+
+    def setUp(self) -> None:
+        GenericSetUp(s3_client)
+
+    def tearDown(self) -> None:
+        GenericTearDown(s3_client)
+
+    def test_process_row_success(self):
+        """
+        Test that process_row gives the expected output. These tests check that the row is valid and matches the expected output.
+        """
+        # set the expected output from 'process_row' in case of success 
+        expected_result = {'resourceType': 'Immunization', 'status': 'completed', 'protocolApplied': [{'targetDisease': [], 'doseNumberPositiveInt': 1}], 'reasonCode': [{'coding': [{'system': 'http://snomed.info/sct', 'code': '1037351000000105'}]}], 'recorded': '2024-09-04', 'identifier': [{'value': 'RSV_002', 'system': 'https://www.ravs.england.nhs.uk/'}], 'patient': {'reference': '#Patient1'}, 'contained': [{'id': 'Patient1', 'resourceType': 'Patient', 'birthDate': '2008-02-17', 'gender': 'male', 'address': [{'postalCode': 'WD25 0DZ'}], 'identifier': [{'system': 'https://fhir.nhs.uk/Id/nhs-number', 'value': '9732928395'}], 'name': [{'family': 'PEEL', 'given': ['PHYLIS']}]}, {'resourceType': 'Practitioner', 'id': 'Practitioner1', 'name': [{'family': "O'Reilly", 'given': ['Ellena']}]}], 'vaccineCode': {'coding': [{'system': 'http://snomed.info/sct', 'code': '42223111000001107', 'display': 'Quadrivalent influenza vaccine (split virion, inactivated)'}]}, 'manufacturer': {'display': 'Sanofi Pasteur'}, 'expirationDate': '2024-09-15', 'lotNumber': 'BN92478105653', 'extension': [{'url': 'https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure', 'valueCodeableConcept': {'coding': [{'system': 'http://snomed.info/sct', 'code': '956951000000104', 'display': 'RSV vaccination in pregnancy (procedure)'}]}}], 'occurrenceDateTime': '2024-09-04T18:33:25+00:00', 'primarySource': True, 'site': {'coding': [{'system': 'http://snomed.info/sct', 'code': '368209003', 'display': 'Right arm'}]}, 'route': {'coding': [{'system': 'http://snomed.info/sct', 'code': '1210999013', 'display': 'Intradermal use'}]}, 'doseQuantity': {'value': Decimal('0.3'), 'unit': 'Inhalation - unit of product usage', 'system': 'http://snomed.info/sct', 'code': '2622896019'}, 'performer': [{'actor': {'type': 'Organization', 'identifier': {'system': 'https://fhir.nhs.uk/Id/ods-organization-code', 'value': 'RVVKC'}}}, {'actor': {'reference': '#Practitioner1'}}], 'location': {'identifier': {'value': 'RJC02', 'system': 'https://fhir.nhs.uk/Id/ods-organization-code'}}}
+        
+        # call 'process_row' with required details
+        imms_fhir_resource = process_row("EMIS", Allowed_Operations, ROW_DETAILS)
+        # validate if the response with expected result
+        self.assertDictEqual(imms_fhir_resource["fhir_json"], expected_result)
+    
+    def test_process_row_invalid_action_flag(self):
+        """
+        Test that process_row gives the expected output. These tests check that the row is valid and matches the expected output.
+        """
+        Mock_Row = deepcopy(ROW_DETAILS)
+        # setting up the invalid action flag other than 'NEW', 'UPDATE' or 'DELETE'
+        Mock_Row['ACTION_FLAG'] = 'Invalid'               
+
+        # call 'process_row' with required details
+        response = process_row("EMIS", Allowed_Operations, Mock_Row)
+
+        # validate if we got INVALID_ACTION_FLAG in response
+        self.assertEqual(response['diagnostics']['error_type'], 'INVALID_ACTION_FLAG')
+
+    def test_process_row_missing_action_flag(self):
+        """
+        Test that process_row gives the expected output. These tests check that the row is valid and matches the expected output.
+        """
+
+        Mock_Row = deepcopy(ROW_DETAILS)
+        # removing action flag from row 
+        Mock_Row.pop('ACTION_FLAG')               
+
+        # call 'process_row' with required details
+        response = process_row("EMIS", Allowed_Operations, Mock_Row)
+        # validate if we got INVALID_ACTION_FLAG in response
+        self.assertEqual(response['diagnostics']['error_type'], 'INVALID_ACTION_FLAG')
+
+    def test_process_row_missing_permission(self):
+        """
+        Test that process_row gives the expected output. These tests check that the row is valid and matches the expected output.
+        """
+        # only create and delete permission. Missing update
+        allowed_operation = {'CREATE', 'DELETE'}  
+        # copy row data with Action_Flag = 'Update'
+        Mock_Row = deepcopy(ROW_DETAILS)                     
+
+        # call 'process_row' with required details
+        response = process_row("EMIS", allowed_operation, Mock_Row)
+        
+        self.assertEqual(response['diagnostics']['error_type'], 'NO_PERMISSIONS')
+        self.assertEqual(response['diagnostics']['statusCode'], 403)
+
+    def test_process_row_missing_unique_id(self):
+        """
+        Test that process_row gives the expected output. These tests check that the row is valid and matches the expected output.
+        """
+        # copy row data and remove 'UNIQUE_ID'
+        Mock_Row = deepcopy(ROW_DETAILS)                     
+        Mock_Row.pop('UNIQUE_ID') 
+        # call 'process_row' with required details
+        response = process_row("EMIS", Allowed_Operations, Mock_Row)
+
+        self.assertEqual(response['diagnostics']['error_type'], 'MISSING_UNIQUE_ID')
+        self.assertEqual(response['diagnostics']['statusCode'], 400)
+
+    def test_process_row_missing_unique_id_uri(self):
+        """
+        Test that process_row gives the expected output. These tests check that the row is valid and matches the expected output.
+        """
+        # copy row data and remove 'UNIQUE_ID_URI'
+        Mock_Row = deepcopy(ROW_DETAILS)                     
+        Mock_Row.pop('UNIQUE_ID_URI') 
+        # call 'process_row' with required details
+        response = process_row("EMIS", Allowed_Operations, Mock_Row)
+        
+        self.assertEqual(response['diagnostics']['error_message'], 'UNIQUE_ID or UNIQUE_ID_URI is missing')
+        self.assertEqual(response['diagnostics']['statusCode'], 400)
+
+if __name__ == '__main__':
+    unittest.main()
+
diff --git a/recordprocessor/tests/test_send_to_kinesis.py b/recordprocessor/tests/test_send_to_kinesis.py
@@ -0,0 +1,46 @@
+import unittest
+from unittest.mock import patch
+from moto import mock_kinesis
+from boto3 import client as boto3_client
+from src.send_to_kinesis import send_to_kinesis
+
+from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import (
+    GenericSetUp,
+    GenericTearDown,
+)
+from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import (   
+    REGION_NAME,
+)
+from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT
+
+with patch("os.environ", MOCK_ENVIRONMENT_DICT):
+    from src.send_to_kinesis import send_to_kinesis
+
+kinesis_client = boto3_client("kinesis", region_name=REGION_NAME)
+
+
+@mock_kinesis
+class TestSendToKinesis(unittest.TestCase):
+
+    def setUp(self) -> None:
+        GenericSetUp(None, None, kinesis_client)
+
+    def tearDown(self) -> None:
+        GenericTearDown(None, None, kinesis_client)
+     
+    @patch.dict("os.environ", MOCK_ENVIRONMENT_DICT)
+    def test_send_to_kinesis_success(self):        
+        
+        kinesis_client.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200}}
+
+        # arrange required parameters
+        supplier = "test_supplier"
+        message_body = {"key": "value"}
+        vaccine_type = "test_vaccine"
+
+        result = send_to_kinesis(supplier, message_body, vaccine_type)
+        self.assertTrue(result)        
+   
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/recordprocessor/tests/test_unique_permission.py b/recordprocessor/tests/test_unique_permission.py
@@ -0,0 +1,21 @@
+"""Tests for the utils_for_recordprocessor module"""
+
+import unittest
+
+from src.unique_permission import get_unique_action_flags_from_s3
+from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import MockFileDetails, ValidMockFileContent
+
+class TestGetUniqueFlagsFromS3(unittest.TestCase):
+    def test_get_unique_action_flags_from_s3(self):
+        csv_data = ValidMockFileContent.with_new_and_update_and_delete
+        expected_output = {"NEW", "UPDATE", "DELETE"}
+        
+        result = get_unique_action_flags_from_s3(csv_data)
+        
+        self.assertEqual(result, expected_output)
+
+
+
+if __name__ == '__main__':
+    unittest.main()
+
diff --git a/recordprocessor/tests/utils_for_recordprocessor_tests/utils_for_recordprocessor_tests.py b/recordprocessor/tests/utils_for_recordprocessor_tests/utils_for_recordprocessor_tests.py
@@ -10,6 +10,12 @@ def convert_string_to_dict_reader(data_string: str):
     """Take a data string and convert it to a csv DictReader"""
     return DictReader(StringIO(data_string), delimiter="|")
 
+def get_csv_file_dict_reader(s3_client, bucket_name: str, file_key: str) -> DictReader:
+    """Download the file from the S3 bucket and return it as a DictReader"""
+    ack_file_csv_obj = s3_client.get_object(Bucket=bucket_name, Key=file_key)
+    csv_content_string = ack_file_csv_obj["Body"].read().decode("utf-8")
+    return DictReader(StringIO(csv_content_string), delimiter="|")
+
 
 class GenericSetUp:
     """

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,7 @@ def send_to_kinesis(supplier: str, message_body: dict, vaccine_type: str) -> boo`
`15`	`15`	`Data=json.dumps(message_body, ensure_ascii=False),`
`16`	`16`	`PartitionKey=f"{supplier}_{vaccine_type}",`
`17`	`17`	`)`
	`18`	`+ return True`
`18`	`19`	`except ClientError as error:`
`19`	`20`	`logger.error("Error sending message to Kinesis: %s", error)`
`20`	`21`	`raise`