Merge branch 'master' into dependabot/pip/ack_backend/pip-minor-patch-54a4d85f02

Author: dlzhry2nhs

backend/src/fhir_repository.py

@@ -372,7 +372,12 @@ def find_immunizations(self, patient_identifier: str, vaccine_types: set):
             items = [x for x in raw_items if x["PatientSK"].split("#")[0] in vaccine_types]
 
             # Return a list of the FHIR immunization resource JSON items
-            final_resources = [json.loads(item["Resource"]) for item in items]
+            final_resources = [{
+                **json.loads(item["Resource"]),
+                "meta": {"versionId": int(item.get("Version", 1))}
+                }
+                for item in items
+                ]
 
             return final_resources
         else:

backend/src/models/utils/generic_utils.py

@@ -1,8 +1,15 @@
 """Generic utilities"""
 
 import datetime
-
-from typing import Literal, Union, Optional
+import json
+from typing import Literal, Union, Optional, Dict, Any
+from fhir.resources.R4B.bundle import (
+    Bundle as FhirBundle,
+    BundleEntry,
+    BundleLink,
+    BundleEntrySearch,
+)
+from fhir.resources.R4B.immunization import Immunization
 from models.constants import Constants
 import urllib.parse
 import base64
@@ -123,7 +130,6 @@ def create_diagnostics():
     exp_error = {"diagnostics": diagnostics}
     return exp_error
 
