test suite complete

Author: Akol125

backend/src/fhir_repository.py

@@ -85,7 +85,7 @@ def __init__(self, table: Table):
         self.table = table
 
     def get_immunization_by_identifier(
-        self, identifier_pk: str, imms_vax_type_perms: str, is_imms_batch_app
+        self, identifier_pk: str, imms_vax_type_perms: str
     ) -> Optional[dict]:
         response = self.table.query(
             IndexName="IdentifierGSI", KeyConditionExpression=Key("IdentifierPK").eq(identifier_pk)
@@ -94,10 +94,9 @@ def get_immunization_by_identifier(
             item = response["Items"][0]
             resp = dict()
             vaccine_type = self._vaccine_type(item["PatientSK"])
-            if not is_imms_batch_app:
-                vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
-                vax_type_perm = self._vaccine_permission(vaccine_type, "search")
-                self._check_permission(vax_type_perm, vax_type_perms)
+            vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
+            vax_type_perm = self._vaccine_permission(vaccine_type, "search")
+            self._check_permission(vax_type_perm, vax_type_perms)
             resource = json.loads(item["Resource"])
             resp["id"] = resource.get("id")
             resp["version"] = int(response["Items"][0]["Version"])
@@ -166,15 +165,14 @@ def get_immunization_by_id_all(self, imms_id: str, imms: dict) -> Optional[dict]
             return None
 
     def create_immunization(
-        self, immunization: dict, patient: any, imms_vax_type_perms, supplier_system, is_imms_batch_app
+        self, immunization: dict, patient: any, imms_vax_type_perms, supplier_system
     ) -> dict:
         new_id = str(uuid.uuid4())
         immunization["id"] = new_id
         attr = RecordAttributes(immunization, patient)
-        if not is_imms_batch_app:
-            vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
-            vax_type_perm = self._vaccine_permission(attr.vaccine_type, "create")
-            self._check_permission(vax_type_perm, vax_type_perms)
+        vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
+        vax_type_perm = self._vaccine_permission(attr.vaccine_type, "create")
+        self._check_permission(vax_type_perm, vax_type_perms)
         query_response = _query_identifier(self.table, "IdentifierGSI", "IdentifierPK", attr.identifier)
 
         if query_response is not None:
@@ -206,10 +204,9 @@ def update_immunization(
         existing_resource_version: int,
         imms_vax_type_perms: str,
         supplier_system: str,
-        is_imms_batch_app: bool,
     ) -> dict:
         attr = RecordAttributes(immunization, patient)
-        self._handle_permissions(is_imms_batch_app, imms_vax_type_perms, attr)
+        self._handle_permissions(imms_vax_type_perms, attr)
         update_exp = self._build_update_expression(is_reinstate=False)
 
         self._check_duplicate_identifier(attr)
@@ -232,10 +229,10 @@ def reinstate_immunization(
         existing_resource_version: int,
         imms_vax_type_perms: str,
         supplier_system: str,
-        is_imms_batch_app: bool,
+
     ) -> dict:
         attr = RecordAttributes(immunization, patient)
-        self._handle_permissions(is_imms_batch_app, imms_vax_type_perms, attr)
+        self._handle_permissions(imms_vax_type_perms, attr)
         update_exp = self._build_update_expression(is_reinstate=True)
 
         self._check_duplicate_identifier(attr)
@@ -258,10 +255,9 @@ def update_reinstated_immunization(
         existing_resource_version: int,
         imms_vax_type_perms: str,
         supplier_system: str,
-        is_imms_batch_app: bool,
     ) -> dict:
         attr = RecordAttributes(immunization, patient)
-        self._handle_permissions(is_imms_batch_app, imms_vax_type_perms, attr)
+        self._handle_permissions(imms_vax_type_perms, attr)
         update_exp = self._build_update_expression(is_reinstate=False)
 
         self._check_duplicate_identifier(attr)
@@ -276,11 +272,10 @@ def update_reinstated_immunization(
             update_reinstated=True,
         )
 
-    def _handle_permissions(self, is_imms_batch_app: bool, imms_vax_type_perms: str, attr: RecordAttributes):
-        if not is_imms_batch_app:
-            vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
-            vax_type_perm = self._vaccine_permission(attr.vaccine_type, "update")
-            self._check_permission(vax_type_perm, vax_type_perms)
+    def _handle_permissions(self, imms_vax_type_perms: str, attr: RecordAttributes):
+        vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
+        vax_type_perm = self._vaccine_permission(attr.vaccine_type, "update")
+        self._check_permission(vax_type_perm, vax_type_perms)
 
     def _build_update_expression(self, is_reinstate: bool) -> str:
         if is_reinstate:
@@ -365,7 +360,7 @@ def _perform_dynamo_update(
                 )
 
     def delete_immunization(
-        self, imms_id: str, imms_vax_type_perms: str, supplier_system: str, is_imms_batch_app
+        self, imms_id: str, imms_vax_type_perms: str, supplier_system: str
     ) -> dict:
         now_timestamp = int(time.time())
         try:
@@ -375,16 +370,14 @@ def delete_immunization(
                 if "DeletedAt" in resp["Item"]:
                     if resp["Item"]["DeletedAt"] == "reinstated":
                         vaccine_type = self._vaccine_type(resp["Item"]["PatientSK"])
-                        if not is_imms_batch_app:
-                            vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
-                            vax_type_perm = self._vaccine_permission(vaccine_type, "delete")
-                            self._check_permission(vax_type_perm, vax_type_perms)
-                else:
-                    vaccine_type = self._vaccine_type(resp["Item"]["PatientSK"])
-                    if not is_imms_batch_app:
                         vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
                         vax_type_perm = self._vaccine_permission(vaccine_type, "delete")
                         self._check_permission(vax_type_perm, vax_type_perms)
+                else:
+                    vaccine_type = self._vaccine_type(resp["Item"]["PatientSK"])
+                    vax_type_perms = self._parse_vaccine_permissions(imms_vax_type_perms)
+                    vax_type_perm = self._vaccine_permission(vaccine_type, "delete")
+                    self._check_permission(vax_type_perm, vax_type_perms)
 
             response = self.table.update_item(
                 Key={"PK": _make_immunization_pk(imms_id)},

backend/src/fhir_service.py

@@ -180,23 +180,20 @@ def update_reinstated_immunization(
         existing_resource_version: int,
         imms_vax_type_perms: str,
         supplier_system: str,
-        is_imms_batch_app,
     ) -> tuple[UpdateOutcome, Immunization]:
         immunization["id"] = imms_id
 
         patient = None
-        if not is_imms_batch_app:
-            patient = self._validate_patient(immunization)
-            if "diagnostics" in patient:
-                return (None, patient)
+        patient = self._validate_patient(immunization)
+        if "diagnostics" in patient:
+            return (None, patient)
         imms = self.immunization_repo.update_reinstated_immunization(
             imms_id,
             immunization,
             patient,
             existing_resource_version,
             imms_vax_type_perms,
             supplier_system,
-            is_imms_batch_app,
         )
 
         return UpdateOutcome.UPDATE, Immunization.parse_obj(imms)

backend/tests/test_fhir_controller_authorization.py

@@ -1,5 +1,6 @@
 import json
 import unittest
+from unittest.mock import patch
 import uuid
 from typing import Set
 from unittest.mock import create_autospec
@@ -75,8 +76,10 @@ def test_get_imms_by_id_unknown_permission(self):
         self.assertEqual(body["issue"][0]["code"], "exception")
 
     # EndpointOperation.CREATE
-    def test_create_imms_authorized(self):
-        aws_event = {"headers":{"VaccineTypePermissions":"COVID19:create", "SupplierSystem" : "Test"},"body": create_covid_19_immunization(str(uuid.uuid4())).json()}
+    @patch("fhir_controller.get_supplier_permissions")
+    def test_create_imms_authorized_supplier_system(self, mock_get_supplier_permissions):
+        mock_get_supplier_permissions.return_value = ["Covid19:create"]
+        aws_event = {"headers":{"SupplierSystem" : "Test"},"body": create_covid_19_immunization(str(uuid.uuid4())).json()}
 
         _ = self.controller.create_immunization(aws_event)
 
@@ -105,17 +108,21 @@ def test_create_imms_unknown_permission(self):
         self.assertEqual(body["issue"][0]["code"], "exception")
 
     # EndpointOperation.UPDATE
-    def test_update_imms_authorized(self):
+    @patch("fhir_controller.get_supplier_permissions")
+    def test_update_imms_authorized(self, mock_get_supplier_permissions):
+        mock_get_supplier_permissions.return_value = ["Covid19:update"]
         imms_id = str(uuid.uuid4())
-        aws_event = {"headers": {"E-Tag":1,"VaccineTypePermissions":"COVID19:update", "SupplierSystem" : "Test"},"pathParameters": {"id": imms_id}, "body": create_covid_19_immunization(imms_id).json()}
+        aws_event = {"headers": {"E-Tag":1, "SupplierSystem" : "Test"},"pathParameters": {"id": imms_id}, "body": create_covid_19_immunization(imms_id).json()}
         self.service.get_immunization_by_id_all.return_value = {"resource":"new_value","Version":2,"DeletedAt": False, "VaccineType":"COVID19"}
         self.service.update_immunization.return_value = UpdateOutcome.UPDATE, "value doesn't matter"
 
         _ = self.controller.update_immunization(aws_event)
 
         self.authorizer.authorize.assert_called_once_with(EndpointOperation.UPDATE, aws_event)
 
-    def test_update_imms_unauthorized_vaxx_in_record(self):
+    @patch("fhir_controller.get_supplier_permissions")
+    def test_update_imms_unauthorized_vaxx_in_record(self,mock_get_supplier_permissions):
+        mock_get_supplier_permissions.return_value = ["Covid19:update"]
         imms_id = str(uuid.uuid4())
         aws_event = {"headers": {"E-Tag":1,"VaccineTypePermissions":"COVID19:update", "SupplierSystem" : "Test"},"pathParameters": {"id": imms_id}, "body": create_covid_19_immunization(imms_id).json()}
         self.service.get_immunization_by_id_all.return_value = {"resource":"new_value","Version":1,"DeletedAt": False, "VaccineType":"Flu"}
@@ -153,7 +160,9 @@ def test_update_imms_unknown_permission(self):
         self.assertEqual(body["issue"][0]["code"], "exception")
 
     # EndpointOperation.DELETE
-    def test_delete_imms_authorized(self):
+    @patch("fhir_controller.get_supplier_permissions")
+    def test_delete_imms_authorized(self, mock_get_supplier_permissions):
+        mock_get_supplier_permissions.return_value = ["COVID19:delete"]
         aws_event = {"pathParameters": {"id": "an-id"},"headers": {"VaccineTypePermissions":"COVID19:delete", "SupplierSystem" : "Test"}}
 
         _ = self.controller.delete_immunization(aws_event)