[fix] Fixed handling of non-existing device in DeviceLocationView

Author: pandafy

openwisp_controller/geo/api/views.py

@@ -105,6 +105,7 @@ def create_location(self, device):
 
 class DeviceLocationView(
     RelatedDeviceProtectedAPIMixin,
+    FilterByParentManaged,
     generics.RetrieveUpdateDestroyAPIView,
 ):
     serializer_class = DeviceLocationSerializer
@@ -124,9 +125,7 @@ def get_queryset(self):
             return qs.none()
 
     def get_parent_queryset(self):
-        return Device.objects.filter(
-            pk=self.kwargs['pk'],
-        )
+        return Device.objects.filter(pk=self.kwargs['pk'])
 
     def get_serializer_context(self):
         context = super().get_serializer_context()

openwisp_controller/geo/tests/test_api.py

@@ -655,6 +655,33 @@ def test_create_devicelocation_outdoor_location_with_floorplan(self):
         )
         self.assertEqual(self.object_location_model.objects.count(), 0)
 
+    def test_endpoints_for_non_existent_device(self):
+        device_id = uuid.uuid4()
+        floorplan = self._create_floorplan()
+        location = floorplan.location
+        url = reverse('geo_api:device_location', args=[device_id])
+
+        with self.subTest('Retrieve operation'):
+            response = self.client.get(
+                url,
+            )
+            self.assertEqual(response.status_code, 404)
+
+        with self.subTest('Update operation'):
+            response = self.client.put(
+                url,
+                data={
+                    'location': str(location.id),
+                    'floorplan': str(floorplan.id),
+                },
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 404)
+
+        with self.subTest('Delete operation'):
+            response = self.client.delete(url)
+            self.assertEqual(response.status_code, 404)
+
     def test_create_devicelocation_using_non_existing_related_ids(self):
         device = self._create_object()
         floorplan = self._create_floorplan()
@@ -695,7 +722,7 @@ def test_create_devicelocation_using_related_ids(self):
         floorplan = self._create_floorplan()
         location = floorplan.location
         url = reverse('geo_api:device_location', args=[device.id])
-        with self.assertNumQueries(15):
+        with self.assertNumQueries(18):
             response = self.client.put(
                 url,
                 data={
@@ -733,7 +760,7 @@ def test_create_devicelocation_location_floorplan(self):
             'floorplan.image': self._get_simpleuploadedfile(),
             'indoor': ['12.342,23.541'],
         }
-        with self.assertNumQueries(29):
+        with self.assertNumQueries(32):
             response = self.client.put(
                 url, encode_multipart(BOUNDARY, data), content_type=MULTIPART_CONTENT
             )
@@ -800,7 +827,7 @@ def test_create_devicelocation_only_location(self):
                 'type': 'indoor',
             }
         }
-        with self.assertNumQueries(18):
+        with self.assertNumQueries(21):
             response = self.client.put(url, data=data, content_type='application/json')
         self.assertEqual(response.status_code, 201)
         self.assertEqual(self.location_model.objects.count(), 1)
@@ -817,7 +844,7 @@ def test_create_devicelocation_only_floorplan(self):
             'floorplan.floor': 1,
             'floorplan.image': self._get_simpleuploadedfile(),
         }
-        with self.assertNumQueries(5):
+        with self.assertNumQueries(8):
             response = self.client.put(
                 url, encode_multipart(BOUNDARY, data), content_type=MULTIPART_CONTENT
             )
@@ -840,7 +867,7 @@ def test_create_devicelocation_existing_location_new_floorplan(self):
             'floorplan.image': self._get_simpleuploadedfile(),
             'indoor': ['12.342,23.541'],
         }
-        with self.assertNumQueries(23):
+        with self.assertNumQueries(26):
             response = self.client.put(
                 url, encode_multipart(BOUNDARY, data), content_type=MULTIPART_CONTENT
             )
@@ -863,7 +890,7 @@ def test_update_devicelocation_change_location_outdoor_to_indoor(self):
         }
         self.assertEqual(device_location.location.type, 'outdoor')
         self.assertEqual(device_location.floorplan, None)
-        with self.assertNumQueries(21):
+        with self.assertNumQueries(23):
             response = self.client.put(
                 path, encode_multipart(BOUNDARY, data), content_type=MULTIPART_CONTENT
             )
@@ -882,7 +909,7 @@ def test_update_devicelocation_patch_indoor(self):
             'indoor': '0,0',
         }
         self.assertEqual(device_location.indoor, '-140.38620,40.369227')
-        with self.assertNumQueries(10):
+        with self.assertNumQueries(12):
             response = self.client.patch(path, data, content_type='application/json')
         self.assertEqual(response.status_code, 200)
         device_location.refresh_from_db()
@@ -899,7 +926,7 @@ def test_update_devicelocation_floorplan_related_id(self):
         data = {
             'floorplan': str(floor2.id),
         }
-        with self.assertNumQueries(12):
+        with self.assertNumQueries(14):
             response = self.client.patch(path, data, content_type='application/json')
         self.assertEqual(response.status_code, 200)
         device_location.refresh_from_db()
@@ -913,7 +940,7 @@ def test_update_devicelocation_location_related_id(self):
         data = {
             'location': str(location2.id),
         }
-        with self.assertNumQueries(9):
+        with self.assertNumQueries(11):
             response = self.client.patch(path, data, content_type='application/json')
         self.assertEqual(response.status_code, 200)
         device_location.refresh_from_db()

openwisp_controller/mixins.py

@@ -12,8 +12,8 @@ def _has_permissions(self, request, view, perm, obj=None):
         if obj:
             device = getattr(obj, self._device_field)
         else:
-            device = view.get_parent_queryset()[0]
-        return perm and not device.is_deactivated()
+            device = view.get_parent_queryset().first()
+        return perm and device and not device.is_deactivated()
 
     def has_permission(self, request, view):
         perm = super().has_permission(request, view)