fix asset class initialization

care/facility/api/viewsets/asset.py

@@ -80,7 +80,7 @@
 
 
 @receiver(post_save, sender=Asset)
 def delete_asset_cache(sender, instance, created, **kwargs):
     cache.delete("asset:" + str(instance.external_id))
     cache.delete("asset:qr:" + str(instance.qr_code_id))
     cache.delete("asset:qr:" + str(instance.id))
@@ -89,7 +89,7 @@
 class AssetLocationFilter(filters.FilterSet):
     bed_is_occupied = filters.BooleanFilter(method="filter_bed_is_occupied")
 
     def filter_bed_is_occupied(self, queryset, name, value):
         asset_locations = (
             AssetBed.objects.select_related("asset", "bed")
             .filter(asset__asset_class=AssetClasses.HL7MONITOR.name)
@@ -135,7 +135,7 @@
         return context
 
     def get_queryset(self):
         user = self.request.user
         queryset = self.queryset
         if user.is_superuser:
             pass
@@ -186,7 +186,7 @@
     is_permanent = filters.BooleanFilter(method="filter_is_permanent")
     warranty_amc_end_of_validity = filters.DateFromToRangeFilter()
 
     def filter_in_use_by_consultation(self, queryset, _, value):
         if value not in EMPTY_VALUES:
             queryset = queryset.annotate(
                 is_in_use=Exists(
@@ -200,7 +200,7 @@
             queryset = queryset.filter(is_in_use=value)
         return queryset.distinct()
 
     def filter_is_permanent(self, queryset, _, value):
         if value not in EMPTY_VALUES:
             if value:
                 queryset = queryset.filter(
@@ -226,7 +226,7 @@
     permission_classes = ()
     authentication_classes = ()
 
     def retrieve(self, request, *args, **kwargs):
         key = "asset:" + kwargs["external_id"]
         hit = cache.get(key)
         if not hit:
@@ -246,7 +246,7 @@
     permission_classes = ()
     authentication_classes = ()
 
     def retrieve(self, request, *args, **kwargs):
         qr_code_id = kwargs["qr_code_id"]
         key = "asset:qr:" + qr_code_id
         hit = cache.get(key)
@@ -349,7 +349,7 @@
         responses={200: UserDefaultAssetLocationSerializer()}, tags=["asset"]
     )
     @action(detail=False, methods=["GET"])
     def get_default_user_location(self, request, *args, **kwargs):
         obj = get_object_or_404(
             UserDefaultAssetLocation.objects.filter(user=request.user)
         )
@@ -363,7 +363,7 @@
         tags=["asset"],
     )
     @action(detail=False, methods=["POST"])
     def set_default_user_location(self, request, *args, **kwargs):
         if "location" not in request.data:
             raise ValidationError({"location": "is required"})
         allowed_locations = get_asset_location_queryset(request.user)
@@ -385,7 +385,7 @@
         tags=["asset"],
     )
     @action(detail=True, methods=["POST"])
     def operate_assets(self, request, *args, **kwargs):
         """
         This API is used to operate assets. API accepts the asset_id and action as parameters.
         """
@@ -401,7 +401,7 @@
             asset_class: BaseAssetIntegration = AssetClasses[asset.asset_class].value(
                 {
                     **asset.meta,
-                    "id": asset.external_id,
+                    "id": str(asset.external_id),
                     "middleware_hostname": middleware_hostname,
                 }
             )
@@ -447,7 +447,7 @@
             )
         ],
     )
     def list(self, request, *args, **kwargs):
         """
         This API is used by the middleware to retrieve assets and their configurations
         for a given facility and middleware hostname.

care/facility/tasks/asset_monitor.py

@@ -65,7 +65,7 @@ def check_asset_status():  # noqa: PLR0912
                 ].value(
                     {
                         **asset.meta,
-                        "id": asset.external_id,
+                        "id": str(asset.external_id),
                         "middleware_hostname": resolved_middleware,
                     }
                 )

care/facility/tests/test_asset_api.py

@@ -6,6 +6,7 @@
 from care.facility.models import Asset, Bed
 from care.users.models import User
 from care.utils.assetintegration.asset_classes import AssetClasses
+from care.utils.assetintegration.base import BaseAssetIntegration
 from care.utils.assetintegration.hl7monitor import HL7MonitorAsset
 from care.utils.assetintegration.onvif import OnvifAsset
 from care.utils.assetintegration.ventilator import VentilatorAsset
@@ -14,8 +15,8 @@
 
 class AssetViewSetTestCase(TestUtils, APITestCase):
     @classmethod
     def setUpTestData(cls) -> None:
         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)
@@ -33,12 +34,32 @@
 
     def setUp(self) -> None:
         super().setUp()
         self.asset = self.create_asset(self.asset_location)
 
     def validate_invalid_meta(self, asset_class, meta):
         with self.assertRaises(ValidationError):
             asset_class(meta)
 
+    def test_asset_class_initialization(self):
+        asset = self.create_asset(
+            self.asset_location,
+            asset_class=AssetClasses.ONVIF.name,
+            meta={
+                "local_ip_address": "192.168.0.1",
+                "camera_access_key": "username:password:access_key",
+                "middleware_hostname": "middleware.local",
+                "insecure_connection": True,
+            },
+        )
+        asset_class = AssetClasses[asset.asset_class].value(
+            {
+                **asset.meta,
+                "id": str(asset.external_id),
+                "middleware_hostname": "middleware.local",
+            }
+        )
+        self.assertIsInstance(asset_class, BaseAssetIntegration)
+
     def test_meta_validations_for_onvif_asset(self):
         valid_meta = {
             "local_ip_address": "192.168.0.1",
@@ -325,8 +346,8 @@
 
 class AssetConfigValidationTestCase(TestUtils, APITestCase):
     @classmethod
     def setUpTestData(cls) -> None:
         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)
@@ -341,7 +362,7 @@
         cls.user = cls.create_user("staff", cls.district, home_facility=cls.facility)
 
     def test_create_asset_with_unique_ip(self):
         sample_data = {
             "name": "Test Asset",
             "asset_type": 50,
             "location": self.asset_location.external_id,
@@ -358,7 +379,7 @@
             asset_class=AssetClasses.HL7MONITOR.name,
             meta={"local_ip_address": "192.168.1.14"},
         )
         sample_data = {
             "name": "Test Asset",
             "asset_type": 50,
             "location": self.asset_location.external_id,
@@ -379,7 +400,7 @@
             asset_class=AssetClasses.HL7MONITOR.name,
             meta={"local_ip_address": "192.168.1.14"},
         )
         sample_data = {
             "name": "Test Asset",
             "asset_type": 50,
             "location": test_location.external_id,