Merge pull request #371 from NHSDigital/VED-73-recover-my-changes

recover reverted changes

Author: Akol125

backend/src/fhir_service.py

@@ -79,27 +79,18 @@ def get_immunization_by_identifier(
 
     def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: str) -> Optional[dict]:
         """
-        Get an Immunization by its ID. Return None if not found. If the patient doesn't have an NHS number,
+        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.
         """
         if not (imms_resp := self.immunization_repo.get_immunization_by_id(imms_id, imms_vax_type_perms)):
             return None
 
-        # Remove fields rom the imms resource which are not to be returned for read
-        imms_filtered_for_read = Filter.read(imms_resp.get("Resource", {}))
-
-        # Handle s-flag filtering, where applicable
-        if not (nhs_number := obtain_field_value(imms_filtered_for_read, FieldNames.patient_identifier_value)):
-            imms_filtered_for_read_and_s_flag = imms_filtered_for_read
-        else:
-            if patient := self.pds_service.get_patient_details(nhs_number):
-                imms_filtered_for_read_and_s_flag = handle_s_flag(imms_filtered_for_read, patient)
-            else:
-                raise UnhandledResponseError("unable to validate NHS number with downstream service")
+        # Returns the Immunisation full resource with no obfuscation
+        resource = imms_resp.get("Resource", {})
 
         return {
             "Version": imms_resp.get("Version", ""),
-            "Resource": Immunization.parse_obj(imms_filtered_for_read_and_s_flag),
+            "Resource": Immunization.parse_obj(resource),
         }
 
     def get_immunization_by_id_all(self, imms_id: str, imms: dict) -> Optional[dict]:

backend/src/filter.py

@@ -94,13 +94,6 @@ def add_use_to_identifier(imms: dict) -> dict:
 
 class Filter:
     """Functions for filtering a FHIR Immunization Resource"""
-
-    @staticmethod
-    def read(imms: dict) -> dict:
-        """Apply filtering for READ request"""
-        imms.pop("identifier")
-        return imms
-
     @staticmethod
     def search(imms: dict, patient_full_url: str, bundle_patient: dict = None) -> dict:
         """Apply filtering for an individual FHIR Immunization Resource as part of SEARCH request"""

…mmunization_event_filtered_for_read.json …covid19_immunization_event_for_read.jsonbackend/tests/sample_data/completed_covid19_immunization_event_filtered_for_read.json renamed to backend/tests/sample_data/completed_covid19_immunization_event_for_read.json backend/tests/sample_data/completed_covid19_immunization_event_filtered_for_read.json renamed to backend/tests/sample_data/completed_covid19_immunization_event_for_read.json

@@ -48,6 +48,11 @@
         ]
       }
     }
