migration: skip purging data, handle exceptions; add tests

care/facility/api/serializers/bed.py

@@ -123,11 +123,14 @@ def validate(self, attrs):
                     {"asset": "Should be in the same facility as the bed"}
                 )
             if (
-                asset.asset_class == AssetClasses.HL7MONITOR.name
-                and AssetBed.objects.filter(
-                    bed=bed, asset__asset_class=asset.asset_class
-                ).exists()
-            ):
+                asset.asset_class
+                in [
+                    AssetClasses.HL7MONITOR.name,
+                    AssetClasses.ONVIF.name,
+                ]
+            ) and AssetBed.objects.filter(
+                bed=bed, asset__asset_class=asset.asset_class
+            ).exists():
                 raise ValidationError(
                     {
                         "asset": "Bed is already in use by another asset of the same class"

care/facility/migrations/0466_camera_presets.py

@@ -47,32 +47,30 @@ def backfill_camera_presets(apps, schema_editor):
                 name = asset_bed.meta.get("preset_name")
 
                 if position := asset_bed.meta.get("position"):
-                    presets_to_create.append(
-                        CameraPreset(
-                            name=name,
-                            asset_bed=AssetBed.objects.filter(
-                                asset=asset_bed.asset, bed=asset_bed.bed
-                            ).order_by("id")[0],
-                            position={
-                                "x": position["x"],
-                                "y": position["y"],
-                                "zoom": position["zoom"],
-                            },
-                            is_migrated=True,
+                    try:
+                        presets_to_create.append(
+                            CameraPreset(
+                                name=name,
+                                asset_bed=AssetBed.objects.filter(
+                                    asset=asset_bed.asset, bed=asset_bed.bed
+                                ).order_by("id")[0],
+                                position={
+                                    "x": float(position["x"]),
+                                    "y": float(position["y"]),
+                                    "zoom": float(position["zoom"]),
+                                },
+                                is_migrated=True,
+                            )
                         )
-                    )
+                    except:
+                        pass
                     if asset_bed.row_number != 1:
                         assetbeds_to_delete.append(asset_bed.id)
                 else:
                     assetbeds_to_delete.append(asset_bed.id)
 
             CameraPreset.objects.bulk_create(presets_to_create)
-            AssetBed.objects.filter(id__in=assetbeds_to_delete).update(
-                deleted=True, meta={}
-            )
-        AssetBed.objects.filter(deleted=False, asset__asset_class="ONVIF").update(
-            meta={}
-        )
+            AssetBed.objects.filter(id__in=assetbeds_to_delete).update(deleted=True)
 
     operations = [
         migrations.CreateModel(

care/facility/tests/test_asset_bed_api.py

@@ -0,0 +1,183 @@
+from rest_framework import status
+from rest_framework.test import APITestCase
+
+from care.users.models import User
+from care.utils.assetintegration.asset_classes import AssetClasses
+from care.utils.tests.test_utils import TestUtils
+
+
+class AssetBedViewSetTestCase(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(
+            User.TYPE_VALUE_MAP["DistrictAdmin"],
+            cls.district,
+            home_facility=cls.facility,
+        )
+        cls.asset_location = cls.create_asset_location(cls.facility)
+        cls.asset = cls.create_asset(cls.asset_location)
+        cls.camera_asset = cls.create_asset(
+            cls.asset_location, asset_class=AssetClasses.ONVIF.name
+        )
+        cls.bed = cls.create_bed(cls.facility, cls.asset_location)
+
+    def test_link_disallowed_asset_class_asset_to_bed(self):
+        data = {
+            "asset": self.asset.external_id,
+            "bed": self.bed.external_id,
+        }
+        res = self.client.post("/api/v1/assetbed/", data)
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_link_asset_to_bed_and_attempt_duplicate_linking(self):
+        data = {
+            "asset": self.camera_asset.external_id,
+            "bed": self.bed.external_id,
+        }
+        res = self.client.post("/api/v1/assetbed/", data)
+        self.assertEqual(res.status_code, status.HTTP_201_CREATED)
+        # Attempt linking same camera to the same bed again.
+        res = self.client.post("/api/v1/assetbed/", data)
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
+        # List asset beds filtered by asset and bed ID and check only 1 result exists
+        res = self.client.get("/api/v1/assetbed/", data)
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        self.assertEqual(res.data["count"], 1)
+
+
+class AssetBedCameraPresetViewSetTestCase(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(
+            User.TYPE_VALUE_MAP["DistrictAdmin"],
+            cls.district,
+            home_facility=cls.facility,
+        )
+        cls.asset_location = cls.create_asset_location(cls.facility)
+        cls.asset1 = cls.create_asset(
+            cls.asset_location, asset_class=AssetClasses.ONVIF.name
+        )
+        cls.asset2 = cls.create_asset(
+            cls.asset_location, asset_class=AssetClasses.ONVIF.name
+        )
+        cls.bed = cls.create_bed(cls.facility, cls.asset_location)
+        cls.asset_bed1 = cls.create_asset_bed(cls.asset1, cls.bed)
+        cls.asset_bed2 = cls.create_asset_bed(cls.asset2, cls.bed)
+
+    def get_base_url(self, asset_bed_id=None):
+        return f"/api/v1/assetbed/{asset_bed_id or self.asset_bed1.external_id}/camera_presets/"
+
+    def test_create_camera_preset_without_position(self):
+        res = self.client.post(
+            self.get_base_url(),
+            {
+                "name": "Preset without position",
+                "position": {},
+            },
+            format="json",
+        )
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_create_camera_preset_with_missing_required_keys_in_position(self):
+        res = self.client.post(
+            self.get_base_url(),
+            {
+                "name": "Preset with invalid position",
+                "position": {"key": "value"},
+            },
+            format="json",
+        )
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_create_camera_preset_with_position_not_number(self):
+        res = self.client.post(
+            self.get_base_url(),
+            {
+                "name": "Preset with invalid position",
+                "position": {
+                    "x": "not a number",
+                    "y": 1,
+                    "zoom": 1,
+                },
+            },
+            format="json",
+        )
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_create_camera_preset_with_position_values_as_string(self):
+        res = self.client.post(
+            self.get_base_url(),
+            {
+                "name": "Preset with invalid position",
+                "position": {
+                    "x": "1",
+                    "y": "1",
+                    "zoom": "1",
+                },
+            },
+            format="json",
+        )
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_create_camera_preset_and_presence_in_various_preset_list_apis(self):
+        asset_bed = self.asset_bed1
+        res = self.client.post(
+            self.get_base_url(asset_bed.external_id),
+            {
+                "name": "Preset with proper position",
+                "position": {
+                    "x": 1.0,
+                    "y": 1.0,
+                    "zoom": 1.0,
+                },
+            },
+            format="json",
+        )
+        self.assertEqual(res.status_code, status.HTTP_201_CREATED)
+        preset_external_id = res.data["id"]
+
+        # Check if preset in asset-bed preset list
+        res = self.client.get(self.get_base_url(asset_bed.external_id))
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        self.assertContains(res, preset_external_id)
+
+        # Check if preset in asset preset list
+        res = self.client.get(
+            f"/api/v1/asset/{asset_bed.asset.external_id}/camera_presets/"
+        )
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        self.assertContains(res, preset_external_id)
+
+        # Check if preset in bed preset list
+        res = self.client.get(
+            f"/api/v1/bed/{asset_bed.bed.external_id}/camera_presets/"
+        )
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        self.assertContains(res, preset_external_id)
+
+    def test_create_camera_preset_with_same_name_in_same_bed(self):
+        data = {
+            "name": "Duplicate Preset Name",
+            "position": {
+                "x": 1.0,
+                "y": 1.0,
+                "zoom": 1.0,
+            },
+        }
+        self.client.post(
+            self.get_base_url(self.asset_bed1.external_id), data, format="json"
+        )
+        res = self.client.post(
+            self.get_base_url(self.asset_bed2.external_id), data, format="json"
+        )
+        self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)

care/utils/tests/test_utils.py

@@ -38,7 +38,7 @@
     Ward,
 )
 from care.facility.models.asset import Asset, AssetLocation
-from care.facility.models.bed import Bed, ConsultationBed
+from care.facility.models.bed import AssetBed, Bed, ConsultationBed
 from care.facility.models.facility import FacilityUser
 from care.facility.models.icd11_diagnosis import (
     ConditionVerificationStatus,
@@ -446,6 +446,12 @@ def create_bed(cls, facility: Facility, location: AssetLocation, **kwargs):
         data.update(kwargs)
         return Bed.objects.create(**data)
 
+    @classmethod
+    def create_asset_bed(cls, asset: Asset, bed: Bed, **kwargs):
+        data = {"asset": asset, "bed": bed}
+        data.update(kwargs)
+        return AssetBed.objects.create(**data)
+
     @classmethod
     def create_consultation_bed(
         cls,