From ba0dbbd4f9f4f101ce5d4e012a4b338316fa3f4c Mon Sep 17 00:00:00 2001 From: Aakash Singh Date: Fri, 24 Jan 2025 14:13:40 +0530 Subject: [PATCH] Add chronic condition api --- care/emr/api/viewsets/base.py | 11 +- care/emr/api/viewsets/condition.py | 100 ++- .../0011_alter_condition_encounter.py | 19 + care/emr/models/condition.py | 4 +- care/emr/resources/condition/spec.py | 16 +- care/emr/tests/test_chronic_condition_api.py | 766 ++++++++++++++++++ config/api_router.py | 9 +- 7 files changed, 906 insertions(+), 19 deletions(-) create mode 100644 care/emr/migrations/0011_alter_condition_encounter.py create mode 100644 care/emr/tests/test_chronic_condition_api.py diff --git a/care/emr/api/viewsets/base.py b/care/emr/api/viewsets/base.py index b2db441bad..26f623d3fc 100644 --- a/care/emr/api/viewsets/base.py +++ b/care/emr/api/viewsets/base.py @@ -145,13 +145,14 @@ def perform_update(self, instance): updated_by=self.request.user, ) - def clean_update_data(self, request_data): + def clean_update_data(self, request_data, keep_fields: set | None = None): if type(request_data) is list: return request_data - request_data.pop("id", None) - request_data.pop("external_id", None) - request_data.pop("patient", None) - request_data.pop("encounter", None) + ignored_fields = {"id", "external_id", "patient", "encounter"} + if keep_fields: + ignored_fields = ignored_fields - set(keep_fields) + for field in ignored_fields: + request_data.pop(field, None) return request_data def update(self, request, *args, **kwargs): diff --git a/care/emr/api/viewsets/condition.py b/care/emr/api/viewsets/condition.py index 52c2ce9a3f..aeccfab3d9 100644 --- a/care/emr/api/viewsets/condition.py +++ b/care/emr/api/viewsets/condition.py @@ -1,22 +1,34 @@ -from django.shortcuts import get_object_or_404 from django_filters import CharFilter, FilterSet, UUIDFilter from django_filters.rest_framework import DjangoFilterBackend -from rest_framework.exceptions import ValidationError +from rest_framework.exceptions import PermissionDenied, ValidationError +from rest_framework.generics import get_object_or_404 -from care.emr.api.viewsets.base import EMRModelViewSet, EMRQuestionnaireResponseMixin +from care.emr.api.viewsets.base import ( + EMRBaseViewSet, + EMRCreateMixin, + EMRListMixin, + EMRModelViewSet, + EMRQuestionnaireResponseMixin, + EMRRetrieveMixin, + EMRUpdateMixin, + EMRUpsertMixin, +) from care.emr.api.viewsets.encounter_authz_base import EncounterBasedAuthorizationBase from care.emr.models.condition import Condition from care.emr.models.encounter import Encounter +from care.emr.models.patient import Patient from care.emr.registries.system_questionnaire.system_questionnaire import ( InternalQuestionnaireRegistry, ) from care.emr.resources.condition.spec import ( CategoryChoices, + ChronicConditionUpdateSpec, + ConditionReadSpec, ConditionSpec, - ConditionSpecRead, - ConditionSpecUpdate, + ConditionUpdateSpec, ) from care.emr.resources.questionnaire.spec import SubjectType +from care.security.authorization import AuthorizationController class ValidateEncounterMixin: @@ -54,8 +66,8 @@ class SymptomViewSet( ): database_model = Condition pydantic_model = ConditionSpec - pydantic_read_model = ConditionSpecRead - pydantic_update_model = ConditionSpecUpdate + pydantic_read_model = ConditionReadSpec + pydantic_update_model = ConditionUpdateSpec # Filters filterset_class = ConditionFilters filter_backends = [DjangoFilterBackend] @@ -94,8 +106,8 @@ class DiagnosisViewSet( ): database_model = Condition pydantic_model = ConditionSpec - pydantic_read_model = ConditionSpecRead - pydantic_update_model = ConditionSpecUpdate + pydantic_read_model = ConditionReadSpec + pydantic_update_model = ConditionUpdateSpec # Filters filterset_class = ConditionFilters @@ -125,3 +137,73 @@ def get_queryset(self): InternalQuestionnaireRegistry.register(DiagnosisViewSet) + + +class ChronicConditionViewSet( + EMRQuestionnaireResponseMixin, + EMRCreateMixin, + EMRRetrieveMixin, + EMRUpdateMixin, + EMRListMixin, + EMRBaseViewSet, + EMRUpsertMixin, +): + database_model = Condition + pydantic_model = ConditionSpec + pydantic_read_model = ConditionReadSpec + pydantic_update_model = ChronicConditionUpdateSpec + + # Filters + filterset_class = ConditionFilters + filter_backends = [DjangoFilterBackend] + # Questionnaire Spec + questionnaire_type = "chronic_condition" + questionnaire_title = "Chronic Condition" + questionnaire_description = "Chronic Condition" + questionnaire_subject_type = SubjectType.patient.value + + def get_patient_obj(self): + return get_object_or_404( + Patient, external_id=self.kwargs["patient_external_id"] + ) + + 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): + self.authorize_create({}) + + def perform_create(self, instance): + instance.category = CategoryChoices.chronic_condition.value + super().perform_create(instance) + + def clean_update_data(self, request_data): + return super().clean_update_data(request_data, keep_fields={"encounter"}) + + def perform_update(self, instance): + if not AuthorizationController.call( + "can_update_encounter_obj", self.request.user, instance.encounter + ): + raise PermissionDenied("You do not have permission to update encounter") + super().perform_update(instance) + + def get_queryset(self): + if not AuthorizationController.call( + "can_view_clinical_data", self.request.user, self.get_patient_obj() + ): + raise PermissionDenied("Permission denied for patient data") + return ( + super() + .get_queryset() + .filter( + patient__external_id=self.kwargs["patient_external_id"], + category=CategoryChoices.chronic_condition.value, + ) + .select_related("patient", "encounter", "created_by", "updated_by") + ) + + +InternalQuestionnaireRegistry.register(ChronicConditionViewSet) diff --git a/care/emr/migrations/0011_alter_condition_encounter.py b/care/emr/migrations/0011_alter_condition_encounter.py new file mode 100644 index 0000000000..8dbd2aa6cb --- /dev/null +++ b/care/emr/migrations/0011_alter_condition_encounter.py @@ -0,0 +1,19 @@ +# Generated by Django 5.1.4 on 2025-01-24 07:58 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('emr', '0010_condition_abatement'), + ] + + operations = [ + migrations.AlterField( + model_name='condition', + name='encounter', + field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to='emr.encounter'), + ), + ] diff --git a/care/emr/models/condition.py b/care/emr/models/condition.py index 700a2114d1..4b33a146c9 100644 --- a/care/emr/models/condition.py +++ b/care/emr/models/condition.py @@ -11,7 +11,9 @@ class Condition(EMRBaseModel): code = models.JSONField(default=dict, null=False, blank=False) body_site = models.JSONField(default=dict, null=False, blank=False) patient = models.ForeignKey("emr.Patient", on_delete=models.CASCADE) - encounter = models.ForeignKey("emr.Encounter", on_delete=models.CASCADE) + encounter = models.ForeignKey( + "emr.Encounter", on_delete=models.CASCADE, null=True, blank=True + ) onset = models.JSONField(default=dict) abatement = models.JSONField(default=dict) recorded_date = models.DateTimeField(null=True, blank=True) diff --git a/care/emr/resources/condition/spec.py b/care/emr/resources/condition/spec.py index baea0fe074..32efb3c712 100644 --- a/care/emr/resources/condition/spec.py +++ b/care/emr/resources/condition/spec.py @@ -2,6 +2,7 @@ from enum import Enum from pydantic import UUID4, Field, field_validator +from rest_framework.generics import get_object_or_404 from care.emr.fhir.schema.base import Coding from care.emr.models.condition import Condition @@ -96,7 +97,7 @@ def perform_extra_deserialization(self, is_update, obj): obj.patient = obj.encounter.patient -class ConditionSpecRead(BaseConditionSpec): +class ConditionReadSpec(BaseConditionSpec): """ Validation for deeper models may not be required on read, Just an extra optimisation """ @@ -119,7 +120,8 @@ class ConditionSpecRead(BaseConditionSpec): @classmethod def perform_extra_serialization(cls, mapping, obj): mapping["id"] = obj.external_id - mapping["encounter"] = obj.encounter.external_id + if obj.encounter: + mapping["encounter"] = obj.encounter.external_id if obj.created_by: mapping["created_by"] = UserSpec.serialize(obj.created_by) @@ -127,7 +129,7 @@ def perform_extra_serialization(cls, mapping, obj): mapping["updated_by"] = UserSpec.serialize(obj.updated_by) -class ConditionSpecUpdate(BaseConditionSpec): +class ConditionUpdateSpec(BaseConditionSpec): clinical_status: ClinicalStatusChoices | None = None verification_status: VerificationStatusChoices severity: SeverityChoices | None = None @@ -142,3 +144,11 @@ def validate_code(cls, code: int): return validate_valueset( "code", cls.model_fields["code"].json_schema_extra["slug"], code ) + + +class ChronicConditionUpdateSpec(ConditionUpdateSpec): + encounter: UUID4 | None + + def perform_extra_deserialization(self, is_update, obj): + if self.encounter: + obj.encounter = get_object_or_404(Encounter, external_id=self.encounter) diff --git a/care/emr/tests/test_chronic_condition_api.py b/care/emr/tests/test_chronic_condition_api.py new file mode 100644 index 0000000000..db97188906 --- /dev/null +++ b/care/emr/tests/test_chronic_condition_api.py @@ -0,0 +1,766 @@ +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 import Condition +from care.emr.resources.condition.spec import ( + CategoryChoices, + ClinicalStatusChoices, + SeverityChoices, + 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 TestChronicConditionViewSet(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( + "chronic-condition-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.condition.spec.validate_valueset", + return_value=self.valid_code, + ) + self.mock_validate_valueset = self.patcher.start() + + def tearDown(self): + self.patcher.stop() + + def _get_chronic_condition_url(self, chronic_condition_id): + """Helper to get the detail URL for a specific chronic_condition.""" + return reverse( + "chronic-condition-detail", + kwargs={ + "patient_external_id": self.patient.external_id, + "external_id": chronic_condition_id, + }, + ) + + def create_chronic_condition(self, encounter, patient, **kwargs): + clinical_status = kwargs.pop( + "clinical_status", choice(list(ClinicalStatusChoices)).value + ) + verification_status = kwargs.pop( + "verification_status", choice(list(VerificationStatusChoices)).value + ) + severity = kwargs.pop("severity", choice(list(SeverityChoices)).value) + + return baker.make( + Condition, + encounter=encounter, + patient=patient, + category=CategoryChoices.chronic_condition.value, + clinical_status=clinical_status, + verification_status=verification_status, + severity=severity, + **kwargs, + ) + + def generate_data_for_chronic_condition(self, encounter, **kwargs): + clinical_status = kwargs.pop( + "clinical_status", choice(list(ClinicalStatusChoices)).value + ) + verification_status = kwargs.pop( + "verification_status", choice(list(VerificationStatusChoices)).value + ) + severity = kwargs.pop("severity", choice(list(SeverityChoices)).value) + code = self.valid_code + return { + "encounter": encounter.external_id, + "category": CategoryChoices.chronic_condition.value, + "clinical_status": clinical_status, + "verification_status": verification_status, + "severity": severity, + "code": code, + **kwargs, + } + + # LIST TESTS + def test_list_chronic_condition_with_permissions(self): + """ + Users with `can_view_clinical_data` on a non-completed encounter + can list chronic_condition (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_chronic_condition_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_chronic_condition_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_chronic_condition_for_single_encounter_with_permissions(self): + """ + Users with `can_view_clinical_data` can list chronic_condition 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_chronic_condition_for_single_encounter_with_permissions_and_encounter_status_completed( + self, + ): + """ + Users with `can_view_clinical_data` on a completed encounter cannot list chronic_condition (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_chronic_condition_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_chronic_condition_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, + ) + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + + response = self.client.post( + self.base_url, chronic_condition_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_chronic_condition_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 chronic_condition. + """ + 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, + ) + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + + response = self.client.post( + self.base_url, chronic_condition_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_chronic_condition_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 chronic_condition data (HTTP 200) and is able to edit chronic_condition + and chronic_condition 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 chronic_condition data (HTTP 200) + test_url = reverse( + "chronic-condition-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, + ) + + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + response = self.client.post( + test_url, chronic_condition_data_dict, format="json" + ) + + self.assertEqual(response.status_code, 200) + + def test_create_chronic_condition_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, + ) + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + + response = self.client.post( + self.base_url, chronic_condition_data_dict, format="json" + ) + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json()["severity"], chronic_condition_data_dict["severity"] + ) + self.assertEqual(response.json()["code"], chronic_condition_data_dict["code"]) + + def test_create_chronic_condition_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, + ) + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + + response = self.client.post( + self.base_url, chronic_condition_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_chronic_condition_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 chronic_condition. + """ + 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, + ) + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + + response = self.client.post( + self.base_url, chronic_condition_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_chronic_condition_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, + ) + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + + response = self.client.post( + self.base_url, chronic_condition_data_dict, format="json" + ) + self.assertEqual(response.status_code, 403) + + def test_create_chronic_condition_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, + ) + chronic_condition_data_dict = self.generate_data_for_chronic_condition( + encounter + ) + chronic_condition_data_dict["encounter"] = uuid.uuid4() + + response = self.client.post( + self.base_url, chronic_condition_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_chronic_condition_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, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + retrieve_response = self.client.get(url) + self.assertEqual(retrieve_response.status_code, 200) + self.assertEqual( + retrieve_response.data["id"], str(chronic_condition.external_id) + ) + + def test_retrieve_chronic_condition_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, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.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(chronic_condition.external_id) + ) + + def test_retrieve_chronic_condition_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, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + retrieve_response = self.client.get(f"{url}?encounter={encounter.external_id}") + self.assertEqual(retrieve_response.status_code, 403) + + def test_retrieve_chronic_condition_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, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + retrieve_response = self.client.get(url) + self.assertEqual(retrieve_response.status_code, 403) + + # UPDATE TESTS + def test_update_chronic_condition_with_permissions(self): + """ + Users with `can_write_patient` + `can_view_clinical_data` + => (HTTP 200) when updating. + """ + 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.patient, + facility=self.facility, + organization=self.organization, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + chronic_condition_data_updated = model_to_dict(chronic_condition) + chronic_condition_data_updated["severity"] = "mild" + chronic_condition_data_updated["code"] = self.valid_code + + response = self.client.put(url, chronic_condition_data_updated, format="json") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["severity"], "mild") + + def test_update_chronic_condition_for_single_encounter_with_permissions(self): + """ + Users with `can_write_patient` + `can_view_clinical_data` + => (HTTP 200). + """ + 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.patient, + facility=self.facility, + organization=self.organization, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + chronic_condition_data_updated = model_to_dict(chronic_condition) + chronic_condition_data_updated["severity"] = "mild" + chronic_condition_data_updated["code"] = self.valid_code + + update_response = self.client.put( + f"{url}?encounter={encounter.external_id}", + chronic_condition_data_updated, + format="json", + ) + self.assertEqual(update_response.status_code, 200) + self.assertEqual(update_response.json()["severity"], "mild") + + def test_update_chronic_condition_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, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + chronic_condition_data_updated = model_to_dict(chronic_condition) + chronic_condition_data_updated["severity"] = "mild" + + update_response = self.client.put( + f"{url}?encounter={encounter.external_id}", + chronic_condition_data_updated, + format="json", + ) + self.assertEqual(update_response.status_code, 403) + + def test_update_chronic_condition_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, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + chronic_condition_data_updated = model_to_dict(chronic_condition) + chronic_condition_data_updated["severity"] = "mild" + + update_response = self.client.put( + url, chronic_condition_data_updated, format="json" + ) + self.assertEqual(update_response.status_code, 403) + + def test_update_chronic_condition_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, + ) + chronic_condition = self.create_chronic_condition( + encounter=encounter, patient=self.patient + ) + + url = self._get_chronic_condition_url(chronic_condition.external_id) + chronic_condition_data_updated = model_to_dict(chronic_condition) + chronic_condition_data_updated["severity"] = "mild" + + update_response = self.client.put( + url, chronic_condition_data_updated, format="json" + ) + self.assertEqual(update_response.status_code, 403) + + def test_update_chronic_condition_changes_encounter_id(self): + """ + When a user with access to a new encounter + updates a chronic_condition 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), + ) + + chronic_condition = self.create_chronic_condition( + 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_chronic_condition_url(chronic_condition.external_id) + chronic_condition_data_updated = model_to_dict(chronic_condition) + chronic_condition_data_updated["encounter"] = new_encounter.external_id + chronic_condition_data_updated["clinical_status"] = "remission" + + update_response = self.client.put( + url, chronic_condition_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_chronic_condition_changes_encounter_id_without_permission(self): + """ + When a user without access to a new encounter + updates a chronic_condition 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), + ) + + chronic_condition = self.create_chronic_condition( + 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_chronic_condition_url(chronic_condition.external_id) + chronic_condition_data_updated = model_to_dict(chronic_condition) + chronic_condition_data_updated["encounter"] = new_encounter.external_id + chronic_condition_data_updated["clinical_status"] = "remission" + + update_response = self.client.put( + url, chronic_condition_data_updated, format="json" + ) + self.assertEqual(update_response.status_code, 403) diff --git a/config/api_router.py b/config/api_router.py index 6cd54c4bc9..913ffe9974 100644 --- a/config/api_router.py +++ b/config/api_router.py @@ -8,7 +8,11 @@ from care.emr.api.otp_viewsets.slot import OTPSlotViewSet from care.emr.api.viewsets.allergy_intolerance import AllergyIntoleranceViewSet from care.emr.api.viewsets.batch_request import BatchRequestView -from care.emr.api.viewsets.condition import DiagnosisViewSet, SymptomViewSet +from care.emr.api.viewsets.condition import ( + ChronicConditionViewSet, + DiagnosisViewSet, + SymptomViewSet, +) from care.emr.api.viewsets.encounter import EncounterViewSet from care.emr.api.viewsets.facility import ( AllFacilityViewSet, @@ -243,6 +247,9 @@ patient_nested_router.register(r"symptom", SymptomViewSet, basename="symptom") patient_nested_router.register(r"diagnosis", DiagnosisViewSet, basename="diagnosis") +patient_nested_router.register( + r"chronic_condition", ChronicConditionViewSet, basename="chronic-condition" +) patient_nested_router.register( "observation", ObservationViewSet, basename="observation"