Merge pull request #2606 from ohcnetwork/develop

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,
@@ -174,11 +174,14 @@ def validate(self, attrs):
 
             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 (

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,
@@ -74,7 +74,10 @@ def validate(self, attrs):
             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 @@ def validate(self, attrs):
             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"}

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
@@ -389,7 +390,6 @@ def operate_assets(self, request, *args, **kwargs):
         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 @@ def operate_assets(self, request, *args, **kwargs):
                     "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

@@ -589,7 +589,7 @@ def list(self, request, *args, **kwargs):
     @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(

care/facility/tests/test_asset_api.py

@@ -1,10 +1,14 @@
 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
 
 
@@ -31,6 +35,143 @@ 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)