Merge Develop To Staging v24.48.0

.github/workflows/qodana_code_quality.yml

@@ -0,0 +1,27 @@
+name: Qodana
+on:
+  workflow_dispatch:
+  pull_request:
+  push:
+    branches: # Specify your branches here
+      - develop
+
+jobs:
+  qodana:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
+      checks: write
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}  # to check out the actual pull request commit, not the merge commit
+          fetch-depth: 0  # a full history is required for pull request analysis
+      - name: 'Qodana Scan'
+        uses: JetBrains/qodana-action@v2024.2
+        with:
+          pr-mode: false
+        env:
+          QODANA_TOKEN: ${{ secrets.QODANA_TOKEN_1210838162 }}
+          QODANA_ENDPOINT: 'https://qodana.cloud'

care/facility/api/serializers/asset.py

@@ -7,7 +7,7 @@
 from django.utils.timezone import now
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import PermissionDenied, ValidationError
 from rest_framework.serializers import (
     CharField,
     JSONField,
@@ -46,13 +46,13 @@
     id = UUIDField(source="external_id", read_only=True)
     location_type = ChoiceField(choices=AssetLocation.RoomTypeChoices)
 
     def validate_middleware_address(self, value):
         value = (value or "").strip()
         if not value:
             return value
 
         # Check if the address is valid
         MiddlewareDomainAddressValidator()(value)
         return value
 
     def validate(self, data):
@@ -174,11 +174,14 @@
 
             facilities = get_facility_queryset(user)
             if not facilities.filter(id=location.facility.id).exists():
-                raise PermissionError
+                error_message = (
+                    "You do not have permission to access this facility's asset."
+                )
+                raise PermissionDenied(error_message)
             del attrs["location"]
             attrs["current_location"] = location
 
-        # validate that warraty date is not in the past
+        # validate that warranty date is not in the past
         if warranty_amc_end_of_validity := attrs.get("warranty_amc_end_of_validity"):
             # pop out warranty date if it is not changed
             if (
@@ -391,10 +394,10 @@
         model = AvailabilityRecord
         fields = ("status", "timestamp", "linked_id", "linked_model")
 
     def get_linked_id(self, obj):
         return obj.object_external_id
 
     def get_linked_model(self, obj):
         return obj.content_type.model
 
 
@@ -407,7 +410,7 @@
 
 
 class AssetActionSerializer(Serializer):
     def action_choices():
         actions = [
             OnvifAsset.OnvifActions,
             HL7MonitorAsset.HL7MonitorActions,

care/facility/api/serializers/bed.py

@@ -2,7 +2,7 @@
 from django.db.models import Exists, OuterRef, Q
 from django.shortcuts import get_object_or_404
 from django.utils import timezone
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import PermissionDenied, ValidationError
 from rest_framework.serializers import (
     BooleanField,
     CharField,
@@ -46,10 +46,10 @@
 
     number_of_beds = IntegerField(required=False, default=1, write_only=True)
 
     def validate_name(self, value):
         return value.strip() if value else value
 
     def validate_number_of_beds(self, value):
         max_beds = 100
         if value > max_beds:
             msg = f"Cannot create more than {max_beds} beds at once."
@@ -74,7 +74,10 @@
             if (not facilities.filter(id=location.facility.id).exists()) or (
                 not facilities.filter(id=facility.id).exists()
             ):
-                raise PermissionError
+                error_message = (
+                    "You do not have permission to access this facility's bed."
+                )
+                raise PermissionDenied(error_message)
             del attrs["location"]
             attrs["location"] = location
             attrs["facility"] = facility
@@ -110,7 +113,10 @@
             if (
                 not facilities.filter(id=asset.current_location.facility.id).exists()
             ) or (not facilities.filter(id=bed.facility.id).exists()):
-                raise PermissionError
+                error_message = (
+                    "You do not have permission to access this facility's assetbed."
+                )
+                raise PermissionDenied(error_message)
             if AssetBed.objects.filter(asset=asset, bed=bed).exists():
                 raise ValidationError(
                     {"non_field_errors": "Asset is already linked to bed"}
@@ -149,7 +155,7 @@
     bed = BedSerializer(read_only=True)
     patient = SerializerMethodField()
 
     def get_patient(self, obj):
         from care.facility.api.serializers.patient import PatientListSerializer
 
         patient = PatientRegistration.objects.filter(

care/facility/api/viewsets/asset.py

@@ -61,6 +61,7 @@
 from care.facility.models.bed import AssetBed, ConsultationBed
 from care.users.models import User
 from care.utils.assetintegration.asset_classes import AssetClasses
+from care.utils.assetintegration.base import BaseAssetIntegration
 from care.utils.cache.cache_allowed_facilities import get_accessible_facilities
 from care.utils.filters.choicefilter import CareChoiceFilter, inverse_choices
 from care.utils.queryset.asset_bed import get_asset_queryset
@@ -88,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)
@@ -134,7 +135,7 @@
         return context
 
     def get_queryset(self):
         user = self.request.user
         queryset = self.queryset
         if user.is_superuser:
             pass
@@ -185,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(
@@ -199,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(
@@ -389,7 +390,6 @@
         This API is used to operate assets. API accepts the asset_id and action as parameters.
         """
         try:
-            action = request.data["action"]
             asset: Asset = self.get_object()
             middleware_hostname = (
                 asset.meta.get(
@@ -405,7 +405,7 @@
                     "middleware_hostname": middleware_hostname,
                 }
             )
-            result = asset_class.handle_action(action)
+            result = asset_class.handle_action(**request.data["action"])
             return Response({"result": result}, status=status.HTTP_200_OK)
 
         except ValidationError as e:

care/facility/api/viewsets/patient.py

@@ -126,7 +126,7 @@
     )
     ration_card_category = filters.ChoiceFilter(choices=RationCardCategory.choices)
 
     def filter_by_category(self, queryset, name, value):
         if value:
             queryset = queryset.filter(
                 (
@@ -234,13 +234,13 @@
     # Other Filters
     has_bed = filters.BooleanFilter(field_name="has_bed", method="filter_bed_not_null")
 
     def filter_bed_not_null(self, queryset, name, value):
         return queryset.filter(
             last_consultation__bed_number__isnull=value,
             last_consultation__discharge_date__isnull=True,
         )
 
     def filter_by_review_missed(self, queryset, name, value):
         if isinstance(value, bool):
             if value:
                 queryset = queryset.filter(
@@ -261,7 +261,7 @@
     diagnoses_differential = MultiSelectFilter(method="filter_by_diagnoses")
     diagnoses_confirmed = MultiSelectFilter(method="filter_by_diagnoses")
 
     def filter_by_diagnoses(self, queryset, name, value):
         if not value:
             return queryset
         filter_q = Q(last_consultation__diagnoses__diagnosis_id__in=value.split(","))
@@ -285,7 +285,7 @@
         method="filter_by_has_consents"
     )
 
     def filter_by_has_consents(self, queryset, name, value: str):
         if not value:
             return queryset
 
@@ -337,7 +337,7 @@
                 queryset = queryset.filter(q_filters)
         return queryset
 
     def filter_list_queryset(self, request, queryset, view):
         try:
             show_without_facility = json.loads(
                 request.query_params.get("without_facility")
@@ -351,7 +351,7 @@
 
 
 class PatientCustomOrderingFilter(BaseFilterBackend):
     def filter_queryset(self, request, queryset, view):
         ordering = request.query_params.get("ordering", "")
 
         if ordering in ("category_severity", "-category_severity"):
@@ -589,7 +589,7 @@
     @action(detail=True, methods=["POST"])
     def transfer(self, request, *args, **kwargs):
         patient = PatientRegistration.objects.get(external_id=kwargs["external_id"])
-        facility = Facility.objects.get(external_id=request.data["facility"])
+        facility = get_object_or_404(Facility, external_id=request.data["facility"])
 
         if patient.is_expired:
             return Response(
@@ -954,7 +954,7 @@
             patient_note__external_id=self.kwargs.get("notes_external_id")
         )
 
         if user.is_superuser:
             return queryset
         if user.user_type >= User.TYPE_VALUE_MAP["StateLabAdmin"]:
             queryset = queryset.filter(

care/facility/tests/test_asset_api.py

@@ -1,17 +1,21 @@
 from django.utils.timezone import now, timedelta
 from rest_framework import status
+from rest_framework.exceptions import ValidationError
 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.assetintegration.hl7monitor import HL7MonitorAsset
+from care.utils.assetintegration.onvif import OnvifAsset
+from care.utils.assetintegration.ventilator import VentilatorAsset
 from care.utils.tests.test_utils import TestUtils
 
 
 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)
@@ -29,8 +33,145 @@
 
     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_meta_validations_for_onvif_asset(self):
+        valid_meta = {
+            "local_ip_address": "192.168.0.1",
+            "camera_access_key": "username:password:access_key",
+            "middleware_hostname": "middleware.local",
+            "insecure_connection": True,
+        }
+        onvif_asset = OnvifAsset(valid_meta)
+        self.assertEqual(onvif_asset.middleware_hostname, "middleware.local")
+        self.assertEqual(onvif_asset.host, "192.168.0.1")
+        self.assertEqual(onvif_asset.username, "username")
+        self.assertEqual(onvif_asset.password, "password")
+        self.assertEqual(onvif_asset.access_key, "access_key")
+        self.assertTrue(onvif_asset.insecure_connection)
+
+        invalid_meta_cases = [
+            # Invalid format for camera_access_key
+            {
+                "id": "123",
+                "local_ip_address": "192.168.0.1",
+                "middleware_hostname": "middleware.local",
+                "camera_access_key": "invalid_format",
+            },
+            # Missing username/password in camera_access_key
+            {
+                "local_ip_address": "192.168.0.1",
+                "middleware_hostname": "middleware.local",
+                "camera_access_key": "invalid_format",
+            },
+            # Missing middleware_hostname
+            {
+                "local_ip_address": "192.168.0.1",
+                "camera_access_key": "username:password:access_key",
+            },
+            # Missing local_ip_address
+            {
+                "middleware_hostname": "middleware.local",
+                "camera_access_key": "username:password:access_key",
+            },
+            # Invalid value for insecure_connection
+            {
+                "local_ip_address": "192.168.0.1",
+                "camera_access_key": "username:password:access_key",
+                "middleware_hostname": "middleware.local",
+                "insecure_connection": "invalid_value",
+            },
+        ]
+        for meta in invalid_meta_cases:
+            self.validate_invalid_meta(OnvifAsset, meta)
+
+    def test_meta_validations_for_ventilator_asset(self):
+        valid_meta = {
+            "id": "123",
+            "local_ip_address": "192.168.0.1",
+            "middleware_hostname": "middleware.local",
+            "insecure_connection": True,
+        }
+        ventilator_asset = VentilatorAsset(valid_meta)
+        self.assertEqual(ventilator_asset.middleware_hostname, "middleware.local")
+        self.assertEqual(ventilator_asset.host, "192.168.0.1")
+        self.assertTrue(ventilator_asset.insecure_connection)
+
+        invalid_meta_cases = [
+            # Missing id
+            {
+                "local_ip_address": "192.168.0.1",
+                "middleware_hostname": "middleware.local",
+            },
+            # Missing middleware_hostname
+            {"id": "123", "local_ip_address": "192.168.0.1"},
+            # Missing local_ip_address
+            {"id": "123", "middleware_hostname": "middleware.local"},
+            # Invalid insecure_connection
+            {
+                "id": "123",
+                "local_ip_address": "192.168.0.1",
+                "middleware_hostname": "middleware.local",
+                "insecure_connection": "invalid_value",
+            },
+            # Camera access key not required for ventilator, invalid meta
+            {
+                "id": "21",
+                "local_ip_address": "192.168.0.1",
+                "camera_access_key": "username:password:access_key",
+                "middleware_hostname": "middleware.local",
+                "insecure_connection": True,
+            },
+        ]
+        for meta in invalid_meta_cases:
+            self.validate_invalid_meta(VentilatorAsset, meta)
+
+    def test_meta_validations_for_hl7monitor_asset(self):
+        valid_meta = {
+            "id": "123",
+            "local_ip_address": "192.168.0.1",
+            "middleware_hostname": "middleware.local",
+            "insecure_connection": True,
+        }
+        hl7monitor_asset = HL7MonitorAsset(valid_meta)
+        self.assertEqual(hl7monitor_asset.middleware_hostname, "middleware.local")
+        self.assertEqual(hl7monitor_asset.host, "192.168.0.1")
+        self.assertEqual(hl7monitor_asset.id, "123")
+        self.assertTrue(hl7monitor_asset.insecure_connection)
+
+        invalid_meta_cases = [
+            # Missing id
+            {
+                "local_ip_address": "192.168.0.1",
+                "middleware_hostname": "middleware.local",
+            },
+            # Missing middleware_hostname
+            {"id": "123", "local_ip_address": "192.168.0.1"},
+            # Missing local_ip_address
+            {"id": "123", "middleware_hostname": "middleware.local"},
+            # Invalid insecure_connection
+            {
+                "id": "123",
+                "local_ip_address": "192.168.0.1",
+                "middleware_hostname": "middleware.local",
+                "insecure_connection": "invalid_value",
+            },
+            # Camera access key not required for HL7Monitor, invalid meta
+            {
+                "id": "123",
+                "local_ip_address": "192.168.0.1",
+                "camera_access_key": "username:password:access_key",
+                "middleware_hostname": "middleware.local",
+                "insecure_connection": True,
+            },
+        ]
+        for meta in invalid_meta_cases:
+            self.validate_invalid_meta(HL7MonitorAsset, meta)
+
     def test_list_assets(self):
         response = self.client.get("/api/v1/asset/")
         self.assertEqual(response.status_code, status.HTTP_200_OK)
@@ -184,8 +325,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)
@@ -200,7 +341,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,
@@ -217,7 +358,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,
@@ -238,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": test_location.external_id,