diff --git a/care/facility/api/serializers/patient.py b/care/facility/api/serializers/patient.py index 048b3ba715..d46316ae29 100644 --- a/care/facility/api/serializers/patient.py +++ b/care/facility/api/serializers/patient.py @@ -442,6 +442,7 @@ class Meta: "facility", "allow_transfer", "is_active", + "is_expired", ) diff --git a/care/facility/api/serializers/patient_consultation.py b/care/facility/api/serializers/patient_consultation.py index 6c0796328a..c4c34bf596 100644 --- a/care/facility/api/serializers/patient_consultation.py +++ b/care/facility/api/serializers/patient_consultation.py @@ -861,6 +861,7 @@ class PatientConsentSerializer(serializers.ModelSerializer): id = serializers.CharField(source="external_id", read_only=True) created_by = UserBaseMinimumSerializer(read_only=True) archived_by = UserBaseMinimumSerializer(read_only=True) + files = serializers.SerializerMethodField() class Meta: model = PatientConsent @@ -869,6 +870,7 @@ class Meta: "id", "type", "patient_code_status", + "files", "archived", "archived_by", "archived_date", @@ -878,6 +880,7 @@ class Meta: read_only_fields = ( "id", + "files", "created_by", "created_date", "archived", @@ -885,6 +888,23 @@ class Meta: "archived_date", ) + def get_files(self, obj): + from care.facility.api.serializers.file_upload import ( + FileUploadListSerializer, + check_permissions, + ) + + user = self.context["request"].user + file_type = FileUpload.FileType.CONSENT_RECORD + if check_permissions(file_type, obj.external_id, user, "read"): + return FileUploadListSerializer( + FileUpload.objects.filter( + associating_id=obj.external_id, file_type=file_type + ), + many=True, + ).data + return None + def validate_patient_code_status(self, value): if value == PatientCodeStatusType.NOT_SPECIFIED: raise ValidationError( diff --git a/care/facility/api/viewsets/asset.py b/care/facility/api/viewsets/asset.py index fbc53f817d..45b976f2f9 100644 --- a/care/facility/api/viewsets/asset.py +++ b/care/facility/api/viewsets/asset.py @@ -280,7 +280,7 @@ class AssetViewSet( lookup_field = "external_id" filter_backends = (filters.DjangoFilterBackend, drf_filters.SearchFilter) search_fields = ["name", "serial_number", "qr_code_id"] - permission_classes = [IsAuthenticated] + permission_classes = (IsAuthenticated, DRYPermissions) filterset_class = AssetFilter def get_queryset(self): diff --git a/care/facility/api/viewsets/encounter_symptom.py b/care/facility/api/viewsets/encounter_symptom.py index 3f49cef2de..8441366382 100644 --- a/care/facility/api/viewsets/encounter_symptom.py +++ b/care/facility/api/viewsets/encounter_symptom.py @@ -24,7 +24,7 @@ def filter_is_cured(self, queryset, name, value): class EncounterSymptomViewSet(ModelViewSet): serializer_class = EncounterSymptomSerializer permission_classes = (IsAuthenticated, DRYPermissions) - queryset = EncounterSymptom.objects.all() + queryset = EncounterSymptom.objects.select_related("created_by", "updated_by") filter_backends = (filters.DjangoFilterBackend,) filterset_class = EncounterSymptomFilter lookup_field = "external_id" diff --git a/care/facility/api/viewsets/patient.py b/care/facility/api/viewsets/patient.py index 9e8f6d3cf3..f216aa6362 100644 --- a/care/facility/api/viewsets/patient.py +++ b/care/facility/api/viewsets/patient.py @@ -279,6 +279,32 @@ def filter_by_diagnoses(self, queryset, name, value): ) return queryset.filter(filter_q) + last_consultation__consent_types = MultiSelectFilter( + method="filter_by_has_consents" + ) + + def filter_by_has_consents(self, queryset, name, value: str): + + if not value: + return queryset + + values = value.split(",") + + filter_q = Q() + + if "None" in values: + filter_q |= ~Q( + last_consultation__has_consents__len__gt=0, + ) + values.remove("None") + + if values: + filter_q |= Q( + last_consultation__has_consents__overlap=values, + ) + + return queryset.filter(filter_q) + class PatientDRYFilter(DRYPermissionFiltersBase): def filter_queryset(self, request, queryset, view): @@ -565,6 +591,14 @@ def transfer(self, request, *args, **kwargs): patient = PatientRegistration.objects.get(external_id=kwargs["external_id"]) facility = Facility.objects.get(external_id=request.data["facility"]) + if patient.is_expired: + return Response( + { + "Patient": "Patient transfer cannot be completed because the patient is expired" + }, + status=status.HTTP_406_NOT_ACCEPTABLE, + ) + if patient.is_active and facility == patient.facility: return Response( { diff --git a/care/facility/migrations/0444_patientconsultation_has_consents_and_more.py b/care/facility/migrations/0444_patientconsultation_has_consents_and_more.py new file mode 100644 index 0000000000..9126f9b48f --- /dev/null +++ b/care/facility/migrations/0444_patientconsultation_has_consents_and_more.py @@ -0,0 +1,77 @@ +# Generated by Django 4.2.10 on 2024-07-04 16:20 + +import uuid + +import django.contrib.postgres.fields +import django.db.models.deletion +from django.db import migrations, models +from django.db.models import Subquery + + +class Migration(migrations.Migration): + + def migrate_has_consents(apps, schema_editor): + FileUpload = apps.get_model("facility", "FileUpload") + PatientConsent = apps.get_model("facility", "PatientConsent") + + consents = PatientConsent.objects.filter(archived=False) + for consent in consents: + consultation = consent.consultation + consent_types = ( + PatientConsent.objects.filter(consultation=consultation, archived=False) + .annotate( + str_external_id=models.functions.Cast( + "external_id", models.CharField() + ) + ) + .annotate( + has_files=models.Exists( + FileUpload.objects.filter( + associating_id=models.OuterRef("str_external_id"), + file_type=7, + is_archived=False, + ) + ) + ) + .filter(has_files=True) + .distinct("type") + .values_list("type", flat=True) + ) + consultation.has_consents = list(consent_types) + consultation.save() + + dependencies = [ + ("facility", "0443_remove_patientconsultation_consent_records_and_more"), + ] + + operations = [ + migrations.AddField( + model_name="patientconsultation", + name="has_consents", + field=django.contrib.postgres.fields.ArrayField( + base_field=models.IntegerField( + choices=[ + (1, "Consent for Admission"), + (2, "Patient Code Status"), + (3, "Consent for Procedure"), + (4, "High Risk Consent"), + (5, "Others"), + ] + ), + default=list, + size=None, + ), + ), + migrations.AlterField( + model_name="patientconsent", + name="consultation", + field=models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="consents", + to="facility.patientconsultation", + ), + ), + migrations.RunPython( + migrate_has_consents, reverse_code=migrations.RunPython.noop + ), + ] diff --git a/care/facility/migrations/0445_merge_20240715_0301.py b/care/facility/migrations/0445_merge_20240715_0301.py new file mode 100644 index 0000000000..c06eb00297 --- /dev/null +++ b/care/facility/migrations/0445_merge_20240715_0301.py @@ -0,0 +1,13 @@ +# Generated by Django 4.2.10 on 2024-07-14 21:31 + +from django.db import migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ("facility", "0444_alter_medicineadministration_dosage_and_more"), + ("facility", "0444_patientconsultation_has_consents_and_more"), + ] + + operations = [] diff --git a/care/facility/models/asset.py b/care/facility/models/asset.py index c85a206e89..9608a1e482 100644 --- a/care/facility/models/asset.py +++ b/care/facility/models/asset.py @@ -178,6 +178,26 @@ def delete(self, *args, **kwargs): AssetBed.objects.filter(asset=self).update(deleted=True) super().delete(*args, **kwargs) + @staticmethod + def has_write_permission(request): + if request.user.asset or request.user.user_type in User.READ_ONLY_TYPES: + return False + return ( + request.user.is_superuser + or request.user.verified + and request.user.user_type >= User.TYPE_VALUE_MAP["Staff"] + ) + + def has_object_write_permission(self, request): + return self.has_write_permission(request) + + @staticmethod + def has_read_permission(request): + return request.user.is_superuser or request.user.verified + + def has_object_read_permission(self, request): + return self.has_read_permission(request) + def __str__(self): return self.name diff --git a/care/facility/models/file_upload.py b/care/facility/models/file_upload.py index 5ac205f82c..2b71c65d46 100644 --- a/care/facility/models/file_upload.py +++ b/care/facility/models/file_upload.py @@ -1,4 +1,5 @@ import time +import uuid from uuid import uuid4 import boto3 @@ -163,5 +164,45 @@ class FileType(models.IntegerChoices): FileTypeChoices = [(x.value, x.name) for x in FileType] FileCategoryChoices = [(x.value, x.name) for x in BaseFileUpload.FileCategory] + def save(self, *args, **kwargs): + from care.facility.models import PatientConsent + + if self.file_type == self.FileType.CONSENT_RECORD: + new_consent = False + if not self.pk and not self.is_archived: + new_consent = True + consent = PatientConsent.objects.filter( + external_id=uuid.UUID(self.associating_id), archived=False + ).first() + consultation = consent.consultation + consent_types = ( + PatientConsent.objects.filter(consultation=consultation, archived=False) + .annotate( + str_external_id=models.functions.Cast( + "external_id", models.CharField() + ) + ) + .annotate( + has_files=( + models.Exists( + FileUpload.objects.filter( + associating_id=models.OuterRef("str_external_id"), + file_type=self.FileType.CONSENT_RECORD, + is_archived=False, + ).exclude(pk=self.pk if self.is_archived else None) + ) + if not new_consent + else models.Value(True) + ) + ) + .filter(has_files=True) + .distinct("type") + .values_list("type", flat=True) + ) + consultation.has_consents = list(consent_types) + consultation.save() + + return super().save(*args, **kwargs) + def __str__(self): return f"{self.FileTypeChoices[self.file_type][1]} - {self.name}{' (Archived)' if self.is_archived else ''}" diff --git a/care/facility/models/patient.py b/care/facility/models/patient.py index 6e8d020f95..b3d42bef01 100644 --- a/care/facility/models/patient.py +++ b/care/facility/models/patient.py @@ -436,6 +436,10 @@ class TestTypeEnum(enum.Enum): objects = BaseManager() + @property + def is_expired(self) -> bool: + return self.death_datetime is not None + def __str__(self): return f"{self.name} - {self.year_of_birth} - {self.get_gender_display()}" diff --git a/care/facility/models/patient_consultation.py b/care/facility/models/patient_consultation.py index 889f2a4067..fa092905ea 100644 --- a/care/facility/models/patient_consultation.py +++ b/care/facility/models/patient_consultation.py @@ -29,6 +29,14 @@ from care.utils.models.base import BaseModel +class ConsentType(models.IntegerChoices): + CONSENT_FOR_ADMISSION = 1, "Consent for Admission" + PATIENT_CODE_STATUS = 2, "Patient Code Status" + CONSENT_FOR_PROCEDURE = 3, "Consent for Procedure" + HIGH_RISK_CONSENT = 4, "High Risk Consent" + OTHERS = 5, "Others" + + class PatientConsultation(PatientBaseModel, ConsultationRelatedPermissionMixin): SUGGESTION_CHOICES = [ (SuggestionChoices.HI, "HOME ISOLATION"), @@ -248,6 +256,11 @@ class PatientConsultation(PatientBaseModel, ConsultationRelatedPermissionMixin): prn_prescription = JSONField(default=dict) discharge_advice = JSONField(default=dict) + has_consents = ArrayField( + models.IntegerField(choices=ConsentType.choices), + default=list, + ) + def get_related_consultation(self): return self @@ -359,14 +372,6 @@ def has_object_generate_discharge_summary_permission(self, request): return self.has_object_read_permission(request) -class ConsentType(models.IntegerChoices): - CONSENT_FOR_ADMISSION = 1, "Consent for Admission" - PATIENT_CODE_STATUS = 2, "Patient Code Status" - CONSENT_FOR_PROCEDURE = 3, "Consent for Procedure" - HIGH_RISK_CONSENT = 4, "High Risk Consent" - OTHERS = 5, "Others" - - class PatientCodeStatusType(models.IntegerChoices): NOT_SPECIFIED = 0, "Not Specified" DNH = 1, "Do Not Hospitalize" @@ -387,7 +392,9 @@ class ConsultationClinician(models.Model): class PatientConsent(BaseModel, ConsultationRelatedPermissionMixin): - consultation = models.ForeignKey(PatientConsultation, on_delete=models.CASCADE) + consultation = models.ForeignKey( + PatientConsultation, on_delete=models.CASCADE, related_name="consents" + ) type = models.IntegerField(choices=ConsentType.choices) patient_code_status = models.IntegerField( choices=PatientCodeStatusType.choices, null=True, blank=True diff --git a/care/facility/tests/test_asset_api.py b/care/facility/tests/test_asset_api.py index 48d934a0f7..a0771b089b 100644 --- a/care/facility/tests/test_asset_api.py +++ b/care/facility/tests/test_asset_api.py @@ -3,6 +3,7 @@ from rest_framework.test import APITestCase from care.facility.models import Asset, Bed +from care.users.models import User from care.utils.assetintegration.asset_classes import AssetClasses from care.utils.tests.test_utils import TestUtils @@ -17,6 +18,11 @@ def setUpTestData(cls) -> None: cls.facility = cls.create_facility(cls.super_user, cls.district, cls.local_body) cls.asset_location = cls.create_asset_location(cls.facility) cls.user = cls.create_user("staff", cls.district, home_facility=cls.facility) + cls.state_admin_ro = cls.create_user( + "stateadmin-ro", + cls.district, + user_type=User.TYPE_VALUE_MAP["StateReadOnlyAdmin"], + ) cls.patient = cls.create_patient( cls.district, cls.facility, local_body=cls.local_body ) @@ -38,6 +44,16 @@ def test_create_asset(self): response = self.client.post("/api/v1/asset/", sample_data) self.assertEqual(response.status_code, status.HTTP_201_CREATED) + def test_create_asset_read_only(self): + sample_data = { + "name": "Test Asset", + "asset_type": 50, + "location": self.asset_location.external_id, + } + self.client.force_authenticate(self.state_admin_ro) + response = self.client.post("/api/v1/asset/", sample_data) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + def test_create_asset_with_warranty_past(self): sample_data = { "name": "Test Asset", diff --git a/care/facility/tests/test_bed_api.py b/care/facility/tests/test_bed_api.py index 92ea35dae4..ce334dd6e4 100644 --- a/care/facility/tests/test_bed_api.py +++ b/care/facility/tests/test_bed_api.py @@ -1,8 +1,8 @@ from rest_framework import status from rest_framework.test import APITestCase -from care.facility.models import Bed -from care.facility.models.bed import AssetBed +from care.facility.models import AssetBed, Bed +from care.users.models import User from care.utils.assetintegration.asset_classes import AssetClasses from care.utils.tests.test_utils import TestUtils @@ -15,30 +15,155 @@ def setUpTestData(cls) -> None: cls.local_body = cls.create_local_body(cls.district) cls.super_user = cls.create_super_user("su", cls.district) cls.facility = cls.create_facility(cls.super_user, cls.district, cls.local_body) + cls.facility2 = cls.create_facility( + cls.super_user, cls.district, cls.local_body + ) cls.asset_location = cls.create_asset_location(cls.facility) + cls.asset_location2 = cls.create_asset_location(cls.facility2) cls.user = cls.create_user("staff", cls.district, home_facility=cls.facility) cls.patient = cls.create_patient( cls.district, cls.facility, local_body=cls.local_body ) - - def setUp(self) -> None: - super().setUp() - self.bed1 = Bed.objects.create( - name="bed1", location=self.asset_location, facility=self.facility + cls.bed1 = Bed.objects.create( + name="bed1", location=cls.asset_location, facility=cls.facility ) - self.bed2 = Bed.objects.create( - name="bed2", location=self.asset_location, facility=self.facility + cls.bed2 = Bed.objects.create( + name="bed2", location=cls.asset_location, facility=cls.facility, bed_type=1 ) - self.bed3 = Bed.objects.create( - name="bed3", location=self.asset_location, facility=self.facility + cls.bed3 = Bed.objects.create( + name="bed3", location=cls.asset_location2, facility=cls.facility2 ) def test_list_beds(self): - # includes 3 queries for auth and 1 for pagination count with self.assertNumQueries(5): response = self.client.get("/api/v1/bed/") self.assertEqual(response.status_code, status.HTTP_200_OK) + # test list beds accessible to user + response = self.client.get("/api/v1/bed/") + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["count"], 2) + + self.client.force_login(self.super_user) + + # test list beds accessible to superuser (all beds) + response = self.client.get("/api/v1/bed/") + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["count"], 3) + + response = self.client.get("/api/v1/bed/", {"bed_type": "ISOLATION"}) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["results"][0]["name"], self.bed2.name) + self.assertEqual(response.data["count"], 1) + + response = self.client.get( + "/api/v1/bed/", + {"facility": self.facility2.external_id}, + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["results"][0]["name"], self.bed3.name) + self.assertEqual(response.data["count"], 1) + + response = self.client.get( + "/api/v1/bed/", + {"location": self.asset_location2.external_id}, + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["count"], 1) + self.assertEqual(response.data["results"][0]["name"], self.bed3.name) + + self.client.logout() + + def test_create_beds(self): + sample_data = { + "name": "Sample Beds", + "bed_type": 2, + "location": self.asset_location.external_id, + "facility": self.facility.external_id, + "number_of_beds": 10, + "description": "This is a sample bed description", + } + response = self.client.post("/api/v1/bed/", sample_data, format="json") + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + self.assertEqual(Bed.objects.filter(bed_type=2).count(), 10) + + # without location + sample_data.update({"location": None}) + response = self.client.post("/api/v1/bed/", sample_data, format="json") + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertEqual(response.data["location"][0], "This field may not be null.") + + # without facility + sample_data.update({"facility": None}) + response = self.client.post("/api/v1/bed/", sample_data, format="json") + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertEqual(response.data["facility"][0], "This field may not be null.") + + # Test - if beds > 100 + sample_data.update({"number_of_beds": 200}) + response = self.client.post("/api/v1/bed/", sample_data, format="json") + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertEqual( + response.data["number_of_beds"][0], + "Cannot create more than 100 beds at once.", + ) + + def test_retrieve_bed(self): + response = self.client.get(f"/api/v1/bed/{self.bed1.external_id}/") + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["name"], "bed1") + + def test_update_bed(self): + updated_data = { + "name": "Updated Bed Name", + "bed_type": 3, + "location": self.asset_location.external_id, + "facility": self.facility.external_id, + } + response = self.client.put( + f"/api/v1/bed/{self.bed2.external_id}/", updated_data, format="json" + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.bed2.refresh_from_db() + self.assertEqual(self.bed2.name, updated_data["name"]) + self.assertEqual(self.bed2.bed_type, updated_data["bed_type"]) + + def test_patch_update_bed(self): + self.client.force_login(self.super_user) + + # we always need to send location and facility since serializer is written that way + partial_data = { + "description": "Updated Bed Description", + "location": self.asset_location.external_id, + "facility": self.facility.external_id, + } + response = self.client.patch( + f"/api/v1/bed/{self.bed3.external_id}/", partial_data, format="json" + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.bed3.refresh_from_db() + self.assertEqual(self.bed3.description, partial_data["description"]) + self.client.logout() + + def test_delete_bed_without_permission(self): + response = self.client.delete(f"/api/v1/bed/{self.bed1.external_id}/") + self.assertFalse(self.user.user_type == User.TYPE_VALUE_MAP["DistrictLabAdmin"]) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertTrue(Bed.objects.filter(id=self.bed1.id).exists()) + + def test_delete_bed(self): + user2 = self.create_user( + "Sample User", + self.district, + home_facility=self.facility, + user_type=User.TYPE_VALUE_MAP["DistrictLabAdmin"], + ) + self.client.force_login(user2) + self.assertTrue(user2.user_type == User.TYPE_VALUE_MAP["DistrictLabAdmin"]) + response = self.client.delete(f"/api/v1/bed/{self.bed1.external_id}/") + self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT) + self.assertFalse(Bed.objects.filter(id=self.bed1.id).exists()) + def test_list_non_occupied_beds(self): linked_bed = Bed.objects.create( name="linked_bed", @@ -50,17 +175,17 @@ def test_list_non_occupied_beds(self): ) AssetBed.objects.create(bed=linked_bed, asset=asset) - # 4 beds 1 linked with HL7MONITOR and 3 created in setup + # 3 beds 1 linked with HL7MONITOR and 2 created in setup [with same facility] response = self.client.get("/api/v1/bed/") - # Assert list returns 4 beds - self.assertEqual(response.json()["count"], 4) + # Assert list returns 3 beds + self.assertEqual(response.json()["count"], 3) response_with_not_occupied_bed = self.client.get( "/api/v1/bed/", {"not_occupied_by_asset_type": "HL7MONITOR"}, ) - # Assert count of unoccupied beds is 3 - self.assertEqual(response_with_not_occupied_bed.json()["count"], 3) + # Assert count of unoccupied beds is 2 (3 in total , 1 occupied) + self.assertEqual(response_with_not_occupied_bed.json()["count"], 2) diff --git a/care/facility/tests/test_patient_api.py b/care/facility/tests/test_patient_api.py index f52f7deb19..3f031d9370 100644 --- a/care/facility/tests/test_patient_api.py +++ b/care/facility/tests/test_patient_api.py @@ -5,11 +5,13 @@ from rest_framework.test import APITestCase from care.facility.models import PatientNoteThreadChoices +from care.facility.models.file_upload import FileUpload from care.facility.models.icd11_diagnosis import ( ConditionVerificationStatus, ICD11Diagnosis, ) from care.facility.models.patient_base import NewDischargeReasonEnum +from care.facility.models.patient_consultation import ConsentType, PatientCodeStatusType from care.utils.tests.test_utils import TestUtils @@ -282,6 +284,150 @@ def test_patient_note_edit(self): self.assertEqual(data[1]["note"], note_content) +class PatientTestCase(TestUtils, APITestCase): + @classmethod + def setUpTestData(cls): + cls.state = cls.create_state() + cls.district = cls.create_district(cls.state) + cls.local_body = cls.create_local_body(cls.district) + cls.super_user = cls.create_super_user("su", cls.district) + cls.facility = cls.create_facility(cls.super_user, cls.district, cls.local_body) + cls.user = cls.create_user( + "doctor1", cls.district, home_facility=cls.facility, user_type=15 + ) + cls.patient = cls.create_patient(cls.district, cls.facility) + cls.consultation = cls.create_consultation( + patient_no="IP5678", + patient=cls.patient, + facility=cls.facility, + created_by=cls.user, + suggestion="A", + encounter_date=now(), + ) + cls.patient_2 = cls.create_patient(cls.district, cls.facility) + cls.consultation_2 = cls.create_consultation( + patient_no="IP5679", + patient=cls.patient_2, + facility=cls.facility, + created_by=cls.user, + suggestion="A", + encounter_date=now(), + ) + + cls.consent = cls.create_patient_consent( + cls.consultation, + created_by=cls.user, + type=ConsentType.CONSENT_FOR_ADMISSION, + patient_code_status=None, + ) + cls.file = FileUpload.objects.create( + internal_name="test.pdf", + file_type=FileUpload.FileType.CONSENT_RECORD, + name="Test File", + associating_id=str(cls.consent.external_id), + file_category=FileUpload.FileCategory.UNSPECIFIED, + ) + + def get_base_url(self) -> str: + return "/api/v1/patient/" + + def test_has_consent(self): + self.client.force_authenticate(user=self.user) + response = self.client.get(self.get_base_url()) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["count"], 2) + patient_1_response = [ + x + for x in response.data["results"] + if x["id"] == str(self.patient.external_id) + ][0] + patient_2_response = [ + x + for x in response.data["results"] + if x["id"] == str(self.patient_2.external_id) + ][0] + self.assertEqual( + patient_1_response["last_consultation"]["has_consents"], + [ConsentType.CONSENT_FOR_ADMISSION], + ) + self.assertEqual(patient_2_response["last_consultation"]["has_consents"], []) + + def test_consent_edit(self): + self.file.name = "Test File 1 Edited" + self.file.save() + self.client.force_authenticate(user=self.user) + response = self.client.get(self.get_base_url()) + self.assertEqual(response.status_code, status.HTTP_200_OK) + patient_1_response = [ + x + for x in response.data["results"] + if x["id"] == str(self.patient.external_id) + ][0] + self.assertEqual( + patient_1_response["last_consultation"]["has_consents"], + [ConsentType.CONSENT_FOR_ADMISSION], + ) + + def test_has_consents_archived(self): + self.client.force_authenticate(user=self.user) + consent = self.create_patient_consent( + self.consultation_2, + created_by=self.user, + type=ConsentType.HIGH_RISK_CONSENT, + patient_code_status=None, + ) + file = FileUpload.objects.create( + internal_name="test.pdf", + file_type=FileUpload.FileType.CONSENT_RECORD, + name="Test File", + associating_id=str(consent.external_id), + file_category=FileUpload.FileCategory.UNSPECIFIED, + ) + response = self.client.get(self.get_base_url()) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["count"], 2) + patient_1_response = [ + x + for x in response.data["results"] + if x["id"] == str(self.patient.external_id) + ][0] + patient_2_response = [ + x + for x in response.data["results"] + if x["id"] == str(self.patient_2.external_id) + ][0] + self.assertEqual( + patient_1_response["last_consultation"]["has_consents"], + [ConsentType.CONSENT_FOR_ADMISSION], + ) + self.assertEqual( + patient_2_response["last_consultation"]["has_consents"], + [ConsentType.HIGH_RISK_CONSENT], + ) + + file.is_archived = True + file.save() + + response = self.client.get(self.get_base_url()) + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data["count"], 2) + patient_1_response = [ + x + for x in response.data["results"] + if x["id"] == str(self.patient.external_id) + ][0] + patient_2_response = [ + x + for x in response.data["results"] + if x["id"] == str(self.patient_2.external_id) + ][0] + self.assertEqual( + patient_1_response["last_consultation"]["has_consents"], + [ConsentType.CONSENT_FOR_ADMISSION], + ) + self.assertEqual(patient_2_response["last_consultation"]["has_consents"], []) + + class PatientFilterTestCase(TestUtils, APITestCase): @classmethod def setUpTestData(cls): @@ -331,6 +477,55 @@ def setUpTestData(cls): verification_status=ConditionVerificationStatus.UNCONFIRMED, ) + cls.consent = cls.create_patient_consent( + cls.consultation, + created_by=cls.user, + type=1, + patient_code_status=None, + ) + + cls.patient_2 = cls.create_patient(cls.district, cls.facility) + cls.consultation_2 = cls.create_consultation( + patient_no="IP5679", + patient=cls.patient_2, + facility=cls.facility, + created_by=cls.user, + suggestion="A", + encounter_date=now(), + ) + cls.consent2 = cls.create_patient_consent( + cls.consultation_2, + created_by=cls.user, + type=ConsentType.PATIENT_CODE_STATUS, + patient_code_status=PatientCodeStatusType.ACTIVE_TREATMENT, + ) + + cls.patient_3 = cls.create_patient(cls.district, cls.facility) + cls.consultation_3 = cls.create_consultation( + patient_no="IP5680", + patient=cls.patient_3, + facility=cls.facility, + created_by=cls.user, + suggestion="A", + encounter_date=now(), + ) + + FileUpload.objects.create( + internal_name="test.pdf", + file_type=FileUpload.FileType.CONSENT_RECORD, + name="Test File", + associating_id=str(cls.consent.external_id), + file_category=FileUpload.FileCategory.UNSPECIFIED, + ) + + FileUpload.objects.create( + internal_name="test.pdf", + file_type=FileUpload.FileType.CONSENT_RECORD, + name="Test File", + associating_id=str(cls.consent2.external_id), + file_category=FileUpload.FileCategory.UNSPECIFIED, + ) + def get_base_url(self) -> str: return "/api/v1/patient/" @@ -353,10 +548,8 @@ def test_filter_by_location(self): }, ) self.assertEqual(response.status_code, status.HTTP_200_OK) - self.assertEqual(response.data["count"], 1) - self.assertEqual( - response.data["results"][0]["id"], str(self.patient.external_id) - ) + self.assertEqual(response.data["count"], 3) + self.assertContains(response, str(self.patient.external_id)) def test_filter_by_diagnoses(self): self.client.force_authenticate(user=self.user) @@ -431,6 +624,44 @@ def test_filter_by_review_missed(self): else: self.assertIsNone(patient["review_time"]) + def test_filter_by_has_consents(self): + + choices = ["1", "2", "3", "4", "5", "None"] + + self.client.force_authenticate(user=self.user) + res = self.client.get( + self.get_base_url(), {"last_consultation__consent_types": choices[5]} + ) + + self.assertEqual(res.status_code, status.HTTP_200_OK) + self.assertEqual(res.json()["count"], 1) + self.assertContains(res, self.patient_3.external_id) + + res = self.client.get( + self.get_base_url(), + {"last_consultation__consent_types": ",".join(choices[:4])}, + ) + + self.assertEqual(res.status_code, status.HTTP_200_OK) + self.assertEqual(res.json()["count"], 2) + self.assertContains(res, self.patient.external_id) + self.assertContains(res, self.patient_2.external_id) + + res = self.client.get( + self.get_base_url(), {"last_consultation__consent_types": choices[0]} + ) + + self.assertEqual(res.status_code, status.HTTP_200_OK) + self.assertEqual(res.json()["count"], 1) + self.assertContains(res, self.patient.external_id) + + res = self.client.get( + self.get_base_url(), {"last_consultation__consent_types": ",".join(choices)} + ) + + self.assertEqual(res.status_code, status.HTTP_200_OK) + self.assertEqual(res.json()["count"], 3) + class PatientTransferTestCase(TestUtils, APITestCase): @classmethod @@ -509,6 +740,31 @@ def test_transfer_with_active_consultation_same_facility(self): "Patient transfer cannot be completed because the patient has an active consultation in the same facility", ) + def test_transfer_with_expired_patient(self): + # Mocking discharged as expired + self.consultation.new_discharge_reason = NewDischargeReasonEnum.EXPIRED + self.consultation.death_datetime = now() + self.consultation.save() + + # Set the patient's facility to allow transfers + self.patient.allow_transfer = True + self.patient.save() + + # Ensure transfer fails if the patient has an active consultation + self.client.force_authenticate(user=self.super_user) + response = self.client.post( + f"/api/v1/patient/{self.patient.external_id}/transfer/", + { + "year_of_birth": 1992, + "facility": self.facility.external_id, + }, + ) + self.assertEqual(response.status_code, status.HTTP_406_NOT_ACCEPTABLE) + self.assertEqual( + response.data["Patient"], + "Patient transfer cannot be completed because the patient is expired", + ) + def test_transfer_disallowed_by_facility(self): # Set the patient's facility to disallow transfers self.patient.allow_transfer = False