Merge branch 'master' into dependabot/github_actions/docker/setup-buildx-action-4.0.0

Author: Thomas-Boyle

lambdas/recordforwarder/tests/service/test_fhir_batch_service.py

@@ -101,6 +101,29 @@ 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_invalid_patient_identifier_system(self):
+        """it should reject create when the patient identifier system is not the NHS number URI"""
+
+        imms = create_covid_immunization_dict_no_id()
+        patient = next(resource for resource in imms["contained"] if resource["resourceType"] == "Patient")
+        patient["identifier"][0]["system"] = "https://1234/Id/nhs-number"
+        patient["identifier"][0]["value"] = "ABCD"
+
+        with self.assertRaises(CustomValidationError) as error:
+            self.pre_validate_fhir_service.create_immunization(
+                immunization=imms,
+                supplier_system="test_supplier",
+                vax_type="test_vax",
+                table=self.mock_table,
+                imms_pk=None,
+            )
+
+        self.assertIn(
+            "contained[?(@.resourceType=='Patient')].identifier[0].system must equal",
+            error.exception.message,
+        )
+        self.mock_repo.create_immunization.assert_not_called()
+
     def test_create_immunization_duplicate_site_snomed_still_rejected_for_batch(self):
         """it should keep batch validation unchanged for duplicate site SNOMED codings"""
 
@@ -192,6 +215,29 @@ 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_invalid_patient_identifier_system(self):
+        """it should reject update when the patient identifier system is not the NHS number URI"""
+
+        imms = create_covid_immunization_dict_no_id()
+        patient = next(resource for resource in imms["contained"] if resource["resourceType"] == "Patient")
+        patient["identifier"][0]["system"] = "https://1234/Id/nhs-number"
+        patient["identifier"][0]["value"] = "ABCD"
+
+        with self.assertRaises(CustomValidationError) as error:
+            self.pre_validate_fhir_service.update_immunization(
+                immunization=imms,
+                supplier_system="test_supplier",
+                vax_type="test_vax",
+                table=self.mock_table,
+                imms_pk=None,
+            )
+
+        self.assertIn(
+            "contained[?(@.resourceType=='Patient')].identifier[0].system must equal",
+            error.exception.message,
+        )
+        self.mock_repo.update_immunization.assert_not_called()
+
 
 class TestDeleteImmunizationBatchService(unittest.TestCase):
     def setUp(self):

lambdas/shared/src/common/models/fhir_immunization_pre_validators.py

@@ -46,6 +46,7 @@ def validate(self):
             self.pre_validate_patient_identifier_extension,
             self.pre_validate_patient_identifier,
             self.pre_validate_patient_identifier_system,
+            self.pre_validate_patient_identifier_nhs_system,
             self.pre_validate_patient_identifier_value,
             self.pre_validate_patient_name,
             self.pre_validate_patient_name_given,
@@ -286,6 +287,21 @@ def pre_validate_patient_identifier_system(self, values: dict) -> None:
         except (KeyError, IndexError):
             pass
 
+    def pre_validate_patient_identifier_nhs_system(self, values: dict) -> None:
+        """
+        Pre-validate that, if the contained Patient has an identifier system,
+        it uses the NHS number system.
+        """
+        field_location = "contained[?(@.resourceType=='Patient')].identifier[0].system"
+        try:
+            field_value = [x for x in values["contained"] if x.get("resourceType") == "Patient"][0]["identifier"][0][
+                "system"
+            ]
+            if field_value != Urls.NHS_NUMBER:
+                raise ValueError(f"{field_location} must equal '{Urls.NHS_NUMBER}'")
+        except (KeyError, IndexError):
+            pass
+
     def pre_validate_patient_identifier_value(self, values: dict) -> None:
         """
         Pre-validate that, if the contained Patient has an NHS-number identifier,

lambdas/shared/tests/test_common/test_immunization_pre_validator.py

@@ -470,24 +470,30 @@ def test_pre_validate_patient_identifier_value(self):
             ],
         )
 
-    def test_pre_validate_patient_identifier_value_accepts_non_nhs_identifier(self):
-        """Test pre_validate_patient_identifier_value ignores non-NHS patient identifiers"""
-        valid_json_data = deepcopy(self.json_data)
-        valid_json_data["contained"][1]["identifier"] = [
+    def test_pre_validate_patient_identifier_rejects_non_nhs_identifier_system(self):
+        """Test pre_validate_patient_identifier rejects non-NHS patient identifier systems"""
+        invalid_json_data = deepcopy(self.json_data)
+        invalid_json_data["contained"][1]["identifier"] = [
             {
                 "system": "https://someother.codeableconcept.com/",
                 "value": "TVC15",
             }
         ]
 
-        self.assertIsNone(self.validator.validate(valid_json_data))
+        with self.assertRaises(ValueError) as error:
+            self.validator.validate(invalid_json_data)
+
+        self.assertIn(
+            "contained[?(@.resourceType=='Patient')].identifier[0].system must equal '",
+            str(error.exception),
+        )
 
     def test_pre_validate_patient_identifier_system(self):
         """Test pre_validate_patient_identifier_system accepts valid values and rejects invalid values"""
         ValidatorModelTests.test_string_value(
             self,
             field_location="contained[?(@.resourceType=='Patient')].identifier[0].system",
-            valid_strings_to_test=[Urls.NHS_NUMBER, "https://someother.codeableconcept.com/"],
+            valid_strings_to_test=[Urls.NHS_NUMBER],
         )
 
     def test_pre_validate_patient_name(self):