added "IsAdminUser" permission to Sensor Hardware api

manifests/tests/test_sensor_hardware.py

@@ -1,9 +1,15 @@
 from django.test import TestCase
 from manifests.models import *
-from node_auth import get_node_token_model, get_node_model
+from node_auth import get_node_token_model, get_node_model, get_node_token_keyword
+from django.contrib.auth import get_user_model
+from rest_framework.authtoken.models import Token as User_Token
+from app import get_user_token_keyword
 
-Token = get_node_token_model()
+Node_Token = get_node_token_model()
+NodeTokenKeyword = get_node_token_keyword()
 Node = get_node_model()
+User = get_user_model()
+UserTokenKeyword = get_user_token_keyword()
 
 class SensorHardwareViewsTest(TestCase):
     def setUp(self):
@@ -134,19 +140,26 @@ def test_detail_view(self):
 
 class SensorHardwareNodeCRUDViewSetTest(TestCase):
     def setUp(self):
+        # Create an admin user
+        self.admin_username = 'admin'
+        self.admin_password = 'adminpassword'
+        self.admin_user = User.objects.create_superuser(self.admin_username, '[email protected]', self.admin_password)
+        self.UserToken = User_Token.objects.create(user=self.admin_user)
+        self.UserKey = self.UserToken.key
+        # Create Node
         self.Myvsn = "W001"
         self.mac = "111"
         self.node = Node.objects.create(vsn=self.Myvsn, mac=self.mac)
-        self.token = Token.objects.get(node=self.node)
-        self.key = self.token.key
+        self.NodeToken = Node_Token.objects.get(node=self.node)
+        self.key = self.NodeToken.key
         self.gpsSensor = SensorHardware.objects.create(hardware="gps", hw_model="A GPS")
         self.raingaugeSensor = SensorHardware.objects.create(hardware="raingauge", hw_model="RG-15")
 
     def test_create(self):
         """Test the Sensor Hardware CREATE endpoint for Authenticated Nodes"""
         data = {"hardware": "test","hw_model": "test-123", "description": "test"}
         r = self.client.post("/sensorhardwares/", data=data, 
-            content_type="application/json", HTTP_AUTHORIZATION=f"node_auth {self.key}"
+            content_type="application/json", HTTP_AUTHORIZATION=f"{NodeTokenKeyword} {self.key}"
         )
         self.assertEqual(r.status_code, 201)
 
@@ -156,7 +169,7 @@ def test_create(self):
 
     def test_get(self):
         """Test the Sensor Hardware GET endpoint for Authenticated Nodes"""
-        r = self.client.get(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"node_auth {self.key}")
+        r = self.client.get(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"{NodeTokenKeyword} {self.key}")
         self.assertEqual(r.status_code, 200)
 
         #assert the correct device was returned in the request
@@ -167,7 +180,7 @@ def test_update(self):
         """Test the Sensor Hardware PATCH endpoint for Authenticated Nodes"""
         data = {"description": "test"}
         r = self.client.patch(f"/sensorhardwares/{self.gpsSensor.hw_model}/",data=data,
-            content_type="application/json", HTTP_AUTHORIZATION=f"node_auth {self.key}"
+            content_type="application/json", HTTP_AUTHORIZATION=f"{NodeTokenKeyword} {self.key}"
         )
         self.assertEqual(r.status_code, 200)
 
@@ -177,7 +190,7 @@ def test_update(self):
 
     def test_delete(self):
         """Test the Sensor Hardware DELETE endpoint for Authenticated Nodes"""
-        r = self.client.delete(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"node_auth {self.key}")
+        r = self.client.delete(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"{NodeTokenKeyword} {self.key}")
         self.assertEqual(r.status_code, 204)
 
         #check if the device was deleted in the db
@@ -189,7 +202,29 @@ def test_unauthenticated(self):
         r = self.client.delete(f"/sensorhardwares/{self.gpsSensor.hw_model}/")
         self.assertEqual(r.status_code, 401)
 