+  ],"identifier": [
+    {
+      "system": "https://supplierABC/identifiers/vacc",
+      "value": "ACME-vacc123456"
+    }
   ],
   "status": "completed",
   "vaccineCode": {

backend/tests/test_fhir_service.py

@@ -187,40 +187,24 @@ def test_immunization_not_found(self):
 
     def test_get_immunization_by_id_patient_not_restricted(self):
         """
-        Test that get_immunization_by_id returns a FHIR Immunization Resource which has been filtered for read,
-        but not for s-flag, when patient is not restricted
+        Test that get_immunization_by_id returns a FHIR Immunization Resource which has not been filtered for read,
+        but also dropped for s-flag, when patient is not restricted
         """
         imms_id = "non_restricted_id"
 
         immunization_data = load_json_data("completed_covid19_immunization_event.json")
         self.imms_repo.get_immunization_by_id.return_value = {"Resource": immunization_data}
         self.fhir_service.pds_service.get_patient_details.return_value = {"meta": {"security": [{"code": "U"}]}}
 
-        expected_imms = load_json_data("completed_covid19_immunization_event_filtered_for_read.json")
+        expected_imms = load_json_data("completed_covid19_immunization_event_for_read.json")
         expected_output = Immunization.parse_obj(expected_imms)
 
         # When
         actual_output = self.fhir_service.get_immunization_by_id(imms_id, "COVID19:read")
 
         # Then
         self.assertEqual(actual_output["Resource"], expected_output)
-
-    def test_get_immunization_by_id_patient_restricted(self):
-        """it should return a filtered Immunization when patient is restricted"""
-        imms_id = "restricted_id"
-        immunization_data = load_json_data("completed_covid19_immunization_event.json")
-        filtered_immunization = load_json_data("completed_covid19_immunization_event_filtered_for_s_flag_and_read.json")
-        self.imms_repo.get_immunization_by_id.return_value = {"Resource": immunization_data}
-        patient_data = {"meta": {"security": [{"code": "R"}]}}
-        self.fhir_service.pds_service.get_patient_details.return_value = patient_data
-
-        # When
-        resp_imms = self.fhir_service.get_immunization_by_id(imms_id, "COVID19:read")
-        act_res = resp_imms["Resource"]
-        filtered_immunization_res = Immunization.parse_obj(filtered_immunization)
-        # Then
-        self.assertEqual(act_res, filtered_immunization_res)
-
+   
     def test_pre_validation_failed(self):
         """it should throw exception if Immunization is not valid"""
         imms_id = "an-id"

backend/tests/test_filter.py

@@ -94,7 +94,6 @@ def test_replace_address_postal_codes(self):
         input_imms = load_json_data("completed_covid19_immunization_event.json")
         expected_output = deepcopy(input_imms)
         expected_output["contained"][1]["address"][0]["postalCode"] = "ZZ99 3CZ"
-
         self.assertEqual(replace_address_postal_codes(input_imms), expected_output)
 
     def test_replace_organization_values(self):
@@ -145,12 +144,6 @@ def test_replace_organization_values(self):
         del expected_output_data["performer"][1]["actor"]["identifier"]
         self.assertEqual(replace_organization_values(input_imms_data), expected_output_data)
 
-    def test_filter_read(self):
-        """Tests to ensure Filter.read appropriately filters a FHIR Immunization Resource"""
-        unfiltered_imms = deepcopy(self.covid_19_immunization_event)
-        expected_output = load_json_data("completed_covid19_immunization_event_filtered_for_read.json")
-        self.assertEqual(Filter.read(unfiltered_imms), expected_output)
-
     def test_filter_search(self):
         """Tests to ensure Filter.search appropriately filters a FHIR Immunization Resource"""
         bundle_patient = deepcopy(self.bundle_patient_resource)

e2e/test_get_immunization.py

@@ -1,4 +1,5 @@
 from decimal import Decimal
+import uuid
 
 from utils.base_test import ImmunizationBaseTest
 from utils.immunisation_api import parse_location
@@ -12,18 +13,26 @@ def test_get_imms(self):
         """it should get a FHIR Immunization resource"""
         for imms_api in self.imms_apis:
             with self.subTest(imms_api):
+                # Create one shared UUID per immunization (covid & rsv)
+                covid_uuid = str(uuid.uuid4())
+                rsv_uuid = str(uuid.uuid4())
                 # Given
                 immunizations = [
                     {
-                        "data": generate_imms_resource(),
+                        "data": generate_imms_resource(imms_identifier_value=covid_uuid),
                         "expected": generate_filtered_imms_resource(
-                            crud_operation_to_filter_for=EndpointOperationNames.READ)
+                            crud_operation_to_filter_for=EndpointOperationNames.READ,
+                            imms_identifier_value=covid_uuid)
                     },
                     {
-                        "data": generate_imms_resource(sample_data_file_name="completed_rsv_immunization_event"),
+                        "data": generate_imms_resource(
+                            sample_data_file_name="completed_rsv_immunization_event",
+                            vaccine_type=VaccineTypes.rsv,
+                            imms_identifier_value=rsv_uuid),
                         "expected": generate_filtered_imms_resource(
                             crud_operation_to_filter_for=EndpointOperationNames.READ,
-                            vaccine_type=VaccineTypes.rsv
+                            vaccine_type=VaccineTypes.rsv,
+                            imms_identifier_value=rsv_uuid
                         )
                     }
                 ]
@@ -40,7 +49,6 @@ def test_get_imms(self):
                 # When - Retrieve and validate each immunization by ID
                 for immunization in immunizations:
                     response = imms_api.get_immunization_by_id(immunization["id"])
-
                     # Then
                     self.assertEqual(response.status_code, 200)
                     self.assertEqual(response.json()["id"], immunization["id"])

e2e/test_s_flag_immunization.py

@@ -1,5 +1,6 @@
 from typing import List
 from decimal import Decimal
