Add logic to uplift legacy identifiers

Author: edhall-nhs

lambdas/recordforwarder/src/service/fhir_batch_service.py

@@ -4,6 +4,23 @@
 
 IMMUNIZATION_VALIDATOR = ImmunizationValidator()
 
+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 _uplift_legacy_identifier(immunization: dict):
+    # This code the above constants can be safely removed once DPS carries out it's data migration to update legacy
+    # identifiers as it will become redundant. Please see issue VED-904 for more information.
+    identifier_system = immunization["identifier"][0]["system"]
+
+    if identifier_system == TPP_V2_SUPPLIER_IDENTIFIER_SYSTEM:
+        immunization["identifier"][0]["system"] = TPP_V5_SUPPLIER_IDENTIFIER_SYSTEM
+
+    if identifier_system == EMIS_V2_SUPPLIER_IDENTIFIER_SYSTEM:
+        immunization["identifier"][0]["system"] = EMIS_V5_SUPPLIER_IDENTIFIER_SYSTEM
+
 
 class ImmunizationBatchService:
     def __init__(
@@ -27,6 +44,10 @@ def create_immunization(
         Exception will be raised if resource exits. Multiple calls to this method won't change
         the record in the database.
         """
+
+        # TODO: Remove after DPS data migration to new identifiers
+        _uplift_legacy_identifier(immunization)
+
         try:
             self.validator.validate(immunization)
         except (ValueError, MandatoryError) as error:
@@ -47,6 +68,10 @@ def update_immunization(
         Exception will be raised if resource didn't exist.Multiple calls to this method won't change
         the record in the database.
         """
+
+        # TODO: Remove after DPS data migration to new identifiers
+        _uplift_legacy_identifier(immunization)
+
         try:
             self.validator.validate(immunization)
         except (ValueError, MandatoryError) as error:
@@ -67,4 +92,8 @@ def delete_immunization(
         Exception will be raised if resource didn't exist.Multiple calls to this method won't change
         the record in the database.
         """
+
+        # TODO: Remove after DPS data migration to new identifiers
+        _uplift_legacy_identifier(immunization)
+
         return self.immunization_repo.delete_immunization(immunization, supplier_system, vax_type, table, is_present)

lambdas/recordforwarder/tests/service/test_fhir_batch_service.py

@@ -97,6 +97,38 @@ def test_create_immunization_post_validation_error(self):
         self.assertTrue(expected_msg in error.exception.message)
         self.mock_repo.create_immunization.assert_not_called()
 
+    def test_create_immunization_uplifts_legacy_identifier_system_tpp(self):
+        """it should recognise a legacy TPP identifier system and use the new style identifier"""
+
+        immunisation = create_covid_immunization_dict_no_id()
+        immunisation["identifier"][0]["system"] = "YGA"
+
+        _ = self.service.create_immunization(
+            immunization=immunisation,
+            supplier_system="test_supplier",
+            vax_type="test_vax",
+            table=self.mock_table,
+            is_present=True,
+        )
+
+        self.assertEqual(immunisation["identifier"][0]["system"], "https://tpp-uk.com/Id/ve/vacc")
+
+    def test_create_immunization_uplifts_legacy_identifier_system_emis(self):
+        """it should recognise a legacy EMIS identifier system and use the new style identifier"""
+
+        immunisation = create_covid_immunization_dict_no_id()
+        immunisation["identifier"][0]["system"] = "YGJ"
+
+        _ = self.service.create_immunization(
+            immunization=immunisation,
+            supplier_system="test_supplier",
+            vax_type="test_vax",
+            table=self.mock_table,
+            is_present=True,
+        )
+
+        self.assertEqual(immunisation["identifier"][0]["system"], "https://emishealth.com/identifiers/vacc")
+
 
 class TestUpdateImmunizationBatchService(TestFhirBatchServiceBase):
     def setUp(self):
@@ -169,6 +201,38 @@ def test_update_immunization_post_validation_error(self):
         self.assertTrue(expected_msg in error.exception.message)
         self.mock_repo.update_immunization.assert_not_called()
 
+    def test_update_immunization_uplifts_legacy_identifier_system_tpp(self):
+        """it should recognise a legacy TPP identifier system and use the new style identifier"""
+
+        immunisation = create_covid_immunization_dict_no_id()
+        immunisation["identifier"][0]["system"] = "YGA"
+
+        _ = self.service.update_immunization(
+            immunization=immunisation,
+            supplier_system="test_supplier",
+            vax_type="test_vax",
+            table=self.mock_table,
+            is_present=True,
+        )
+
+        self.assertEqual(immunisation["identifier"][0]["system"], "https://tpp-uk.com/Id/ve/vacc")
+
+    def test_update_immunization_uplifts_legacy_identifier_system_emis(self):
+        """it should recognise a legacy EMIS identifier system and use the new style identifier"""
+
+        immunisation = create_covid_immunization_dict_no_id()
+        immunisation["identifier"][0]["system"] = "YGJ"
+
+        _ = self.service.update_immunization(
+            immunization=immunisation,
+            supplier_system="test_supplier",
+            vax_type="test_vax",
+            table=self.mock_table,
+            is_present=True,
+        )
+
+        self.assertEqual(immunisation["identifier"][0]["system"], "https://emishealth.com/identifiers/vacc")
+
 
 class TestDeleteImmunizationBatchService(unittest.TestCase):
     def setUp(self):
@@ -194,6 +258,38 @@ def test_delete_immunization_valid(self):
         )
         self.assertEqual(result, imms_id)
 
+    def test_delete_immunization_uplifts_legacy_identifier_system_tpp(self):
+        """it should recognise a legacy TPP identifier system and use the new style identifier"""
+
+        immunisation = create_covid_immunization_dict_no_id()
+        immunisation["identifier"][0]["system"] = "YGA"
+
+        _ = self.service.delete_immunization(
+            immunization=immunisation,
+            supplier_system="test_supplier",
+            vax_type="test_vax",
+            table=self.mock_table,
+            is_present=True,
+        )
+
+        self.assertEqual(immunisation["identifier"][0]["system"], "https://tpp-uk.com/Id/ve/vacc")
+
+    def test_delete_immunization_uplifts_legacy_identifier_system_emis(self):
+        """it should recognise a legacy EMIS identifier system and use the new style identifier"""
+
+        immunisation = create_covid_immunization_dict_no_id()
+        immunisation["identifier"][0]["system"] = "YGJ"
+
+        _ = self.service.delete_immunization(
+            immunization=immunisation,
+            supplier_system="test_supplier",
+            vax_type="test_vax",
+            table=self.mock_table,
+            is_present=True,
+        )
+
+        self.assertEqual(immunisation["identifier"][0]["system"], "https://emishealth.com/identifiers/vacc")
+
 
 if __name__ == "__main__":
     unittest.main()