-
 def create_diagnostics_error(value):
     if value == "Both":
         diagnostics = (
@@ -134,46 +140,49 @@ def create_diagnostics_error(value):
     exp_error = {"diagnostics": diagnostics}
     return exp_error
 
-
-def form_json(response, _element, identifier, baseurl):
-    self_url = f"{baseurl}?identifier={identifier}" + (f"&_elements={_element}" if _element else "")
-    json = {
-            "resourceType": "Bundle",
-            "type": "searchset",
-            "link": [
-                {"relation": "self", "url": self_url}
-            ]
+def make_empty_bundle(self_url: str) -> Dict[str, Any]:
+    return {
+        "resourceType": "Bundle",
+        "type": "searchset",
+        "link": [{"relation": "self", "url": self_url}],
+        "entry": [],
+        "total": 0, 
     }
+
+def form_json(response, _elements, identifier, baseurl):
+    self_url = f"{baseurl}?identifier={identifier}" + (f"&_elements={_elements}" if _elements else "")
+
     if not response:
-        json["entry"] = []
-        json["total"] = 0
-        return json
+        return make_empty_bundle(self_url)
 
-    # Full Immunization payload to be returned if only the identifier parameter was provided
-    if identifier and not _element:
-        resource = response["resource"]
+    meta = {"versionId": response["version"]} if "version" in response else {}
 
-    elif identifier and _element:
-        element = {e.strip().lower() for e in _element.split(",") if e.strip()}
+    # Full Immunization payload to be returned if only the identifier parameter was provided and truncated when _elements is used
+    if _elements:
+        elements = {e.strip().lower() for e in _elements.split(",") if e.strip()}
         resource = {"resourceType": "Immunization"}
+        if "id" in elements: resource["id"] = response["id"]
+        if "meta" in elements and meta: resource["meta"] = meta
 
-        # Add 'id' if specified
-        if "id" in element:
-            resource["id"] = response["id"]
+    else:
+        resource = response["resource"]
+        resource["meta"] = meta
 
-        # Add 'meta' if specified
-        if "meta" in element:
-            resource["id"] = response["id"]
-            resource["meta"] = {"versionId": response["version"]}
+    entry = BundleEntry(fullUrl=f"{baseurl}/{response['id']}",
+        resource=Immunization.construct(**resource) if _elements else Immunization.parse_obj(resource),
+        search=BundleEntrySearch.construct(mode="match") if not _elements else None,
+    )
 
-    json["entry"] = [{
-        "fullUrl": f"https://api.service.nhs.uk/immunisation-fhir-api/Immunization/{response['id']}",
-        "resource": resource,
-            }
-        ]
-    json["total"] = 1
-    return json
+    fhir_bundle = FhirBundle(
+        resourceType="Bundle", type="searchset", 
+        link = [BundleLink(relation="self", url=self_url)], 
+        entry=[entry], 
+        total=1)
 
+    # Reassigned total to ensure it appears last in the response to match expected output
+    data = json.loads(fhir_bundle.json(by_alias=True))
+    data["total"] = data.pop("total")
+    return data
 
 def check_keys_in_sources(event, not_required_keys):
     # Decode and parse the body, assuming it is JSON and base64-encoded

backend/tests/sample_data/completed_covid19_immunization_event.json

@@ -51,6 +51,7 @@
   ],
   "identifier": [
     {
+      "use": "official",
       "system": "https://supplierABC/identifiers/vacc",
       "value": "ACME-vacc123456"
     }

backend/tests/sample_data/completed_covid19_immunization_event_for_read.json

@@ -50,6 +50,7 @@
     }
   ],"identifier": [
     {
+      "use": "official",
       "system": "https://supplierABC/identifiers/vacc",
       "value": "ACME-vacc123456"
     }

backend/tests/test_fhir_repository.py

@@ -581,16 +581,18 @@ def test_exclude_deleted(self):
 
     def test_map_results_to_immunizations(self):
         """it should map Resource list into a list of Immunizations"""
-        imms1 = {"id": 1}
-        imms2 = {"id": 2}
+        imms1 = {"id": 1, "meta": {"versionId": 1}}
+        imms2 = {"id": 2, "meta": {"versionId": 1}}
         items = [
             {
                 "Resource": json.dumps(imms1),
                 "PatientSK": "COVID19#some_other_text",
+                "Version": "1"
             },
             {
                 "Resource": json.dumps(imms2),
                 "PatientSK": "COVID19#some_other_text",
+                "Version": "1"
             },
         ]
 

backend/tests/test_fhir_service.py

@@ -346,25 +346,34 @@ def tearDown(self):
 
     def test_get_immunization_by_identifier(self):
         """it should find an Immunization by id"""
-        imms = "an-id#an-id"
+        imms_id = "an-id#an-id"
         identifier = "test"
         element = "id,mEta,DDD"
+
         mock_resource = create_covid_19_immunization_dict(identifier)
         self.authoriser.authorise.return_value = True
         self.imms_repo.get_immunization_by_identifier.return_value = {
             "resource": mock_resource,
-            "id": identifier,
+            "id": imms_id,
             "version": 1
         }, "covid19"
 
         # When
-        service_resp = self.fhir_service.get_immunization_by_identifier(imms, self.MOCK_SUPPLIER_NAME, identifier,
+        service_resp = self.fhir_service.get_immunization_by_identifier(imms_id, self.MOCK_SUPPLIER_NAME, identifier,
                                                                         element)
 
         # Then
-        self.imms_repo.get_immunization_by_identifier.assert_called_once_with(imms)
+        self.imms_repo.get_immunization_by_identifier.assert_called_once_with(imms_id)
         self.authoriser.authorise.assert_called_once_with(self.MOCK_SUPPLIER_NAME, ApiOperationCode.SEARCH, {"covid19"})
         self.assertEqual(service_resp["resourceType"], "Bundle")
+        self.assertEqual(service_resp.get("type"), "searchset")
+        self.assertIn("entry", service_resp)
+        self.assertEqual(len(service_resp["entry"]), 1)
+        self.assertIn("total", service_resp)
+        self.assertEqual(service_resp["total"], 1)
+        res = service_resp["entry"][0]["resource"]
+        self.assertEqual(res["resourceType"], "Immunization")
+        self.assertEqual(res["id"], imms_id)
 
     def test_get_immunization_by_identifier_raises_error_when_not_authorised(self):
         """it should find an Immunization by id"""
@@ -382,6 +391,7 @@ def test_get_immunization_by_identifier_raises_error_when_not_authorised(self):
         self.imms_repo.get_immunization_by_identifier.assert_called_once_with(imms)
         self.authoriser.authorise.assert_called_once_with(self.MOCK_SUPPLIER_NAME, ApiOperationCode.SEARCH, {"covid19"})
 
+
     def test_immunization_not_found(self):
         """it should return None if Immunization doesn't exist"""
         imms_id = "none"

backend/tests/utils/immunization_utils.py

@@ -27,6 +27,7 @@ def create_covid_19_immunization_dict(
 
     return immunization_json
 
+
 def create_covid_19_immunization_dict_no_id(
     nhs_number=VALID_NHS_NUMBER, occurrence_date_time="2021-02-07T13:28:17.271+00:00"
 ):

backend/tests/utils/test_generic_utils.py

@@ -1,20 +1,22 @@
 """Generic utils for tests"""
 
-
 import unittest
 from src.models.utils.generic_utils import form_json
+from tests.utils.generic_utils import load_json_data
 
 
 class TestFormJson(unittest.TestCase):
     def setUp(self):
         self.baseurl = "https://api.service.nhs.uk/immunisation-fhir-api/Immunization"
         self.identifier = "https://supplierABC/identifiers/vacc|f10b59b3-fc73-4616-99c9-9e882ab31184"
         self.response = {
-            "resource": {"resourceType": "Immunization", "id": "f10b59b3-fc73-4616-99c9-9e882ab31184", "status": "completed"},
+            "resource": load_json_data("completed_covid19_immunization_event.json"),
             "id": "f10b59b3-fc73-4616-99c9-9e882ab31184",
-            "version": 2,
+            "version": "2",
         }
 
+        self.maxDiff = None
+
     def test_no_response(self):
         out = form_json(None, None, self.identifier, self.baseurl)
         self.assertEqual(out["resourceType"], "Bundle")
@@ -45,7 +47,6 @@ def test_identifier_with_meta_element_truncates_to_meta(self):
         self.assertEqual(out["total"], 1)
         self.assertEqual(out["link"][0]["url"], f"{self.baseurl}?identifier={self.identifier}&_elements=meta")
         self.assertEqual(res["resourceType"], "Immunization")
-        self.assertEqual(res["id"], self.response["id"])
         self.assertIn("meta", res)
         self.assertEqual(res["meta"]["versionId"], self.response["version"])
 

e2e/test_search_immunization.py

@@ -88,6 +88,7 @@ def test_search_backwards_compatible(self):
                     vaccine_type=VaccineTypes.covid_19,
                 )
                 expected_imms_resource["id"] = imms_id
+                expected_imms_resource["meta"] = {"versionId": "1"}
 
                 # When
                 response = imms_api.search_immunizations(valid_nhs_number1, VaccineTypes.covid_19)