Merge branch 'master' into VED-167-Audit-table-Ack-Lambda-Tests

Author: nhsdevws

backend/Makefile

@@ -8,4 +8,10 @@ package: build
 test:
 	@PYTHONPATH=src:tests python -m unittest
 
+coverage-run:
+	@PYTHONPATH=src:tests coverage run -m unittest discover
+
+coverage-report:
+	coverage report -m 
+
 .PHONY: build package test

backend/src/fhir_controller.py

@@ -482,16 +482,15 @@ def _validate_id(self, _id: str) -> Optional[dict]:
         else:
             return None
 
-    def _validate_identifier_system(self, _id: str, _element: str) -> Optional[dict]:
-
+    def _validate_identifier_system(self, _id: str, _elements: str) -> Optional[dict]:
         if not _id:
             return create_operation_outcome(
                 resource_id=str(uuid.uuid4()),
                 severity=Severity.error,
                 code=Code.invalid,
                 diagnostics=(
-                    "Search parameter immunization.identifier must have one value and must be in the format of "
-                    '"immunization.identifier.system|immunization.identifier.value" '
+                    "Search parameter identifier must have one value and must be in the format of "
+                    '"identifier.system|identifier.value" '
                     'e.g. "http://xyz.org/vaccs|2345-gh3s-r53h7-12ny"'
                 ),
             )
@@ -501,27 +500,23 @@ def _validate_identifier_system(self, _id: str, _element: str) -> Optional[dict]
                 severity=Severity.error,
                 code=Code.invalid,
                 diagnostics=(
-                    "Search parameter immunization.identifier must be in the format of "
-                    '"immunization.identifier.system|immunization.identifier.value" '
+                    "Search parameter identifier must be in the format of "
+                    '"identifier.system|identifier.value" '
                     'e.g. "http://xyz.org/vaccs|2345-gh3s-r53h7-12ny"'
                 ),
             )
-        if not _element:
-            return create_operation_outcome(
-                resource_id=str(uuid.uuid4()),
-                severity=Severity.error,
-                code=Code.invalid,
-                diagnostics="_element must be one or more of the following: id,meta",
-            )
-        element_lower = _element.lower()
-        result = element_lower.split(",")
-        is_present = all(key in ["id", "meta"] for key in result)
-        if not is_present:
+        
+        if not _elements:
+            return None
+
+        requested_elements = {e.strip().lower() for e in _elements.split(",") if e.strip()}
+        requested_elements_valid = requested_elements.issubset({"id", "meta"})
+        if _elements and not requested_elements_valid:
             return create_operation_outcome(
                 resource_id=str(uuid.uuid4()),
                 severity=Severity.error,
                 code=Code.invalid,
-                diagnostics="_element must be one or more of the following: id,meta",
+                diagnostics="_elements must be one or more of the following: id,meta",
             )
 
     def _create_bad_request(self, message):
@@ -549,47 +544,62 @@ def authorize_request(self, aws_event: dict) -> Optional[dict]:
             return self.create_response(500, id_error)
 
     def fetch_identifier_system_and_element(self, event: dict):
+        """
+        Extracts `identifier` and `_elements` from an incoming FHIR search request.
+
+        FHIR search supports two input formats:
+        1. GET search: parameters appear in the query string (e.g. ?identifier=abc123&_elements=id,meta)
+        2. POST search: parameters appear in the request body, form-encoded (e.g. identifier=abc123&_elements=id,meta)
+
+        This function handles both cases, returning:
+        - The extracted identifier value
+        - The extracted _elements value
+        - Any validation check result for disallowed keys
+        - Booleans indicating whether identifier/_elements were present
+        """
+
         query_params = event.get("queryStringParameters", {})
         body = event["body"]
         not_required_keys = ["-date.from", "-date.to", "-immunization.target", "_include", "patient.identifier"]
+
+        # Get Search Query Parameters
         if query_params and not body:
-            # Check for the presence of 'immunization.identifier' and '_element'
-            query_string_has_immunization_identifier = "immunization.identifier" in event.get(
-                "queryStringParameters", {}
-            )
-            query_string_has_element = "_element" in event.get("queryStringParameters", {})
-            immunization_identifier = query_params.get("immunization.identifier", "")
-            element = query_params.get("_element", "")
+            query_string_has_immunization_identifier = "identifier" in query_params
+            query_string_has_element = "_elements" in query_params
+            identifier = query_params.get("identifier", "")
+            element = query_params.get("_elements", "")
             query_check = check_keys_in_sources(event, not_required_keys)
 
             return (
-                immunization_identifier,
+                identifier,
                 element,
                 query_check,
                 query_string_has_immunization_identifier,
                 query_string_has_element,
             )
+        
+        # Post Search Identifier by body form
         if body and not query_params:
             decoded_body = base64.b64decode(body).decode("utf-8")
             parsed_body = urllib.parse.parse_qs(decoded_body)