-    def test_wrongToken(self):
-        """Test request with wrong token returns an error"""
-        r = self.client.get(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"node_auth wrong_token")
+    def test_wrong_NodeToken(self):
+        """Test request with wrong node token returns an error"""
+        r = self.client.get(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"{NodeTokenKeyword} wrong_token")
+        self.assertEqual(r.status_code, 401)
+
+    def test_create_User(self):
+        """Test the Sensor Hardware CREATE endpoint with a User"""
+        data = {"hardware": "test","hw_model": "test-123", "description": "test"}
+        r = self.client.post("/sensorhardwares/", data=data, 
+            content_type="application/json", HTTP_AUTHORIZATION=f"{UserTokenKeyword} {self.UserKey}"
+        )
+        self.assertEqual(r.status_code, 201)
+
+        #check if the device was created in the db
+        sensor_exists = SensorHardware.objects.filter(hw_model=data["hw_model"]).exists()
+        self.assertTrue(sensor_exists)
+
+    def test_wrong_UserToken(self):
+        """Test request with wrong user token returns an error"""
+        r = self.client.get(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"{UserTokenKeyword} wrong_token")
+        self.assertEqual(r.status_code, 401)
+
+    def test_mismatch_token(self):
+        """Test request with mismatching keyword and token returns an error"""
+        r = self.client.get(f"/sensorhardwares/{self.gpsSensor.hw_model}/", HTTP_AUTHORIZATION=f"{NodeTokenKeyword} {self.UserKey}")
         self.assertEqual(r.status_code, 401)

manifests/urls.py

@@ -8,7 +8,7 @@
     LorawanDeviceView,
     LorawanConnectionView,
     LorawanKeysView,
-    SensorHardwareViewSet_NodeCRUD
+    SensorHardwareViewSet_CRUD
 )
 
 app_name = "manifests"
@@ -22,7 +22,7 @@
 router.register(
     r"lorawanconnections", LorawanConnectionView, basename="lorawanconnections"
 )
-router.register(r"sensorhardwares", SensorHardwareViewSet_NodeCRUD, basename="sensorhardwares")
+router.register(r"sensorhardwares", SensorHardwareViewSet_CRUD, basename="sensorhardwares")
 
 urlpatterns = [
     path("", include(router.urls)),

manifests/views.py

@@ -1,7 +1,7 @@
 from django.contrib.auth.models import *
 from django.http import Http404
 from django.core.exceptions import ObjectDoesNotExist
-from rest_framework.permissions import IsAuthenticatedOrReadOnly
+from rest_framework.permissions import IsAuthenticatedOrReadOnly, IsAdminUser
 from rest_framework.viewsets import ReadOnlyModelViewSet, ModelViewSet
 from .models import *
 from .serializers import (
@@ -18,6 +18,7 @@
 from rest_framework import status
 from django.db import IntegrityError
 from node_auth.mixins import NodeAuthMixin, NodeOwnedObjectsMixin
+from app.authentication import TokenAuthentication as UserTokenAuthentication
 
 
 class ManifestViewSet(ReadOnlyModelViewSet):
@@ -84,11 +85,12 @@ def list(self, request, *args, **kwargs):
 
         return res
 
-class SensorHardwareViewSet_NodeCRUD(NodeAuthMixin, ModelViewSet):
+class SensorHardwareViewSet_CRUD(NodeAuthMixin, ModelViewSet):
     queryset = SensorHardware.objects.all()
     serializer_class = SensorHardwareCRUDSerializer
     lookup_field = "hw_model"
-
+    authentication_classes = (NodeAuthMixin.authentication_classes[0],UserTokenAuthentication)
+    permission_classes = (NodeAuthMixin.permission_classes[0]|IsAdminUser,)    
 
 class NodeBuildViewSet(ReadOnlyModelViewSet):
     queryset = (