+import uuid
 
 from utils.base_test import ImmunizationBaseTest
 from utils.constants import valid_nhs_number1, valid_nhs_number_with_s_flag
@@ -11,43 +12,37 @@
 class SFlagBaseTest(ImmunizationBaseTest):
     """Parent class with helper for storing immunisation events in the IEDS"""
 
-    def store_imms(self, imms_api: ImmunisationApi, patient_is_restricted: bool) -> dict:
+    def store_imms(self, imms_api: ImmunisationApi, patient_is_restricted: bool, imms_identifier_value=None) -> dict:
         """
         Store an immunisation event in the IEDS using valid_nhs_number_with_s_flag if patient is restricted, or
         valid_nhs_number_1 otherwise
         """
         nhs_number = valid_nhs_number_with_s_flag if patient_is_restricted else valid_nhs_number1
-        imms = generate_imms_resource(nhs_number=nhs_number, vaccine_type=VaccineTypes.covid_19)
+        imms = generate_imms_resource(
+            nhs_number=nhs_number, vaccine_type=VaccineTypes.covid_19,
+            imms_identifier_value=imms_identifier_value)
         return self.create_immunization_resource(imms_api, imms)
 
 
 class TestGetSFlagImmunization(SFlagBaseTest):
-    """Test that sensitive data is filtered out for a READ if and only if the patient is s-flagged"""
-
-    def test_get_s_flagged_imms(self):
-        """Test that sensitive data is filtered out for a READ when the patient is s-flagged"""
-        for imms_api in self.imms_apis:
-            with self.subTest(imms_api):
-                imms_id = self.store_imms(imms_api, patient_is_restricted=True)
-                read_imms = imms_api.get_immunization_by_id(imms_id).json(parse_float=Decimal)
-                expected_response = generate_filtered_imms_resource(
-                    crud_operation_to_filter_for="READ",
-                    filter_for_s_flag=True,
-                    nhs_number=valid_nhs_number_with_s_flag,
-                )
-                expected_response["id"] = read_imms["id"]
-                self.assertEqual(read_imms, expected_response)
-
     def test_get_not_s_flagged_imms(self):
         """Test that sensitive data is not filtered out for a READ when the patient is not s-flagged"""
         for imms_api in self.imms_apis:
             with self.subTest(imms_api):
-                imms = self.store_imms(imms_api, patient_is_restricted=False)
+
+                shared_uuid = str(uuid.uuid4())
+
+                imms = self.store_imms(
+                    imms_api,
+                    patient_is_restricted=False,
+                    imms_identifier_value=shared_uuid)
+
                 read_imms = imms_api.get_immunization_by_id(imms).json(parse_float=Decimal)
                 expected_response = generate_filtered_imms_resource(
                     crud_operation_to_filter_for="READ",
                     filter_for_s_flag=False,
                     nhs_number=valid_nhs_number1,
+                    imms_identifier_value=shared_uuid
                 )
                 expected_response["id"] = read_imms["id"]
                 self.assertEqual(read_imms, expected_response)

e2e/test_search_immunization.py

@@ -162,13 +162,17 @@ def test_search_immunization_parameter_smoke_tests(self):
         time_1 = "2024-01-30T13:28:17.271+00:00"
         time_2 = "2024-02-01T13:28:17.271+00:00"
         stored_records = [
-            generate_imms_resource(valid_nhs_number1, VaccineTypes.mmr),
-            generate_imms_resource(valid_nhs_number1, VaccineTypes.flu),
-            generate_imms_resource(valid_nhs_number1, VaccineTypes.covid_19),
-            generate_imms_resource(valid_nhs_number1, VaccineTypes.covid_19, time_1),
-            generate_imms_resource(valid_nhs_number1, VaccineTypes.covid_19, time_2),
-            generate_imms_resource(valid_nhs_number2, VaccineTypes.flu),
-            generate_imms_resource(valid_nhs_number2, VaccineTypes.covid_19),
+            generate_imms_resource(valid_nhs_number1, VaccineTypes.mmr, imms_identifier_value=str(uuid.uuid4())),
+            generate_imms_resource(valid_nhs_number1, VaccineTypes.flu, imms_identifier_value=str(uuid.uuid4())),
+            generate_imms_resource(valid_nhs_number1, VaccineTypes.covid_19, imms_identifier_value=str(uuid.uuid4())),
+            generate_imms_resource(valid_nhs_number1, VaccineTypes.covid_19,
+                                   occurrence_date_time=time_1,
+                                   imms_identifier_value=str(uuid.uuid4())),
+            generate_imms_resource(valid_nhs_number1, VaccineTypes.covid_19,
+                                   occurrence_date_time=time_2,
+                                   imms_identifier_value=str(uuid.uuid4())),
+            generate_imms_resource(valid_nhs_number2, VaccineTypes.flu, imms_identifier_value=str(uuid.uuid4())),
+            generate_imms_resource(valid_nhs_number2, VaccineTypes.covid_19, imms_identifier_value=str(uuid.uuid4())),
         ]
 
         created_resource_ids = list(self.store_records(*stored_records))