-            # Attempt to extract 'immunization.identifier' and '_element'
-            converted_identifer = ""
-            converted_element = ""
-            immunization_identifier = parsed_body.get("immunization.identifier", "")
-            if immunization_identifier:
-                converted_identifer = "".join(immunization_identifier)
-            _element = parsed_body.get("_element", "")
-            if _element:
-                converted_element = "".join(_element)
-            body_has_immunization_identifier = "immunization.identifier" in parsed_body
-            body_has_immunization_element = "_element" in parsed_body
+            # Attempt to extract 'identifier' and '_elements'
+            converted_identifier = ""
+            converted_elements = ""
+            identifier = parsed_body.get("identifier", "")
+            if identifier:
+                converted_identifier = "".join(identifier)
+            _elements = parsed_body.get("_elements", "")
+            if _elements:
+                converted_elements = "".join(_elements)
+            body_has_identifier = "identifier" in parsed_body
+            body_has_immunization_elements = "_elements" in parsed_body
             body_check = check_keys_in_sources(event, not_required_keys)
             return (
-                converted_identifer,
-                converted_element,
+                converted_identifier,
+                converted_elements,
                 body_check,
-                body_has_immunization_identifier,
-                body_has_immunization_element,
+                body_has_identifier,
+                body_has_immunization_elements,
             )
 
     def create_response_for_identifier(self, not_required, has_identifier, has_element):
@@ -598,16 +608,7 @@ def create_response_for_identifier(self, not_required, has_identifier, has_eleme
                 resource_id=str(uuid.uuid4()),
                 severity=Severity.error,
                 code=Code.server_error,
-                diagnostics="Search parameter should have either immunization.identifier or patient.identifier",
-            )
-            return self.create_response(400, error)
-
-        if "patient.identifier" not in not_required and not_required and has_identifier:
-            error = create_operation_outcome(
-                resource_id=str(uuid.uuid4()),
-                severity=Severity.error,
-                code=Code.server_error,
-                diagnostics="Search parameter immunization.identifier must have the following parameter: _element",
+                diagnostics="Search parameter should have either identifier or patient.identifier",
             )
             return self.create_response(400, error)
 
@@ -616,7 +617,7 @@ def create_response_for_identifier(self, not_required, has_identifier, has_eleme
                 resource_id=str(uuid.uuid4()),
                 severity=Severity.error,
                 code=Code.server_error,
-                diagnostics="Search parameter _element must have  the following parameter: immunization.identifier",
+                diagnostics="Search parameter _elements must have the following parameter: identifier",
             )
             return self.create_response(400, error)
 

backend/src/fhir_repository.py

@@ -95,14 +95,16 @@ def get_immunization_by_identifier(
         )
         if "Items" in response and len(response["Items"]) > 0:
             item = response["Items"][0]
-            resp = dict()
             vaccine_type = self._vaccine_type(item["PatientSK"])
             if not validate_permissions(imms_vax_type_perms,ApiOperationCode.SEARCH, [vaccine_type]):
                 raise UnauthorizedVaxError()
             resource = json.loads(item["Resource"])
-            resp["id"] = resource.get("id")
-            resp["version"] = int(response["Items"][0]["Version"])
-            return resp
+            version = int(response["Items"][0]["Version"])
+            return {
+                "resource": resource,
+                "id": resource.get("id"),
+                "version": version
+            }
         else:
             return None
 

backend/src/fhir_service.py

@@ -62,7 +62,7 @@ def get_immunization_by_identifier(
     ) -> Optional[dict]:
         """
         Get an Immunization by its ID. Return None if not found. If the patient doesn't have an NHS number,
-        return the Immunization without calling PDS or checking S flag.
+        return the Immunization.
         """
         imms_resp = self.immunization_repo.get_immunization_by_identifier(
             identifier_pk, imms_vax_type_perms
@@ -79,7 +79,7 @@ def get_immunization_by_identifier(
     def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: list[str]) -> Optional[dict]:
         """
         Get an Immunization by its ID. Return None if it is not found. If the patient doesn't have an NHS number,
-        return the Immunization without calling PDS or checking S flag.
+        return the Immunization.
         """
         if not (imms_resp := self.immunization_repo.get_immunization_by_id(imms_id, imms_vax_type_perms)):
             return None
@@ -95,7 +95,7 @@ def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: list[str]) -
     def get_immunization_by_id_all(self, imms_id: str, imms: dict) -> Optional[dict]:
         """
         Get an Immunization by its ID. Return None if not found. If the patient doesn't have an NHS number,
-        return the Immunization without calling PDS or checking S flag.
+        return the Immunization.
         """
         imms["id"] = imms_id
         try:

backend/src/models/utils/generic_utils.py

@@ -136,43 +136,42 @@ def create_diagnostics_error(value):
 
 
 def form_json(response, _element, identifier, baseurl):
