From 6add747d24a81840d02e18d99f426ae63e1454b6 Mon Sep 17 00:00:00 2001 From: Aakash Singh Date: Fri, 24 Jan 2025 18:21:18 +0530 Subject: [PATCH] Update allergy_intolerance update spec and added tests (#2778) --- care/emr/api/viewsets/allergy_intolerance.py | 28 +- .../emr/resources/allergy_intolerance/spec.py | 34 +- .../emr/tests/test_allergy_intolerance_api.py | 778 ++++++++++++++++++ 3 files changed, 823 insertions(+), 17 deletions(-) create mode 100644 care/emr/tests/test_allergy_intolerance_api.py diff --git a/care/emr/api/viewsets/allergy_intolerance.py b/care/emr/api/viewsets/allergy_intolerance.py index d174fb5f4b..979af26197 100644 --- a/care/emr/api/viewsets/allergy_intolerance.py +++ b/care/emr/api/viewsets/allergy_intolerance.py @@ -15,13 +15,14 @@ ) from care.emr.models import Patient from care.emr.models.allergy_intolerance import AllergyIntolerance +from care.emr.models.encounter import Encounter from care.emr.registries.system_questionnaire.system_questionnaire import ( InternalQuestionnaireRegistry, ) from care.emr.resources.allergy_intolerance.spec import ( - AllergyIntoleranceSpec, + AllergyIntoleranceReadSpec, + AllergyIntoleranceUpdateSpec, AllergyIntoleranceWriteSpec, - AllergyIntrolanceSpecRead, ) from care.emr.resources.questionnaire.spec import SubjectType from care.security.authorization import AuthorizationController @@ -32,7 +33,7 @@ class AllergyIntoleranceFilters(FilterSet): @extend_schema_view( - create=extend_schema(request=AllergyIntoleranceSpec), + create=extend_schema(request=AllergyIntoleranceWriteSpec), ) class AllergyIntoleranceViewSet( EMRQuestionnaireResponseMixin, @@ -44,9 +45,9 @@ class AllergyIntoleranceViewSet( EMRUpsertMixin, ): database_model = AllergyIntolerance - pydantic_model = AllergyIntoleranceSpec - pydantic_read_model = AllergyIntrolanceSpecRead - pydantic_update_model = AllergyIntoleranceWriteSpec + pydantic_model = AllergyIntoleranceWriteSpec + pydantic_read_model = AllergyIntoleranceReadSpec + pydantic_update_model = AllergyIntoleranceUpdateSpec questionnaire_type = "allergy_intolerance" questionnaire_title = "Allergy Intolerance" questionnaire_description = "Allergy Intolerance" @@ -59,15 +60,24 @@ def get_patient_obj(self): Patient, external_id=self.kwargs["patient_external_id"] ) - def authorize_update(self, request_obj, model_instance): - self.authorize_create({}) - def authorize_create(self, instance): if not AuthorizationController.call( "can_write_patient_obj", self.request.user, self.get_patient_obj() ): raise PermissionDenied("You do not have permission to update encounter") + def authorize_update(self, request_obj, model_instance): + encounter = get_object_or_404(Encounter, external_id=request_obj.encounter) + if not AuthorizationController.call( + "can_update_encounter_obj", + self.request.user, + encounter, + ): + raise PermissionDenied("You do not have permission to update encounter") + + def clean_update_data(self, request_data): + return super().clean_update_data(request_data, keep_fields={"encounter"}) + def get_queryset(self): if not AuthorizationController.call( "can_view_clinical_data", self.request.user, self.get_patient_obj() diff --git a/care/emr/resources/allergy_intolerance/spec.py b/care/emr/resources/allergy_intolerance/spec.py index d67e5bc77b..40a16c95fb 100644 --- a/care/emr/resources/allergy_intolerance/spec.py +++ b/care/emr/resources/allergy_intolerance/spec.py @@ -52,27 +52,39 @@ class BaseAllergyIntoleranceSpec(EMRResource): id: UUID4 = None -class AllergyIntoleranceWriteSpec(BaseAllergyIntoleranceSpec): +class AllergyIntoleranceUpdateSpec(BaseAllergyIntoleranceSpec): clinical_status: ClinicalStatusChoices verification_status: VerificationStatusChoices - category: CategoryChoices criticality: CriticalityChoices last_occurrence: datetime.datetime | None = None - recorded_date: datetime.datetime | None = None + note: str | None = None + encounter: UUID4 - onset: AllergyIntoleranceOnSetSpec = {} + @field_validator("encounter") + @classmethod + def validate_encounter_exists(cls, encounter): + if not Encounter.objects.filter(external_id=encounter).exists(): + err = "Encounter not found" + raise ValueError(err) + return encounter def perform_extra_deserialization(self, is_update, obj): - if not is_update: + if self.encounter: obj.encounter = Encounter.objects.get(external_id=self.encounter) - obj.patient = obj.encounter.patient -class AllergyIntoleranceSpec(AllergyIntoleranceWriteSpec): +class AllergyIntoleranceWriteSpec(BaseAllergyIntoleranceSpec): + clinical_status: ClinicalStatusChoices + verification_status: VerificationStatusChoices + category: CategoryChoices + criticality: CriticalityChoices + last_occurrence: datetime.datetime | None = None + recorded_date: datetime.datetime | None = None encounter: UUID4 code: Coding = Field( {}, json_schema_extra={"slug": CARE_ALLERGY_CODE_VALUESET.slug} ) + onset: AllergyIntoleranceOnSetSpec = {} @field_validator("code") @classmethod @@ -89,8 +101,12 @@ def validate_encounter_exists(cls, encounter): raise ValueError(err) return encounter + def perform_extra_deserialization(self, is_update, obj): + obj.encounter = Encounter.objects.get(external_id=self.encounter) + obj.patient = obj.encounter.patient + -class AllergyIntrolanceSpecRead(BaseAllergyIntoleranceSpec): +class AllergyIntoleranceReadSpec(BaseAllergyIntoleranceSpec): """ Validation for deeper models may not be required on read, Just an extra optimisation """ @@ -115,3 +131,5 @@ def perform_extra_serialization(cls, mapping, obj): mapping["created_by"] = UserSpec.serialize(obj.created_by) if obj.updated_by: mapping["updated_by"] = UserSpec.serialize(obj.updated_by) + if obj.encounter: + mapping["encounter"] = obj.encounter.external_id diff --git a/care/emr/tests/test_allergy_intolerance_api.py b/care/emr/tests/test_allergy_intolerance_api.py new file mode 100644 index 0000000000..1cda88db8b --- /dev/null +++ b/care/emr/tests/test_allergy_intolerance_api.py @@ -0,0 +1,778 @@ +import uuid +from secrets import choice +from unittest.mock import patch + +from django.forms import model_to_dict +from django.urls import reverse +from model_bakery import baker + +from care.emr.models.allergy_intolerance import AllergyIntolerance +from care.emr.resources.allergy_intolerance.spec import ( + CategoryChoices, + ClinicalStatusChoices, + CriticalityChoices, + VerificationStatusChoices, +) +from care.emr.resources.resource_request.spec import StatusChoices +from care.security.permissions.encounter import EncounterPermissions +from care.security.permissions.patient import PatientPermissions +from care.utils.tests.base import CareAPITestBase + + +class TestAllergyIntoleranceViewSet(CareAPITestBase): + def setUp(self): + super().setUp() + self.user = self.create_user() + self.facility = self.create_facility(user=self.user) + self.organization = self.create_facility_organization(facility=self.facility) + self.patient = self.create_patient() + self.client.force_authenticate(user=self.user) + + self.base_url = reverse( + "allergy-intolerance-list", + kwargs={"patient_external_id": self.patient.external_id}, + ) + self.valid_code = { + "display": "Test Value", + "system": "http://test_system.care/test", + "code": "123", + } + # Mocking validate_valueset + self.patcher = patch( + "care.emr.resources.allergy_intolerance.spec.validate_valueset", + return_value=self.valid_code, + ) + self.mock_validate_valueset = self.patcher.start() + + def tearDown(self): + self.patcher.stop() + + def _get_allergy_intolerance_url(self, allergy_intolerance_id): + """Helper to get the detail URL for a specific allergy_intolerance.""" + return reverse( + "allergy-intolerance-detail", + kwargs={ + "patient_external_id": self.patient.external_id, + "external_id": allergy_intolerance_id, + }, + ) + + def create_allergy_intolerance(self, encounter, patient, **kwargs): + clinical_status = kwargs.pop( + "clinical_status", choice(list(ClinicalStatusChoices)).value + ) + verification_status = kwargs.pop( + "verification_status", choice(list(VerificationStatusChoices)).value + ) + category = kwargs.pop("category", choice(list(CategoryChoices)).value) + criticality = kwargs.pop("criticality", choice(list(CriticalityChoices)).value) + + return baker.make( + AllergyIntolerance, + encounter=encounter, + patient=patient, + category=category, + clinical_status=clinical_status, + verification_status=verification_status, + criticality=criticality, + **kwargs, + ) + + def generate_data_for_allergy_intolerance(self, encounter, **kwargs): + clinical_status = kwargs.pop( + "clinical_status", choice(list(ClinicalStatusChoices)).value + ) + verification_status = kwargs.pop( + "verification_status", choice(list(VerificationStatusChoices)).value + ) + category = kwargs.pop("category", choice(list(CategoryChoices)).value) + criticality = kwargs.pop("criticality", choice(list(CriticalityChoices)).value) + code = self.valid_code + return { + "encounter": encounter.external_id, + "category": category, + "clinical_status": clinical_status, + "verification_status": verification_status, + "criticality": criticality, + "code": code, + **kwargs, + } + + # LIST TESTS + def test_list_allergy_intolerance_with_permissions(self): + """ + Users with `can_view_clinical_data` on a non-completed encounter + can list allergy_intolerance (HTTP 200). + """ + # Attach the needed role/permission + permissions = [PatientPermissions.can_view_clinical_data.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + # Create an active encounter + self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + + response = self.client.get(self.base_url) + self.assertEqual(response.status_code, 200) + + def test_list_allergy_intolerance_with_permissions_and_encounter_status_as_completed( + self, + ): + """ + Users with `can_view_clinical_data` but a completed encounter => (HTTP 403). + """ + permissions = [PatientPermissions.can_view_clinical_data.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=StatusChoices.completed.value, + ) + response = self.client.get(self.base_url) + self.assertEqual(response.status_code, 403) + + def test_list_allergy_intolerance_without_permissions(self): + """ + Users without `can_view_clinical_data` => (HTTP 403). + """ + # No permission attached + self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + response = self.client.get(self.base_url) + self.assertEqual(response.status_code, 403) + + def test_list_allergy_intolerance_for_single_encounter_with_permissions(self): + """ + Users with `can_view_clinical_data` can list allergy_intolerance for that encounter (HTTP 200). + """ + permissions = [PatientPermissions.can_view_clinical_data.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + + url = f"{self.base_url}?encounter={encounter.external_id}" + response = self.client.get(url) + self.assertEqual(response.status_code, 200) + + def test_list_allergy_intolerance_for_single_encounter_with_permissions_and_encounter_status_completed( + self, + ): + """ + Users with `can_view_clinical_data` on a completed encounter cannot list allergy_intolerance (HTTP 200). + """ + permissions = [PatientPermissions.can_view_clinical_data.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=StatusChoices.completed.value, + ) + url = f"{self.base_url}?encounter={encounter.external_id}" + response = self.client.get(url) + self.assertEqual(response.status_code, 403) + + def test_list_allergy_intolerance_for_single_encounter_without_permissions(self): + """ + Users without `can_view_clinical_data` or `can_view_clinical_data` => (HTTP 403). + """ + # No relevant permission + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + url = f"{self.base_url}?encounter={encounter.external_id}" + response = self.client.get(url) + self.assertEqual(response.status_code, 403) + + # CREATE TESTS + def test_create_allergy_intolerance_without_permissions(self): + """ + Users who lack `can_write_patient` get (HTTP 403) when creating. + """ + # No permission attached + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + + response = self.client.post( + self.base_url, allergy_intolerance_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_allergy_intolerance_without_permissions_on_facility(self): + """ + Tests that a user with `can_write_patient` permissions but belonging to a different + organization receives (HTTP 403) when attempting to create a allergy_intolerance. + """ + permissions = [ + PatientPermissions.can_view_clinical_data.name, + PatientPermissions.can_write_patient.name, + ] + role = self.create_role_with_permissions(permissions) + external_user = self.create_user() + external_facility = self.create_facility(user=external_user) + external_organization = self.create_facility_organization( + facility=external_facility + ) + self.attach_role_facility_organization_user( + external_organization, self.user, role + ) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + + response = self.client.post( + self.base_url, allergy_intolerance_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_allergy_intolerance_with_organization_user_with_permissions(self): + """ + Ensures that a user from a certain organization, who has both + `can_write_patient` and `can_view_clinical_data`, can successfully + view allergy_intolerance data (HTTP 200) and is able to edit allergy_intolerance + and allergy_intolerance can change across encounters. + """ + organization = self.create_organization(org_type="govt") + patient = self.create_patient(geo_organization=organization) + + permissions = [ + PatientPermissions.can_write_patient.name, + PatientPermissions.can_view_clinical_data.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_organization_user(organization, self.user, role) + + # Verify the user can view allergy_intolerance data (HTTP 200) + test_url = reverse( + "allergy-intolerance-list", + kwargs={"patient_external_id": patient.external_id}, + ) + response = self.client.get(test_url) + self.assertEqual(response.status_code, 200) + + encounter = self.create_encounter( + patient=patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + response = self.client.post( + test_url, allergy_intolerance_data_dict, format="json" + ) + + self.assertEqual(response.status_code, 200) + + def test_create_allergy_intolerance_with_permissions(self): + """ + Users with `can_write_patient` on a non-completed encounter => (HTTP 200). + """ + permissions = [PatientPermissions.can_write_patient.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + + response = self.client.post( + self.base_url, allergy_intolerance_data_dict, format="json" + ) + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json()["criticality"], allergy_intolerance_data_dict["criticality"] + ) + self.assertEqual(response.json()["code"], allergy_intolerance_data_dict["code"]) + + def test_create_allergy_intolerance_with_permissions_and_encounter_status_completed( + self, + ): + """ + Users with `can_write_patient` on a completed encounter => (HTTP 403). + """ + permissions = [PatientPermissions.can_write_patient.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=StatusChoices.completed.value, + ) + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + + response = self.client.post( + self.base_url, allergy_intolerance_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_allergy_intolerance_with_permissions_and_no_association_with_facility( + self, + ): + """ + Test that users with `can_write_patient` permission, but who are not + associated with the facility, receive an HTTP 403 (Forbidden) response + when attempting to create a allergy_intolerance. + """ + permissions = [PatientPermissions.can_write_patient.name] + role = self.create_role_with_permissions(permissions) + organization = self.create_organization(org_type="govt") + self.attach_role_organization_user(organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=None, + ) + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + + response = self.client.post( + self.base_url, allergy_intolerance_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_allergy_intolerance_with_permissions_with_mismatched_patient_id( + self, + ): + """ + Users with `can_write_patient` on a encounter with different patient => (HTTP 403). + """ + permissions = [ + PatientPermissions.can_view_clinical_data.name, + PatientPermissions.can_write_patient.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.create_patient(), + facility=self.facility, + organization=self.organization, + status=None, + ) + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + + response = self.client.post( + self.base_url, allergy_intolerance_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_allergy_intolerance_with_permissions_with_invalid_encounter_id( + self, + ): + """ + Users with `can_write_patient` on a incomplete encounter => (HTTP 400). + """ + permissions = [PatientPermissions.can_write_patient.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.create_patient(), + facility=self.facility, + organization=self.organization, + status=None, + ) + allergy_intolerance_data_dict = self.generate_data_for_allergy_intolerance( + encounter + ) + allergy_intolerance_data_dict["encounter"] = uuid.uuid4() + + response = self.client.post( + self.base_url, allergy_intolerance_data_dict, format="json" + ) + response_data = response.json() + self.assertEqual(response.status_code, 400) + self.assertIn("errors", response_data) + error = response_data["errors"][0] + self.assertEqual(error["type"], "value_error") + self.assertIn("Encounter not found", error["msg"]) + + # RETRIEVE TESTS + def test_retrieve_allergy_intolerance_with_permissions(self): + """ + Users with `can_view_clinical_data` => (HTTP 200). + """ + permissions = [PatientPermissions.can_view_clinical_data.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + retrieve_response = self.client.get(url) + self.assertEqual(retrieve_response.status_code, 200) + self.assertEqual( + retrieve_response.data["id"], str(allergy_intolerance.external_id) + ) + + def test_retrieve_allergy_intolerance_for_single_encounter_with_permissions(self): + """ + Users with `can_view_clinical_data` => (HTTP 200). + """ + permissions = [ + PatientPermissions.can_view_clinical_data.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + retrieve_response = self.client.get(f"{url}?encounter={encounter.external_id}") + self.assertEqual(retrieve_response.status_code, 200) + self.assertEqual( + retrieve_response.data["id"], str(allergy_intolerance.external_id) + ) + + def test_retrieve_allergy_intolerance_for_single_encounter_without_permissions( + self, + ): + """ + Lacking `can_view_clinical_data` => (HTTP 403). + """ + # No relevant permission + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + retrieve_response = self.client.get(f"{url}?encounter={encounter.external_id}") + self.assertEqual(retrieve_response.status_code, 403) + + def test_retrieve_allergy_intolerance_without_permissions(self): + """ + Users who have only `can_write_patient` => (HTTP 403). + """ + # No relevant permission + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + retrieve_response = self.client.get(url) + self.assertEqual(retrieve_response.status_code, 403) + + # UPDATE TESTS + def test_update_allergy_intolerance_with_permissions(self): + """ + Users with `can_write_encounter` + `can_write_patient` + `can_view_clinical_data` + => (HTTP 200) when updating. + """ + permissions = [ + PatientPermissions.can_view_clinical_data.name, + PatientPermissions.can_write_patient.name, + EncounterPermissions.can_write_encounter.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + allergy_intolerance_data_updated = model_to_dict(allergy_intolerance) + allergy_intolerance_data_updated["encounter"] = encounter.external_id + allergy_intolerance_data_updated["criticality"] = "high" + allergy_intolerance_data_updated["code"] = self.valid_code + + response = self.client.put(url, allergy_intolerance_data_updated, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["criticality"], "high") + + def test_update_allergy_intolerance_for_single_encounter_with_permissions(self): + """ + Users with `can_write_encounter` + `can_write_patient` + `can_view_clinical_data` + => (HTTP 200). + """ + permissions = [ + PatientPermissions.can_view_clinical_data.name, + PatientPermissions.can_write_patient.name, + EncounterPermissions.can_write_encounter.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + allergy_intolerance_data_updated = model_to_dict(allergy_intolerance) + allergy_intolerance_data_updated["encounter"] = encounter.external_id + allergy_intolerance_data_updated["criticality"] = "high" + allergy_intolerance_data_updated["code"] = self.valid_code + + update_response = self.client.put( + f"{url}?encounter={encounter.external_id}", + allergy_intolerance_data_updated, + format="json", + ) + self.assertEqual(update_response.status_code, 200) + self.assertEqual(update_response.json()["criticality"], "high") + + def test_update_allergy_intolerance_for_single_encounter_without_permissions(self): + """ + Lacking `can_view_clinical_data` => (HTTP 403). + """ + # Only write permission + permissions = [PatientPermissions.can_write_patient.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + allergy_intolerance_data_updated = model_to_dict(allergy_intolerance) + allergy_intolerance_data_updated["criticality"] = "high" + + update_response = self.client.put( + f"{url}?encounter={encounter.external_id}", + allergy_intolerance_data_updated, + format="json", + ) + self.assertEqual(update_response.status_code, 403) + + def test_update_allergy_intolerance_without_permissions(self): + """ + Users with only `can_write_patient` but not `can_view_clinical_data` + => (HTTP 403). + """ + # Only write permission (same scenario as above but no read or view clinical) + + permissions = [PatientPermissions.can_write_patient.name] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + allergy_intolerance_data_updated = model_to_dict(allergy_intolerance) + allergy_intolerance_data_updated["criticality"] = "high" + + update_response = self.client.put( + url, allergy_intolerance_data_updated, format="json" + ) + self.assertEqual(update_response.status_code, 403) + + def test_update_allergy_intolerance_for_closed_encounter_with_permissions(self): + """ + Encounter completed => (HTTP 403) on update, + even if user has `can_write_patient` + `can_view_clinical_data`. + """ + permissions = [ + PatientPermissions.can_write_patient.name, + PatientPermissions.can_view_clinical_data.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + status=StatusChoices.completed.value, + ) + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, patient=self.patient + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + allergy_intolerance_data_updated = model_to_dict(allergy_intolerance) + allergy_intolerance_data_updated["criticality"] = "high" + + update_response = self.client.put( + url, allergy_intolerance_data_updated, format="json" + ) + self.assertEqual(update_response.status_code, 403) + + def test_update_allergy_intolerance_changes_encounter_id(self): + """ + When a user with access to a new encounter + updates a allergy_intolerance added by a different encounter, + the encounter_id should be updated to the new encounter. + """ + permissions = [ + PatientPermissions.can_write_patient.name, + PatientPermissions.can_view_clinical_data.name, + EncounterPermissions.can_write_encounter.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + temp_facility = self.create_facility(user=self.create_user()) + encounter = self.create_encounter( + patient=self.patient, + facility=temp_facility, + organization=self.create_facility_organization(facility=temp_facility), + ) + + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, + patient=self.patient, + code=self.valid_code, + ) + + new_encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + allergy_intolerance_data_updated = model_to_dict(allergy_intolerance) + allergy_intolerance_data_updated["encounter"] = new_encounter.external_id + allergy_intolerance_data_updated["clinical_status"] = "inactive" + + update_response = self.client.put( + url, allergy_intolerance_data_updated, format="json" + ) + self.assertEqual(update_response.status_code, 200) + self.assertEqual( + update_response.json()["encounter"], str(new_encounter.external_id) + ) + + def test_update_allergy_intolerance_changes_encounter_id_without_permission(self): + """ + When a user without access to a new encounter + updates a allergy_intolerance added by a different encounter, + the encounter_id should not be updated to the new encounter. + """ + permissions = [ + PatientPermissions.can_write_patient.name, + PatientPermissions.can_view_clinical_data.name, + ] + role = self.create_role_with_permissions(permissions) + self.attach_role_facility_organization_user(self.organization, self.user, role) + + temp_facility = self.create_facility(user=self.create_user()) + encounter = self.create_encounter( + patient=self.patient, + facility=temp_facility, + organization=self.create_facility_organization(facility=temp_facility), + ) + + allergy_intolerance = self.create_allergy_intolerance( + encounter=encounter, + patient=self.patient, + code=self.valid_code, + ) + + new_encounter = self.create_encounter( + patient=self.patient, + facility=self.facility, + organization=self.organization, + ) + + url = self._get_allergy_intolerance_url(allergy_intolerance.external_id) + allergy_intolerance_data_updated = model_to_dict(allergy_intolerance) + allergy_intolerance_data_updated["encounter"] = new_encounter.external_id + allergy_intolerance_data_updated["clinical_status"] = "inactive" + + update_response = self.client.put( + url, allergy_intolerance_data_updated, format="json" + ) + self.assertEqual(update_response.status_code, 403)