Merge branch 'master' into AMB-2307--500-list-out-of-range

.github/workflows/sonarcloud.yml

@@ -25,9 +25,9 @@ jobs:
       
       - name: Run unittest with coverage
         run: |
-            pip install poetry moto==4.2.11 coverage botocore==1.35.49 simplejson responses structlog fhir.resources jsonpath_ng pydantic==1.10.13 requests aws-lambda-typing cffi pyjwt boto3-stubs-lite[dynamodb]~=1.26.90
-            poetry run coverage run --source=backend -m unittest discover -s backend
-            poetry run coverage xml -o sonarcloud-coverage.xml    
+          pip install poetry moto==4.2.11 coverage redis botocore==1.35.49 simplejson responses structlog fhir.resources jsonpath_ng pydantic==1.10.13 requests aws-lambda-typing cffi pyjwt boto3-stubs-lite[dynamodb]~=1.26.90 python-stdnum==1.20
+          poetry run coverage run --source=backend -m unittest discover -s backend
+          poetry run coverage xml -o sonarcloud-coverage.xml
 
       - name: SonarCloud Scan
         uses: SonarSource/sonarcloud-github-action@master

backend/pyproject.toml

@@ -22,6 +22,7 @@ cffi                    = "~1.16.0"
 jsonpath-ng             = "^1.6.0"
 simplejson              = "^3.19.2"
 structlog               = "^24.1.0"
+python-stdnum           = "^1.20"
 
 [build-system]
 requires      = ["poetry-core ~= 1.5.0"]

backend/src/models/constants.py

@@ -6,6 +6,7 @@ class Constants:
 
     STATUSES = ["completed"]
     GENDERS = ["male", "female", "other", "unknown"]