-    # Elements to include, based on the '_element' parameter
-    if not response:
-        json = {
+    self_url = f"{baseurl}?identifier={identifier}" + (f"&_elements={_element}" if _element else "")
+    json = {
             "resourceType": "Bundle",
             "type": "searchset",
             "link": [
-                {"relation": "self", "url": f"{baseurl}?immunization.identifier={identifier}&_elements={_element}"}
-            ],
-            "entry": [],
-            "total": 0,
-        }
+                {"relation": "self", "url": self_url}
+            ]
+    }
+    if not response:
+        json["entry"] = []
+        json["total"] = 0
         return json
 
-    # Basic structure for the JSON output
-    json = {
-        "resourceType": "Bundle",
-        "type": "searchset",
-        "link": [{"relation": "self", "url": f"{baseurl}?immunization.identifier={identifier}&_elements={_element}"}],
-        "entry": [
-            {
-                "fullUrl": f"https://api.service.nhs.uk/immunisation-fhir-api/Immunization/{response['id']}",
-                "resource": {"resourceType": "Immunization"},
-            }
-        ],
-        "total": 1,
-    }
-    __elements = _element.lower()
-    element = __elements.split(",")
-    # Add 'id' if specified
-    if "id" in element:
-        json["entry"][0]["resource"]["id"] = response["id"]
+    # Full Immunization payload to be returned if only the identifier parameter was provided
+    if identifier and not _element:
+        resource = response["resource"]
+
+    elif identifier and _element:
+        element = {e.strip().lower() for e in _element.split(",") if e.strip()}
+        resource = {"resourceType": "Immunization"}
 
-    # Add 'meta' if specified
-    if "meta" in element:
-        json["entry"][0]["resource"]["id"] = response["id"]
-        json["entry"][0]["resource"]["meta"] = {"versionId": response["version"]}
+        # Add 'id' if specified
+        if "id" in element:
+            resource["id"] = response["id"]
 
+        # Add 'meta' if specified
+        if "meta" in element:
+            resource["id"] = response["id"]
+            resource["meta"] = {"versionId": response["version"]}
+
+    json["entry"] = [{
+        "fullUrl": f"https://api.service.nhs.uk/immunisation-fhir-api/Immunization/{response['id']}",
+        "resource": resource,
+            }
+        ]
+    json["total"] = 1
     return json
 
 

backend/src/models/utils/pre_validator_utils.py

@@ -142,7 +142,7 @@ def for_date_time(field_value: str, field_location: str, strict_timezone: bool =
                 continue
 
         raise ValueError(error_message)
- 
+
     @staticmethod
     def for_snomed_code(field_value: str, field_location: str):
         """

backend/src/search_imms_handler.py

@@ -32,19 +32,19 @@ def search_imms(event: events.APIGatewayProxyEventV1, controller: FhirController
         body_has_immunization_element = False
         if not (query_params == None and body == None):
             if query_params:
-                query_string_has_immunization_identifier = "immunization.identifier" in event.get(
+                query_string_has_immunization_identifier = "identifier" in event.get(
                     "queryStringParameters", {}
                 )
-                query_string_has_element = "_element" in event.get("queryStringParameters", {})
+                query_string_has_element = "_elements" in event.get("queryStringParameters", {})
             # Decode body from base64
             if body:
                 decoded_body = base64.b64decode(body).decode("utf-8")
                 # Parse the URL encoded body
                 parsed_body = urllib.parse.parse_qs(decoded_body)
 
-                # Check for 'immunization.identifier' in body
-                body_has_immunization_identifier = "immunization.identifier" in parsed_body
-                body_has_immunization_element = "_element" in parsed_body
+                # Check for 'identifier' in body
+                body_has_immunization_identifier = "identifier" in parsed_body
+                body_has_immunization_element = "_elements" in parsed_body
             if (
                 query_string_has_immunization_identifier
                 or body_has_immunization_identifier
@@ -97,13 +97,13 @@ def search_imms(event: events.APIGatewayProxyEventV1, controller: FhirController
     parser.add_argument("--date.from", type=str, required=False, dest="date_from")
     parser.add_argument("--date.to", type=str, required=False, dest="date_to")
     parser.add_argument(
-        "--immunization.identifier",
+        "--identifier",
         help="Identifier of System",
         type=str,
         required=False,
-        dest="immunization_identifier",
+        dest="identifier",
     )
-    parser.add_argument("--element", help="Identifier of System", type=str, required=False, dest="_element")
+    parser.add_argument("--elements", help="Identifier of System", type=str, required=False, dest="_elements")
     args = parser.parse_args()
 
     event: events.APIGatewayProxyEventV1 = {
@@ -113,8 +113,8 @@ def search_imms(event: events.APIGatewayProxyEventV1, controller: FhirController
             "-date.from": [args.date_from] if args.date_from else [],
             "-date.to": [args.date_to] if args.date_to else [],
             "_include": ["Immunization:patient"],
-            "immunization_identifier": [args.immunization_identifier] if args.immunization_identifier else [],
-            "_element": [args._element] if args._element else [],
+            "identifier": [args.immunization_identifier] if args.immunization_identifier else [],
+            "_elements": [args._element] if args._element else [],
         },
         "httpMethod": "POST",
         "headers": {