Privacy consultation bed

care/facility/api/viewsets/asset.py

@@ -20,7 +20,7 @@
 from rest_framework import exceptions, status
 from rest_framework import filters as drf_filters
 from rest_framework.decorators import action
-from rest_framework.exceptions import APIException, ValidationError
+from rest_framework.exceptions import APIException, PermissionDenied, ValidationError
 from rest_framework.mixins import (
     CreateModelMixin,
     DestroyModelMixin,
@@ -405,12 +405,18 @@ def operate_assets(self, request, *args, **kwargs):
                     "middleware_hostname": middleware_hostname,
                 }
             )
-            result = asset_class.handle_action(action)
+            result = asset_class.handle_action(action, request.user)
             return Response({"result": result}, status=status.HTTP_200_OK)
 
         except ValidationError as e:
             return Response({"detail": e.detail}, status=status.HTTP_400_BAD_REQUEST)
 
+        except PermissionDenied as e:
+            return Response(
+                {**e.detail},
+                status=status.HTTP_403_FORBIDDEN,
+            )
+
         except KeyError as e:
             return Response(
                 {"message": {key: "is required" for key in e.args}},

care/facility/api/viewsets/bed.py

@@ -5,6 +5,7 @@
 from drf_spectacular.utils import extend_schema, extend_schema_view
 from rest_framework import filters as drf_filters
 from rest_framework import status
+from rest_framework.decorators import action
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.exceptions import ValidationError as DRFValidationError
 from rest_framework.fields import get_error_detail
@@ -230,3 +231,30 @@ def get_queryset(self):
             allowed_facilities = get_accessible_facilities(user)
             queryset = queryset.filter(bed__facility__id__in=allowed_facilities)
         return queryset
+
+    @action(detail=True, methods=["patch"])
+    def set_privacy(self, request, external_id):
+        consultation_bed: ConsultationBed = self.get_object()
+
+        is_privacy_enabled = request.data.get("is_privacy_enabled")
+
+        if is_privacy_enabled is None and not isinstance(is_privacy_enabled, bool):
+            return Response(
+                {"detail": "is_privacy_enabled is required and should be a boolean"},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        if not consultation_bed.consultation.patient.has_object_update_permission(
+            request
+        ) or request.user.user_type in [
+            User.TYPE_VALUE_MAP["DistrictLabAdmin"],
+            User.TYPE_VALUE_MAP["StateLabAdmin"],
+        ]:
+            raise PermissionDenied
+
+        consultation_bed.is_privacy_enabled = is_privacy_enabled
+        consultation_bed.save(update_fields=["is_privacy_enabled"])
+
+        return Response(
+            ConsultationBedSerializer(consultation_bed).data, status=status.HTTP_200_OK
+        )

care/facility/migrations/0467_consultationbed_is_privacy_enabled.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.1 on 2024-09-28 17:40
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('facility', '0466_camera_presets'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='consultationbed',
+            name='is_privacy_enabled',
+            field=models.BooleanField(default=False),
+        ),
+    ]

care/facility/models/bed.py

@@ -92,6 +92,7 @@ class ConsultationBed(BaseModel):
     bed = models.ForeignKey(Bed, on_delete=models.PROTECT, null=False, blank=False)
     start_date = models.DateTimeField(null=False, blank=False)
     end_date = models.DateTimeField(null=True, blank=True, default=None)
+    is_privacy_enabled = models.BooleanField(default=False)
     meta = JSONField(default=dict, blank=True)
     assets = models.ManyToManyField(
         Asset, through="ConsultationBedAsset", related_name="assigned_consultation_beds"

care/facility/tests/test_asset_usage_manager.py

@@ -0,0 +1,118 @@
+from unittest import mock
+
+from rest_framework.test import APITestCase
+
+from care.users.models import User
+from care.utils.assetintegration.usage_manager import UsageManager
+from care.utils.tests.test_utils import TestUtils
+
+
+class UsageManagerTestCase(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.asset_location = cls.create_asset_location(cls.facility)
+        cls.asset = cls.create_asset(cls.asset_location)
+
+    def setUp(self):
+        self.user1 = self.create_user(
+            username="test_user_1",
+            state=self.state,
+            district=self.district,
+            user_type=User.TYPE_VALUE_MAP["StateAdmin"],
+        )
+        self.user2 = self.create_user(
+            username="test_user_2",
+            state=self.state,
+            district=self.district,
+            user_type=User.TYPE_VALUE_MAP["StateAdmin"],
+        )
+
+        self.mock_cache = mock.MagicMock()
+        self.cache_patcher = mock.patch(
+            "care.utils.assetintegration.usage_manager.cache", self.mock_cache
+        )
+        self.cache_patcher.start()
+
+        self.usage_manager_user1 = UsageManager(
+            asset_id=self.asset.external_id, user=self.user1
+        )
+        self.usage_manager_user2 = UsageManager(
+            asset_id=self.asset.external_id, user=self.user2
+        )
+
+        self.mock_redis_client = mock.MagicMock()
+        self.usage_manager_user1.redis_client = self.mock_redis_client
+        self.usage_manager_user2.redis_client = self.mock_redis_client
+
+    def tearDown(self):
+        self.cache_patcher.stop()
+
+    def test_has_access(self):
+        self.mock_cache.get.return_value = None
+        self.assertTrue(self.usage_manager_user1.has_access())
+
+        self.mock_cache.get.return_value = self.user1.id
+        self.assertTrue(self.usage_manager_user1.has_access())
+
+        self.mock_cache.get.return_value = self.user2.id
+        self.assertFalse(self.usage_manager_user1.has_access())
+
+    def test_unlock_camera(self):
+        self.mock_cache.get.return_value = self.user1.id
+
+        with mock.patch.object(
+            self.usage_manager_user1, "notify_waiting_list_on_asset_availabe"
+        ) as mock_notify:
+            self.usage_manager_user1.unlock_camera()
+
+            self.mock_cache.delete.assert_called_once_with(
+                self.usage_manager_user1.current_user_cache_key
+            )
+
+            mock_notify.assert_called_once()
+
+    def test_request_access(self):
+        self.mock_cache.get.return_value = None
+        self.assertTrue(self.usage_manager_user1.request_access())
+
+        self.mock_cache.get.return_value = self.user2.id
+        with mock.patch(
+            "care.utils.notification_handler.send_webpush"
+        ) as mock_send_webpush:
+            result = self.usage_manager_user1.request_access()
+            self.assertFalse(result)
+            mock_send_webpush.assert_called_once()
+
+    def test_lock_camera(self):
+        self.mock_cache.get.return_value = None
+        self.assertTrue(self.usage_manager_user1.lock_camera())
+        self.mock_cache.set.assert_called_once_with(
+            self.usage_manager_user1.current_user_cache_key,
+            self.user1.id,
+            timeout=60 * 5,
+        )
+
+        self.mock_cache.get.return_value = self.user2.id
+        self.assertFalse(self.usage_manager_user1.lock_camera())
+
+    def test_current_user(self):
+        self.mock_cache.get.return_value = self.user1.id
+
+        mock_serializer = mock.MagicMock()
+        mock_serializer.data = {
+            "id": self.user1.id,
+            "username": self.user1.username,
+        }
+
+        with mock.patch(
+            "care.facility.api.serializers.asset.UserBaseMinimumSerializer",
+            return_value=mock_serializer,
+        ):
+            current_user_data = self.usage_manager_user1.current_user()
+            self.assertIsNotNone(current_user_data)
+            self.assertEqual(current_user_data["id"], self.user1.id)

care/facility/tests/test_consultation_bed_api.py

@@ -4,8 +4,9 @@
 from rest_framework import status
 from rest_framework.test import APITestCase
 
-from care.facility.models import Asset, Bed, ConsultationBedAsset
+from care.facility.models import Asset, Bed, ConsultationBedAsset, FacilityUser
 from care.facility.models.bed import ConsultationBed
+from care.users.models import User
 from care.utils.tests.test_utils import TestUtils
 
 
@@ -427,3 +428,77 @@ def test_link_asset_to_consultation_bed_with_asset_already_in_use(self):
             },
         )
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_set_privacy_endpoint(self):
+        consultation_bed = ConsultationBed.objects.create(
+            consultation=self.consultation,
+            bed=self.bed1,
+            start_date=timezone.now(),
+            end_date=timezone.now() + timedelta(days=1),
+            is_privacy_enabled=False,
+        )
+
+        response = self.client.patch(
+            f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/",
+            {},
+        )
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+        allowed_user_types = [
+            "StateAdmin",
+            "DistrictAdmin",
+            "Doctor",
+        ]
+        for user_type in allowed_user_types:
+            user = self.create_user(
+                username="test_user_" + user_type.lower(),
+                state=self.state,
+                district=self.district,
+                user_type=User.TYPE_VALUE_MAP[user_type],
+                home_facility=self.facility,
+            )
+
+            self.client.force_authenticate(user)
+
+            response = self.client.patch(
+                f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/",
+                {"is_privacy_enabled": True},
+            )
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            self.assertTrue(
+                ConsultationBed.objects.get(id=consultation_bed.id).is_privacy_enabled
+            )
+
+            response = self.client.patch(
+                f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/",
+                {"is_privacy_enabled": False},
+            )
+            self.assertEqual(response.status_code, status.HTTP_200_OK)
+            self.assertFalse(
+                ConsultationBed.objects.get(id=consultation_bed.id).is_privacy_enabled
+            )
+
+        not_allowed_user_types = [
+            user_type
+            for user_type in User.TYPE_VALUE_MAP
+            if user_type not in allowed_user_types
+        ]
+        for user_type in not_allowed_user_types:
+            user = self.create_user(
+                username="test_user_" + user_type.lower(),
+                district=self.district,
+                user_type=User.TYPE_VALUE_MAP[user_type],
+            )
+            FacilityUser.objects.update_or_create(
+                user=user,
+                facility=self.facility,
+                defaults={"created_by": self.super_user},
+            )
+
+            self.client.force_authenticate(user)
+
+            response = self.client.patch(
+                f"/api/v1/consultationbed/{consultation_bed.external_id}/set_privacy/",
+                {"is_privacy_enabled": True},
+            )
+            self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

care/utils/assetintegration/base.py

@@ -19,7 +19,7 @@ def __init__(self, meta):
         self.insecure_connection = self.meta.get("insecure_connection", False)
         self.timeout = settings.MIDDLEWARE_REQUEST_TIMEOUT
 
-    def handle_action(self, action):
+    def handle_action(self, action, user):
         pass
 
     def get_url(self, endpoint):

care/utils/assetintegration/hl7monitor.py

@@ -20,7 +20,7 @@ def __init__(self, meta):
                 {key: f"{key} not found in asset metadata" for key in e.args}
             ) from e
 
-    def handle_action(self, action):
+    def handle_action(self, action, user):
         action_type = action["type"]
 
         if action_type == self.HL7MonitorActions.GET_VITALS.value: