add more expression rules and fields

Akol125 · Akol125 · commit b08c49f4e315 · 2025-11-13T16:16:23.000Z
diff --git a/backend/src/models/fhir_immunization_pre_validators.py b/backend/src/models/fhir_immunization_pre_validators.py
@@ -96,7 +96,9 @@ def validate(self):
             self.pre_validate_value_codeable_concept,
             self.pre_validate_extension_length,
             self.pre_validate_vaccination_procedure_code,
+            self.pre_validate_vaccination_procedure_display,
             self.pre_validate_vaccine_code,
+            self.pre_validate_vaccine_display,
         ]
 
         for method in validation_methods:
@@ -590,7 +592,7 @@ def pre_validate_vaccination_procedure_code(self, values: dict) -> dict:
         (legacy CSV field name: VACCINATION_PROCEDURE_CODE) exists, then it is a non-empty string
         """
         url = "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-" + "VaccinationProcedure"
-        system = "http://snomed.info/sct"
+        system = Urls.snomed
         field_type = "code"
         field_location = generate_field_location_for_extension(url, system, field_type)
         try:
@@ -600,14 +602,30 @@ def pre_validate_vaccination_procedure_code(self, values: dict) -> dict:
         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 = Urls.snomed
+        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)
+        except (KeyError, IndexError):
+            pass
+
     def pre_validate_vaccination_situation_code(self, values: dict) -> dict:
         """
         Pre-validate that, if extension[?(@.url=='https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-
         VaccinationSituation')].valueCodeableConcept.coding[?(@.system=='http://snomed.info/sct')].code
         (legacy CSV field name: VACCINATION_SITUATION_CODE) exists, then it is a non-empty string
         """
         url = "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationSituation"
-        system = "http://snomed.info/sct"
+        system = Urls.snomed
         field_type = "code"
         field_location = generate_field_location_for_extension(url, system, field_type)
         try:
@@ -623,7 +641,7 @@ def pre_validate_vaccination_situation_display(self, values: dict) -> dict:
         (legacy CSV field name: VACCINATION_SITUATION_TERM) exists, then it is a non-empty string
         """
         url = "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationSituation"
-        system = "http://snomed.info/sct"
+        system = Urls.snomed
         field_type = "display"
         field_location = generate_field_location_for_extension(url, system, field_type)
         try:
@@ -702,7 +720,7 @@ def pre_validate_disease_type_coding_codes(self, values: dict) -> dict:
         Pre-validate that, if protocolApplied[0].targetDisease[{i}].coding[?(@.system=='http://snomed.info/sct')].code
         exists, then it is a non-empty string
         """
-        url = "http://snomed.info/sct"
+        url = Urls.snomed
         try:
             for i in range(len(values["protocolApplied"][0]["targetDisease"])):
                 field_location = f"protocolApplied[0].targetDisease[{i}].coding[?(@.system=='{url}')].code"
@@ -761,7 +779,7 @@ def pre_validate_site_coding_code(self, values: dict) -> dict:
         Pre-validate that, if site.coding[?(@.system=='http://snomed.info/sct')].code
         (legacy CSV field name: SITE_OF_VACCINATION_CODE) exists, then it is a non-empty string
         """
-        url = "http://snomed.info/sct"
+        url = Urls.snomed
         field_location = f"site.coding[?(@.system=='{url}')].code"
         try:
             site_coding_code = [x for x in values["site"]["coding"] if x.get("system") == url][0]["code"]
@@ -774,7 +792,7 @@ def pre_validate_site_coding_display(self, values: dict) -> dict:
         Pre-validate that, if site.coding[?(@.system=='http://snomed.info/sct')].display
         (legacy CSV field name: SITE_OF_VACCINATION_TERM) exists, then it is a non-empty string
         """
-        url = "http://snomed.info/sct"
+        url = Urls.snomed
         field_location = f"site.coding[?(@.system=='{url}')].display"
         try:
             field_value = [x for x in values["site"]["coding"] if x.get("system") == url][0]["display"]
@@ -795,7 +813,7 @@ def pre_validate_route_coding_code(self, values: dict) -> dict:
         Pre-validate that, if route.coding[?(@.system=='http://snomed.info/sct')].code
         (legacy CSV field name: ROUTE_OF_VACCINATION_CODE) exists, then it is a non-empty string
         """
-        url = "http://snomed.info/sct"
+        url = Urls.snomed
         field_location = f"route.coding[?(@.system=='{url}')].code"
         try:
             field_value = [x for x in values["route"]["coding"] if x.get("system") == url][0]["code"]
@@ -808,7 +826,7 @@ def pre_validate_route_coding_display(self, values: dict) -> dict:
         Pre-validate that, if route.coding[?(@.system=='http://snomed.info/sct')].display
         (legacy CSV field name: ROUTE_OF_VACCINATION_TERM) exists, then it is a non-empty string
         """
-        url = "http://snomed.info/sct"
+        url = Urls.snomed
         field_location = f"route.coding[?(@.system=='{url}')].display"
         try:
             field_value = [x for x in values["route"]["coding"] if x.get("system") == url][0]["display"]
@@ -951,11 +969,24 @@ def pre_validate_vaccine_code(self, values: dict) -> dict:
         NOTE: vaccineCode is a mandatory FHIR field. A value of None will be rejected by the
         FHIR model before pre-validators are run.
         """
-        url = "http://snomed.info/sct"
+        url = Urls.snomed
         field_location = f"vaccineCode.coding[?(@.system=='{url}')].code"
         try:
             field_value = [x for x in values["vaccineCode"]["coding"] if x.get("system") == url][0]["code"]
             PreValidation.for_string(field_value, field_location)
             PreValidation.for_snomed_code(field_value, field_location)
         except (KeyError, IndexError):
             pass
+
+    def pre_validate_vaccine_display(self, values: dict) -> dict:
+        """
+        Pre-validate that, if vaccineCode.coding[?(@.system=='http://snomed.info/sct')].display
+        (legacy CSV field : VACCINE_PRODUCT_TERM) exists, then it is a non-empty string
+        """
+        url = Urls.snomed
+        field_location = f"vaccineCode.coding[?(@.system=='{url}')].display"
+        try:
+            field_value = [x for x in values["vaccineCode"]["coding"] if x.get("system") == url][0]["display"]
+            PreValidation.for_string(field_value, field_location)
+        except (KeyError, IndexError):
+            pass
diff --git a/lambdas/shared/src/common/validator/constants/constants.py b/lambdas/shared/src/common/validator/constants/constants.py
@@ -2,3 +2,28 @@ class Constants:
     NHS_NUMBER_LENGTH = 10
     PERSON_NAME_ELEMENT_MAX_LENGTH = 35
     GENDERS = ["male", "female", "other", "unknown"]
+    DATETIME_FORMAT = "%Y-%m-%dT%H:%M"
+    DATETIME_FORMAT = [
+        "%Y-%m-%d",
+        "%Y-%m-%dT%H:%M:%S%z",
+        "%Y-%m-%dT%H:%M:%S.%f%z",
+    ]
+    ALLOWED_SUFFIXES = {
+        "+00:00",
+        "+01:00",
+        "+0000",
+        "+0100",
+    }
+    field_name = "FIELD_TO_REPLACE"
+
+    DATETIME_ERROR_MESSAGE = (
+        f"{field_name} must be a valid datetime in one of the following formats:"
+        "- 'YYYY-MM-DD' — Full date only"
+        "- 'YYYY-MM-DDThh:mm:ss%z' — Full date and time with timezone (e.g. +00:00 or +01:00)"
+        "- 'YYYY-MM-DDThh:mm:ss.f%z' — Full date and time with milliseconds and timezone"
+        "-  Date must not be in the future."
+    )
+    STRICT_DATETIME_ERROR_MESSAGE = (
+        "Only '+00:00' and '+01:00' are accepted as valid timezone offsets.\n"
+        f"Note that partial dates are not allowed for {field_name} in this service.\n"
+    )
diff --git a/lambdas/shared/src/common/validator/error_report/record_error.py b/lambdas/shared/src/common/validator/error_report/record_error.py
@@ -21,6 +21,9 @@ def __init__(
         self.id = None
         self.error_level = error_level
 
+    def __repr__(self):
+        return f"<ErrorReport code={self.code}, field={self.field}, message={self.message!r}, details={self.details!r}>"
+
     # function to return the object as a dictionary
     def to_dict(self):
         ret = {"code": self.code, "message": self.message}
diff --git a/lambdas/shared/src/common/validator/expression_checker.py b/lambdas/shared/src/common/validator/expression_checker.py
@@ -9,6 +9,7 @@
 from common.validator.lookup_expressions.key_data import KeyData
 from common.validator.lookup_expressions.lookup_data import LookUpData
 from common.validator.validation_utils import check_if_future_date
+from src.common.validator.constants.constants import Constants
 
 
 class ExpressionChecker:
@@ -51,7 +52,7 @@ def validate_expression(
                 return "Schema expression not found! Check your expression type : " + expression_type
 
     # ISO 8601 date/datetime validate (currently date-only)
-    def validation_for_date(self, _expression_rule, field_name, field_value):
+    def validation_for_date(self, _expression_rule, field_name, field_value) -> ErrorReport:
         """
         Apply pre-validation to a date field to ensure that it is a string (JSON dates must be
         written as strings) containing a valid date in the format "YYYY-MM-DD"
@@ -201,63 +202,50 @@ def validation_for_list(self, expression_rule: str, field_name: str, field_value
                 message = MESSAGES[ExceptionLevels.UNEXPECTED_EXCEPTION] % (e.__class__.__name__, e)
                 return ErrorReport(ExceptionLevels.UNEXPECTED_EXCEPTION, message, None, field_name, "")
 
-    def validation_for_date_time(
-        self, expression_rule: str, field_name: str, field_value: str, row: dict, strict_timezone: bool = True
-    ):
+    def validation_for_date_time(self, expression_rule: str, field_name: str, field_value: str) -> ErrorReport:
         """
         Apply pre-validation to a datetime field to ensure that it is a string (JSON dates must be written as strings)
         containing a valid datetime. Note that partial dates are valid for FHIR, but are not allowed for this API.
-        Valid formats are any of the following:
-        * 'YYYY-MM-DD' - Full date only
-        * 'YYYY-MM-DDThh:mm:ss%z' - Full date, time without milliseconds, timezone
+        Valid formats are any of the following: * 'YYYY-MM-DD' - Full date only * 'YYYY-MM-DDThh:mm:ss%z' - Full date, time without milliseconds, timezone
         * 'YYYY-MM-DDThh:mm:ss.f%z' - Full date, time with milliseconds (any level of precision), timezone
         """
+        rules = expression_rule_per_field(expression_rule) if expression_rule else {}
+        strict_timezone = rules.get("strict_time_zone", False)
+        try:
+            if not isinstance(field_value, str):
+                raise TypeError(f"{field_name} must be a string")
 
-        if not isinstance(field_value, str):
-            raise TypeError(f"{field_name} must be a string")
-
-        error_message = (
-            f"{field_name} must be a valid datetime in one of the following formats:"
-            "- 'YYYY-MM-DD' — Full date only"
-            "- 'YYYY-MM-DDThh:mm:ss%z' — Full date and time with timezone (e.g. +00:00 or +01:00)"
-            "- 'YYYY-MM-DDThh:mm:ss.f%z' — Full date and time with milliseconds and timezone"
-            "-  Date must not be in the future."
-        )
-        if strict_timezone:
-            error_message += (
-                "Only '+00:00' and '+01:00' are accepted as valid timezone offsets.\n"
-                f"Note that partial dates are not allowed for {field_name} in this service.\n"
-            )
-
-        allowed_suffixes = {
-            "+00:00",
-            "+01:00",
-            "+0000",
-            "+0100",
-        }
-
-        # List of accepted strict formats
-        formats = [
-            "%Y-%m-%d",
-            "%Y-%m-%dT%H:%M:%S%z",
-            "%Y-%m-%dT%H:%M:%S.%f%z",
-        ]
-
-        for fmt in formats:
-            try:
-                fhir_date = datetime.strptime(field_value, fmt)
-                # Enforce future-date rule using central checker after successful parse
-                if check_if_future_date(fhir_date):
-                    raise ValueError(f"{field_name} must not be in the future")
-                # After successful parse, enforce timezone and future-date rules
-                if strict_timezone and fhir_date.tzinfo is not None:
-                    if not any(field_value.endswith(suffix) for suffix in allowed_suffixes):
-                        raise ValueError(error_message)
-                return fhir_date.isoformat()
-            except ValueError:
-                continue
-
-        raise ValueError(error_message)
+            error_message = Constants.DATETIME_ERROR_MESSAGE.replace("FIELD_TO_REPLACE", field_name)
+            if strict_timezone:
+                error_message += Constants.STRICT_DATETIME_ERROR_MESSAGE.replace("FIELD_TO_REPLACE", field_name)
+
+            # List of accepted strict formats and suffixes
+            allowed_suffixes = Constants.ALLOWED_SUFFIXES
+            formats = Constants.DATETIME_FORMAT
+
+            for fmt in formats:
+                try:
+                    fhir_date = datetime.strptime(field_value, fmt)
+                    # Enforce future-date rule using central checker after successful parse
+                    if check_if_future_date(fhir_date):
+                        raise ValueError(f"{field_name} must not be in the future")
+                    # After successful parse, enforce timezone and future-date rules
+                    if strict_timezone and fhir_date.tzinfo is not None:
+                        if not any(field_value.endswith(suffix) for suffix in allowed_suffixes):
+                            raise ValueError(error_message)
+                    return None
+                except ValueError:
+                    continue
+            raise ValueError(error_message)
+        except (TypeError, ValueError) as e:
+            code = ExceptionLevels.RECORD_CHECK_FAILED
+            message = MESSAGES[ExceptionLevels.RECORD_CHECK_FAILED]
+            details = str(e)
+            return ErrorReport(code, message, None, field_name, details)
+        except Exception as e:
+            if self.report_unexpected_exception:
+                message = MESSAGES[ExceptionLevels.UNEXPECTED_EXCEPTION] % (e.__class__.__name__, e)
+                return ErrorReport(ExceptionLevels.UNEXPECTED_EXCEPTION, message, None, field_name)
 
     # Not Equal Validate
     def _validate_not_equal(self, expression_rule: str, field_name: str, field_value: str, row: dict) -> ErrorReport:
@@ -365,7 +353,7 @@ def validation_for_string_values(self, expression_rule: str, field_name: str, fi
         defined_length = rules.get("defined_length", None)
         max_length = rules.get("max_length", None)
         predefined_values = rules.get("predefined_values", None)
-        spaces_allowed = rules.get("spaces_allowed", None)
+        spaces_allowed = rules.get("spaces_allowed", True)
 
         try:
             if not isinstance(field_value, str):
diff --git a/lambdas/shared/src/common/validator/expression_rule.py b/lambdas/shared/src/common/validator/expression_rule.py
@@ -15,5 +15,7 @@ def expression_rule_per_field(expression_type: str) -> dict:
             return {"max_length": Constants.PERSON_NAME_ELEMENT_MAX_LENGTH}
         case "GENDER":
             return {"predefined_values": Constants.GENDERS}
+        case "DATETIME":
+            return {"strict_time_zone": True}
         case _:
             raise ValueError(f"Expression rule not found for type: {expression_type}")
diff --git a/lambdas/shared/tests/test_common/validator/test_expression_checker.py b/lambdas/shared/tests/test_common/validator/test_expression_checker.py
@@ -75,24 +75,67 @@ def test_list_valid_and_invalid(self):
         self.assertIsInstance(checker.validate_expression("LIST", "PERSON_NAME", "PERSON_FORENAME", []), ErrorReport)
         self.assertIsInstance(checker.validate_expression("LIST", "", "PERSON_FORENAME", "Alice"), ErrorReport)
 
-    # # DATE
-    # def test_date_valid_and_invalid(self):
-    #     checker = self.make_checker()
-    #     self.assertIsNone(checker.validate_expression("DATE", "", "contained|#:Patient|name|#:official|given|0", "2025-01-01"))
-    #     self.assertIsInstance(checker.validate_expression("DATE", "", "date_field", "2025-13-01"), ErrorReport)
-
+    # DATE
+    def test_date_valid_and_invalid(self):
+        checker = self.make_checker()
+        self.assertIsNone(checker.validate_expression("DATE", "", "contained|#:Patient|birthDate", "2025-01-01"))
+        self.assertIsNone(checker.validate_expression("DATE", "", "PERSON_DOB", "2025-01-01"))
+        self.assertIsInstance(
+            checker.validate_expression("DATE", "", "contained|#:Patient|birthDate", "2025-13-01"), ErrorReport
+        )
+        self.assertIsInstance(checker.validate_expression("DATE", "", "PERSON_DOB", "2025-02-30"), ErrorReport)
 
-#     # DATETIME
-#     def test_datetime_valid_and_invalid(self):
-#         checker = self.make_checker()
-#         # Full date only allowed
-#         self.assertIsNone(checker.validate_expression("DATETIME", "", "dt_field", "2025-01-01", 1))
-#         # Bad format should raise
-#         with self.assertRaises(Exception):
-#             checker.validate_expression("DATETIME", "", "dt_field", "2025-01-01T10:00:00Z", 1)
+    # DATETIME
+    def test_datetime_valid_and_invalid(self):
+        checker = self.make_checker()
+        # Full date only allowed
+        self.assertIsNone(
+            checker.validate_expression("DATETIME", "DATETIME", "occurrenceDateTime", "2025-01-01T05:00:00+00:00")
+        )
+        self.assertIsNone(
+            checker.validate_expression("DATETIME", "DATETIME", "DATE_AND_TIME", "2025-01-01T05:00:00+00:00")
+        )
+        # Bad format should raise
+        self.assertIsInstance(
+            checker.validate_expression("DATETIME", "", "occurrenceDateTime", "2026-01-01T10:00:00Z"), ErrorReport
+        )
+        self.assertIsInstance(
+            checker.validate_expression("DATETIME", "", "DATE_AND_TIME", "2026-01-01T10:00:00Z"), ErrorReport
+        )
 
 
 #     # BOOLEAN
+
+# # STRING with GENDER rule on real field
+# def test_gender_string_rule_valid_and_invalid(self):
+#     checker = self.make_checker()
+#     field_path = "contained|#:Patient|gender"
+#     # Valid genders per schema constants (male, female, other, unknown)
+#     self.assertIsNone(checker.validate_expression("STRING", "GENDER", field_path, "male"))
+#     self.assertIsNone(checker.validate_expression("STRING", "GENDER", field_path, "female"))
+#     # Invalid values should error
+#     self.assertIsInstance(
+#         checker.validate_expression("STRING", "GENDER", field_path, "M"),
+#         ErrorReport,
+#     )
+
+# # STRING with no rule for PERSON_POSTCODE on real field
+# def test_postcode_string_rule_valid_and_invalid(self):
+#     checker = self.make_checker()
+#     field_path = "contained|#:Patient|address|#:postalCode|postalCode"
+#     # With empty rule, generic string constraints apply: non-empty and no spaces
+#     self.assertIsNone(checker.validate_expression("STRING", "", field_path, "SW1A1AA"))
+#     # Real-world postcode with a space should fail as spaces are not allowed without a rule override
+#     field_path = "POST_CODE"
+#     self.assertIsInstance(
+#         checker.validate_expression("STRING", "", field_path, "AB12 3CD"),
+#         ErrorReport,
+#     )
+#     # Empty should also fail
+#     self.assertIsInstance(
+#         checker.validate_expression("STRING", "", field_path, ""),
+#         ErrorReport,
+#     )
 #     def test_boolean_valid_and_invalid(self):
 #         checker = self.make_checker()
 #         self.assertIsNone(checker.validate_expression("BOOLEAN", "", "bool_field", True, 1))
diff --git a/lambdas/shared/tests/test_common/validator/test_schemas/test_schema.json b/lambdas/shared/tests/test_common/validator/test_schemas/test_schema.json
diff --git a/lambdas/shared/tests/test_common/validator/testing_utils/constants.py b/lambdas/shared/tests/test_common/validator/testing_utils/constants.py
