Move constants to relevant file

Author: edhall-nhs

lambdas/recordprocessor/src/constants.py

@@ -49,6 +49,11 @@
     "LOCATION_CODE_TYPE_URI",
 ]
 
+TPP_V2_SUPPLIER_IDENTIFIER_SYSTEM = "YGA"
+TPP_V5_SUPPLIER_IDENTIFIER_SYSTEM = "https://tpp-uk.com/Id/ve/vacc"
+EMIS_V2_SUPPLIER_IDENTIFIER_SYSTEM = "YGJ"
+EMIS_V5_SUPPLIER_IDENTIFIER_SYSTEM = "https://emishealth.com/identifiers/vacc"
+
 
 class FileStatus:
     """File status constants"""

lambdas/recordprocessor/src/process_row.py

@@ -1,15 +1,16 @@
 """Function to process a single row of a csv file"""
 
 from common.clients import logger
-from constants import Diagnostics
+from constants import (
+    EMIS_V2_SUPPLIER_IDENTIFIER_SYSTEM,
+    EMIS_V5_SUPPLIER_IDENTIFIER_SYSTEM,
+    TPP_V2_SUPPLIER_IDENTIFIER_SYSTEM,
+    TPP_V5_SUPPLIER_IDENTIFIER_SYSTEM,
+    Diagnostics,
+)
 from convert_to_fhir_imms_resource import convert_to_fhir_imms_resource
 from utils_for_recordprocessor import create_diagnostics_dictionary
 
-TPP_V2_SUPPLIER_IDENTIFIER_SYSTEM = "YGA"
-TPP_V5_SUPPLIER_IDENTIFIER_SYSTEM = "https://tpp-uk.com/Id/ve/vacc"
-EMIS_V2_SUPPLIER_IDENTIFIER_SYSTEM = "YGJ"
-EMIS_V5_SUPPLIER_IDENTIFIER_SYSTEM = "https://emishealth.com/identifiers/vacc"
-
 
 def process_row(target_disease: list, allowed_operations: set, row: dict) -> dict:
     """
@@ -20,7 +21,7 @@ def process_row(target_disease: list, allowed_operations: set, row: dict) -> dic
     action_flag = (row.get("ACTION_FLAG") or "").upper()
     unique_id_uri = row.get("UNIQUE_ID_URI")
 
-    # This code the above constants can be safely removed once DPS carries out it's data migration to update legacy
+    # This code the relevant constants can be safely removed once DPS carries out it's data migration to update legacy
     # identifiers as it should become redundant. However, it may be worth keeping in case legacy format identifiers are
     # received for some reason. Please see issue VED-904 for more information.
     if unique_id_uri == TPP_V2_SUPPLIER_IDENTIFIER_SYSTEM:

lambdas/recordprocessor/tests/test_process_row.py

@@ -160,8 +160,6 @@ 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.
         """
-        self.maxDiff = None
-
         # call 'process_row' with required details
         imms_fhir_resource = process_row(TargetDiseaseElements.RSV, Allowed_Operations, ROW_DETAILS)
         # validate if the response with expected result
@@ -243,41 +241,37 @@ def test_process_row_missing_unique_id_uri(self):
         )
         self.assertEqual(response["diagnostics"]["statusCode"], 400)
 
-    def test_process_row_successfully_uplifts_legacy_TPP_uri(self):
+    def test_process_row_successfully_uplifts_legacy_tpp_uri(self):
         """
         Test that process_row gives the expected output.
         These tests check that the row is valid and matches the expected output.
         """
-        legacy_TPP_row = deepcopy(ROW_DETAILS)
-        legacy_TPP_row["UNIQUE_ID_URI"] = "YGA"
-
-        expected_successful_result_TPP = deepcopy(expected_successful_result)
-        expected_successful_result_TPP["identifier"][0]["system"] = "https://tpp-uk.com/Id/ve/vacc"
+        legacy_tpp_row = deepcopy(ROW_DETAILS)
+        legacy_tpp_row["UNIQUE_ID_URI"] = "YGA"
 
-        self.maxDiff = None
+        expected_successful_result_tpp = deepcopy(expected_successful_result)
+        expected_successful_result_tpp["identifier"][0]["system"] = "https://tpp-uk.com/Id/ve/vacc"
 
         # call 'process_row' with required details
-        imms_fhir_resource = process_row(TargetDiseaseElements.RSV, Allowed_Operations, legacy_TPP_row)
+        imms_fhir_resource = process_row(TargetDiseaseElements.RSV, Allowed_Operations, legacy_tpp_row)
         # validate if the response with expected result
-        self.assertDictEqual(imms_fhir_resource["fhir_json"], expected_successful_result_TPP)
+        self.assertDictEqual(imms_fhir_resource["fhir_json"], expected_successful_result_tpp)
 
-    def test_process_row_successfully_uplifts_legacy_EMIS_uri(self):
+    def test_process_row_successfully_uplifts_legacy_emis_uri(self):
         """
         Test that process_row gives the expected output.
         These tests check that the row is valid and matches the expected output.
         """
-        legacy_EMIS_row = deepcopy(ROW_DETAILS)
-        legacy_EMIS_row["UNIQUE_ID_URI"] = "YGJ"
-
-        expected_successful_result_EMIS = deepcopy(expected_successful_result)
-        expected_successful_result_EMIS["identifier"][0]["system"] = "https://emishealth.com/identifiers/vacc"
+        legacy_emis_row = deepcopy(ROW_DETAILS)
+        legacy_emis_row["UNIQUE_ID_URI"] = "YGJ"
 
-        self.maxDiff = None
+        expected_successful_result_emis = deepcopy(expected_successful_result)
+        expected_successful_result_emis["identifier"][0]["system"] = "https://emishealth.com/identifiers/vacc"
 
         # call 'process_row' with required details
-        imms_fhir_resource = process_row(TargetDiseaseElements.RSV, Allowed_Operations, legacy_EMIS_row)
+        imms_fhir_resource = process_row(TargetDiseaseElements.RSV, Allowed_Operations, legacy_emis_row)
         # validate if the response with expected result
-        self.assertDictEqual(imms_fhir_resource["fhir_json"], expected_successful_result_EMIS)
+        self.assertDictEqual(imms_fhir_resource["fhir_json"], expected_successful_result_emis)
 
 
 if __name__ == "__main__":