@@ -308,6 +312,13 @@ class SearchTestParams(NamedTuple):
 
                 result_ids = [result["resource"]["id"] for result in results["entry"]]
                 created_and_returned_ids = list(set(result_ids) & set(created_resource_ids))
+                print("\n Search Test Debug Info:")
+                print("Search method:", search.method)
+                print("Search query string:", search.query_string)
+                print("Expected indexes:", search.expected_indexes)
+                print("Expected IDs:", [created_resource_ids[i] for i in search.expected_indexes])
+                print("Actual returned IDs:", result_ids)
+                print("Matched IDs:", created_and_returned_ids)
                 assert len(created_and_returned_ids) == len(search.expected_indexes)
                 for expected_index in search.expected_indexes:
                     assert created_resource_ids[expected_index] in result_ids

e2e/utils/resource.py

@@ -22,20 +22,21 @@ def load_example(path: str) -> dict:
 def generate_imms_resource(
     nhs_number=valid_nhs_number1,
     vaccine_type=VaccineTypes.covid_19,
+    imms_identifier_value: str = None,
     occurrence_date_time: str = None,
     sample_data_file_name: str = "completed_[vaccine_type]_immunization_event",
 ) -> dict:
     """
     Creates a FHIR Immunization Resource dictionary, which includes an id, using the sample data for the given
     vaccine type as a base, and updates the id, nhs_number and occurrence_date_time as required.
-    The unique_identifier is also updated to ensure uniqueness.
+    The unique_identifier is also updated to ensure uniqueness...
     """
     # Load the data
     sample_data_file_name = sample_data_file_name.replace("[vaccine_type]", vaccine_type.lower())
     imms = deepcopy(load_example(f"Immunization/{sample_data_file_name}.json"))
 
-    # Amend data fields as appropriate
-    imms["identifier"][0]["value"] = str(uuid.uuid4())
+    # Apply identifier directly
+    imms["identifier"][0]["value"] = imms_identifier_value or str(uuid.uuid4())
 
     if nhs_number is not None:
         imms["contained"][1]["identifier"][0]["value"] = nhs_number
@@ -49,8 +50,8 @@ def generate_imms_resource(
 def generate_filtered_imms_resource(
     crud_operation_to_filter_for: Literal["READ", "SEARCH", ""] = "",
     filter_for_s_flag: bool = False,
-    imms_identifier_value: str = None,
     nhs_number=valid_nhs_number1,
+    imms_identifier_value: str = None,
     vaccine_type=VaccineTypes.covid_19,
     occurrence_date_time: str = None,
 ) -> dict:
@@ -72,10 +73,8 @@ def generate_filtered_imms_resource(
         + f"_filtered_for_{crud_operation_to_filter_for.lower()}{s_flag_string}"
     )
     imms = deepcopy(load_example(f"{file_name}.json"))
-
-    # Amend values as required
-    if imms_identifier_value:
-        imms["identifier"][0]["value"] = imms_identifier_value
+    # Apply identifier directly
+    imms["identifier"][0]["value"] = imms_identifier_value or str(uuid.uuid4())
 
     # Note that NHS number is found in a different place on a search return
     if crud_operation_to_filter_for == "SEARCH":

specification/components/examples/Immunization/completed_covid19_immunization_event.json

@@ -52,7 +52,7 @@
   "identifier": [
     {
       "system": "https://supplierABC/identifiers/vacc",
-      "value": "ACME-vacc123456"
+      "value": "ACME-vacc123456789"
     }
   ],
   "status": "completed",
@@ -149,4 +149,4 @@
       "doseNumberPositiveInt": 1
     }
   ]
-}
+}