+    EXTENSION_URL = ["https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure"]
     NOT_DONE_VACCINE_CODES = ["NAVU", "UNC", "UNK", "NA"]
     ALLOWED_KEYS = {
         "Immunization": {

backend/src/models/fhir_immunization_post_validators.py

@@ -38,7 +38,7 @@ def __init__(self, imms, vaccine_type):
             FieldNames.recorded,
             FieldNames.primary_source,
             FieldNames.vaccination_procedure_code,
-            FieldNames.vaccination_procedure_display,
+            # FieldNames.vaccination_procedure_display,
             FieldNames.dose_number_positive_int,
             # FieldNames.vaccine_code_coding_code,
             # FieldNames.vaccine_code_coding_display,

backend/src/models/fhir_immunization_pre_validators.py

@@ -9,7 +9,6 @@
 from models.utils.pre_validator_utils import PreValidation
 from models.errors import MandatoryError
 from constants import Urls
-import re
 
 
 class PreValidators:
@@ -62,11 +61,6 @@ def validate(self):
             self.pre_validate_practitioner_name_family,
             self.pre_validate_recorded,
             self.pre_validate_primary_source,
-            self.pre_validate_extension,
-            self.pre_validate_extension_urls,
-            self.pre_validate_extension_value_codeable_concept_codings,
-            self.pre_validate_vaccination_procedure_code,
-            self.pre_validate_vaccination_procedure_display,
             self.pre_validate_vaccination_situation_code,
             self.pre_validate_vaccination_situation_display,
             self.pre_validate_protocol_applied,
@@ -92,6 +86,9 @@ def validate(self):
             self.pre_validate_organization_identifier_system,
             self.pre_validate_location_identifier_value,
             self.pre_validate_location_identifier_system,
+            self.pre_validate_value_codeable_concept,
+            self.pre_validate_extension_length,
+            self.pre_validate_vaccination_procedure_code,
         ]
 
         for method in validation_methods:
@@ -531,35 +528,40 @@ def pre_validate_primary_source(self, values: dict) -> dict:
             PreValidation.for_boolean(primary_source, "primarySource")
         except KeyError:
             pass
+
 
-    def pre_validate_extension(self, values: dict) -> dict:
-        """Pre-validate that extension exists"""
-        if not "extension" in values:
-            raise MandatoryError("extension is a mandatory field")
-
-    def pre_validate_extension_urls(self, values: dict) -> dict:
-        """Pre-validate that, if extension exists, then each url is unique"""
-        try:
-            PreValidation.for_unique_list(values["extension"], "url", "extension[?(@.url=='FIELD_TO_REPLACE')]")
-        except (KeyError, IndexError):
-            pass
 
-    def pre_validate_extension_value_codeable_concept_codings(self, values: dict) -> dict:
-        """Pre-validate that, if they exist, each extension[{index}].valueCodeableConcept.coding.system is unique"""
-        try:
-            for i in range(len(values["extension"])):
-                try:
-                    extension_value_codeable_concept_coding = values["extension"][i]["valueCodeableConcept"]["coding"]
-                    PreValidation.for_unique_list(
-                        extension_value_codeable_concept_coding,
-                        "system",
-                        f"extension[?(@.URL=='{values['extension'][i]['url']}']"
-                        + ".valueCodeableConcept.coding[?(@.system=='FIELD_TO_REPLACE')]",
-                    )
-                except KeyError:
-                    pass
-        except KeyError:
-            pass
+    def pre_validate_value_codeable_concept(self, values: dict) -> dict:
+        """Pre-validate that valueCodeableConcept with coding exists within extension"""
+        if "extension" not in values:
+            raise MandatoryError("extension is a mandatory field")
+
+        # Iterate over each extension and check for valueCodeableConcept and coding
+        for extension in values["extension"]:
+            if "valueCodeableConcept" not in extension:
+                raise MandatoryError("extension[?(@.url=='https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure')].valueCodeableConcept is a mandatory field")
+
+            # Check that coding exists within valueCodeableConcept
+            if "coding" not in extension["valueCodeableConcept"]:
+                raise MandatoryError("extension[?(@.url=='https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure')].valueCodeableConcept.coding is a mandatory field")    
+
+    def pre_validate_extension_length(self, values: dict) -> dict:
+            """Pre-validate that, if extension exists, then the length of the list should be 1"""
+            try:
+                field_value = values["extension"]
+                PreValidation.for_list(field_value, "extension", defined_length=1)
+                # Call the second validation method if the first validation passes
+                self.pre_validate_extension_url(values)
+            except KeyError:
+                pass 
+
+    def pre_validate_extension_url(self, values: dict) -> dict:
+            """Pre-validate that, if extension exists, then its url should be a valid one"""
+            try:
+                field_value = values["extension"][0]["url"]
+                PreValidation.for_string(field_value, "extension[0].url", predefined_values=Constants.EXTENSION_URL)
+            except KeyError:
+                pass
 
     def pre_validate_vaccination_procedure_code(self, values: dict) -> dict:
         """
@@ -574,22 +576,7 @@ def pre_validate_vaccination_procedure_code(self, values: dict) -> dict:
         try:
             field_value = get_generic_extension_value(values, url, system, field_type)
             PreValidation.for_string(field_value, field_location)
-        except (KeyError, IndexError):
-            pass
-
-    def pre_validate_vaccination_procedure_display(self, values: dict) -> dict:
-        """
-        Pre-validate that, if extension[?(@.url=='https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-
-        VaccinationProcedure')].valueCodeableConcept.coding[?(@.system=='http://snomed.info/sct')].display
-        (legacy CSV field name: VACCINATION_PROCEDURE_TERM) exists, then it is a non-empty string
-        """
-        url = "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-" + "VaccinationProcedure"
-        system = "http://snomed.info/sct"
-        field_type = "display"
-        field_location = generate_field_location_for_extension(url, system, field_type)
-        try:
-            field_value = get_generic_extension_value(values, url, system, field_type)
-            PreValidation.for_string(field_value, field_location)
+            PreValidation.for_snomed_code(field_value, field_location)
         except (KeyError, IndexError):
             pass
 

backend/src/models/field_names.py

@@ -25,7 +25,7 @@ class FieldNames:
     recorded = "recorded"
     primary_source = "primary_source"
     vaccination_procedure_code = "vaccination_procedure_code"
-    vaccination_procedure_display = "vaccination_procedure_display"
+    # vaccination_procedure_display = "vaccination_procedure_display"
     dose_number_positive_int = "dose_number_positive_int"
     vaccine_code_coding_code = "vaccine_code_coding_code"
     vaccine_code_coding_display = "vaccine_code_coding_display"

backend/src/models/obtain_field_value.py

@@ -122,10 +122,10 @@ def vaccination_procedure_code(imms: dict):
         """Obtains vaccination_procedure_code value"""
         return get_generic_extension_value(imms, Urls.vaccination_procedure, Urls.snomed, "code")
 
-    @staticmethod
-    def vaccination_procedure_display(imms: dict):
-        """Obtains vaccination_procedure_display value"""
-        return get_generic_extension_value(imms, Urls.vaccination_procedure, Urls.snomed, "display")
+    # @staticmethod
+    # def vaccination_procedure_display(imms: dict):
+    #     """Obtains vaccination_procedure_display value"""
+    #     return get_generic_extension_value(imms, Urls.vaccination_procedure, Urls.snomed, "display")
 
     @staticmethod
     def dose_number_positive_int(imms: dict):

backend/src/models/utils/generic_utils.py

@@ -6,6 +6,7 @@
 from models.constants import Constants
 import urllib.parse
 import base64
+from stdnum.verhoeff import validate
 
 
 def get_contained_resource(imms: dict, resource: Literal["Patient", "Practitioner", "QuestionnaireResponse"]):
@@ -66,6 +67,15 @@ def check_for_unknown_elements(resource, resource_type) -> Union[None, list]:
             errors.append(f"{key} is not an allowed element of the {resource_type} resource for this service")
     return errors
 
+def is_valid_simple_snomed(simple_snomed: str) -> bool:
+    "check the snomed code valid or not."
+    min_snomed_length = 6
+    max_snomed_length = 18
+    return (simple_snomed is not None
+    and simple_snomed.isdigit()
+    and min_snomed_length <= len(simple_snomed) <= max_snomed_length
+    and validate(simple_snomed)
+    and (simple_snomed[-3:-1] in ("00", "10")))
 
 def nhs_number_mod11_check(nhs_number: str) -> bool:
     """

backend/src/models/utils/pre_validator_utils.py

@@ -3,7 +3,7 @@
 from decimal import Decimal
 from typing import Union
 
-from .generic_utils import nhs_number_mod11_check
+from .generic_utils import nhs_number_mod11_check, is_valid_simple_snomed
 
 
 class PreValidation:
@@ -146,6 +146,24 @@ def for_date_time(field_value: str, field_location: str):
                     datetime.strptime(field_value, "%Y-%m-%dT%H:%M:%S.%f%z")
                 except ValueError as error:
                     raise ValueError(error_message) from error
+
+    @staticmethod
+    def for_snomed_code(field_value: str, field_location: str):
+        """
+        Apply prevalidation to snomed code to ensure that its a valid one.
+        """
+
+        error_message = (
+            f"{field_location} is not a valid snomed code"
+        )
+
+        try:
+            is_valid = is_valid_simple_snomed(field_value)  
+        except Exception:
+            raise ValueError(error_message)
+        if not is_valid:
+            raise ValueError(error_message)
+
 
     @staticmethod
     def for_boolean(field_value: str, field_location: str):

backend/tests/sample_data/completed_hpv_immunization_event.json

@@ -36,7 +36,7 @@
         "coding": [
           {
             "system": "http://snomed.info/sct",
-            "code": "mockHPVcode1",
+            "code": "822851000000102",
             "display": "Seasonal influenza vaccination (procedure)"
           }
         ]

backend/tests/sample_data/completed_mmr_immunization_event.json

@@ -36,7 +36,7 @@
         "coding": [
           {
             "system": "http://snomed.info/sct",
-            "code": "mockMMRcode1",
+            "code": "822851000000102",
             "display": "Seasonal influenza vaccination (procedure)"
           }
         ]

backend/tests/test_immunization_post_validator.py

@@ -206,14 +206,6 @@ def test_post_primary_source(self):
         """Test that the JSON data is rejected if it does not contain primary_source"""
         MandationTests.test_missing_mandatory_field_rejected(self, "primarySource")
 
-    def test_post_vaccination_procedure_display(self):
-        """Test that the JSON data is accepted if it does not contain vaccination_procedure_display"""
-        field_location = (
-            "extension[?(@.url=='https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure')]"
-            + ".valueCodeableConcept.coding[?(@.system=='http://snomed.info/sct')].display"
-        )
-        MandationTests.test_missing_field_accepted(self, field_location)
-
     # TODO: To confirm with imms if dose number string validation is correct (current working assumption is yes)
     def test_post_dose_number_positive_int(self):
         """
@@ -358,6 +350,23 @@ def test_post_organization_identifier_system(self):
         MandationTests.test_missing_mandatory_field_rejected(
             self, "performer[?(@.actor.type=='Organization')].actor.identifier.system"
         )
+
+    def test_pre_validate_extension_url(self):
+        """
+        Test pre_validate_extension_url accepts valid values and rejects 
+        if the snomed code are unable to fetch if the url is invalid
+        and get passed only with the snomed url.
+        """
+        # Test case: missing "extension"
+        invalid_json_data = deepcopy(self.completed_json_data[VaccineTypes.covid_19])
+        invalid_json_data["extension"][0]["valueCodeableConcept"]["coding"][0]["system"]='https://xyz/Extension-UKCore-VaccinationProcedure'
+
+        with self.assertRaises(Exception) as error:
+            self.validator.validate(invalid_json_data)
+
+        full_error_message = str(error.exception)
+        actual_error_messages = full_error_message.replace("Validation errors: ", "").split("; ")
+        self.assertIn("extension[?(@.url=='https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure')].valueCodeableConcept.coding[?(@.system=='http://snomed.info/sct')].code is a mandatory field", actual_error_messages)
 
     def test_post_location_identifier_value(self):
         """