[fix] Fixed DeviceConnection and Command API endpoints for non-superuser

openwisp_controller/connection/api/views.py

@@ -43,6 +43,8 @@ class BaseCommandView(
     BaseProtectedAPIMixin,
     FilterByParentManaged,
 ):
+    organization_field = "device__organization"
+    organization_lookup = "organization__in"
     model = Command
     queryset = Command.objects.prefetch_related("device")
     serializer_class = CommandSerializer

openwisp_controller/connection/tests/test_api.py

@@ -315,36 +315,91 @@ def test_endpoints_for_deactivated_device(self):
             )
             self.assertEqual(response.status_code, 200)
 
-    def test_non_superuser(self):
-        list_url = self._get_path("device_command_list", self.device_id)
+    def test_endpoints_for_org_operators_own_org(self):
+        self.client.logout()
+        operator = self._create_operator(organizations=[self._get_org()])
+        self.client.force_login(operator)
+        list_path = self._get_path("device_command_list", self.device_id)
         command = self._create_command(device_conn=self.device_conn)
-        device = command.device
 
-        with self.subTest("Test with unauthenticated user"):
-            self.client.logout()
-            response = self.client.get(list_url)
-            self.assertEqual(response.status_code, 401)
+        with self.subTest("Operator can list commands for device"):
+            response = self.client.get(list_path)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data["results"][0]["id"], str(command.id))
 
-        with self.subTest("Test with organization member"):
-            org_user = self._create_org_user(is_admin=True)
-            org_user.user.groups.add(Group.objects.get(name="Operator"))
-            self.client.force_login(org_user.user)
-            self.assertEqual(device.organization, org_user.organization)
+        with self.subTest("Operator can create command for device"):
+            self.assertEqual(Command.objects.count(), 1)
+            payload = {
+                "type": "custom",
+                "input": {"command": "echo test"},
+            }
+            response = self.client.post(
+                list_path, data=payload, content_type="application/json"
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertEqual(response.data["type"], "Custom commands")
+            self.assertEqual(response.data["input"], "echo test")
+            self.assertEqual(Command.objects.count(), 2)
 
-            response = self.client.get(list_url)
+        with self.subTest("Operator can retrieve command for device"):
+            detail_path = self._get_path(
+                "device_command_details", self.device_id, command.id
+            )
+            response = self.client.get(detail_path)
             self.assertEqual(response.status_code, 200)
-            self.assertEqual(response.data["count"], 1)
+            self.assertEqual(response.data["id"], str(command.id))
 
-        with self.subTest("Test with org member of different org"):
-            org2 = self._create_org(name="org2", slug="org2")
-            org2_user = self._create_user(username="org2user", email="[email protected]")
-            self._create_org_user(organization=org2, user=org2_user, is_admin=True)
-            self.client.force_login(org2_user)
-            org2_user.groups.add(Group.objects.get(name="Operator"))
+    def test_endpoints_for_org_operator_different_org(self):
+        org2 = self._create_org(name="org2", slug="org2")
+        org2_admin = self._create_operator(organizations=[org2])
+        org1_command = self._create_command(device_conn=self.device_conn)
+        list_url = self._get_path("device_command_list", self.device_id)
+        detail_url = self._get_path(
+            "device_command_details", self.device_id, org1_command.id
+        )
 
+        self.client.logout()
+        self.client.force_login(org2_admin)
+
+        with self.subTest("List operation"):
             response = self.client.get(list_url)
             self.assertEqual(response.status_code, 404)
 
+        with self.subTest("Create operation"):
+            response = self.client.post(
+                list_url,
+                data={"type": "custom", "input": {"command": "echo test"}},
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 404)
+
+        with self.subTest("Retrieve operation"):
+            response = self.client.get(detail_url)
+            self.assertEqual(response.status_code, 404)
+
+    def test_unauthenticated_user(self):
+        list_url = self._get_path("device_command_list", self.device_id)
+        command = self._create_command(device_conn=self.device_conn)
+        self.client.logout()
+
+        with self.subTest("List operation"):
+            response = self.client.get(list_url)
+            self.assertEqual(response.status_code, 401)
+
+        with self.subTest("Create operation"):
+            response = self.client.post(
+                list_url,
+                data={"type": "custom", "input": {"command": "echo test"}},
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 401)
+
+        with self.subTest("Retrieve operation"):
+            response = self.client.get(
+                self._get_path("device_command_details", self.device_id, command.id)
+            )
+            self.assertEqual(response.status_code, 401)
+
     def test_non_existent_command(self):
         url = self._get_path("device_command_list", self.device_id)
         with patch.dict(
@@ -387,6 +442,61 @@ def test_create_command_without_connection(self):
         )
 
 
+# The same tests, but with a normal user
+class TestCommandsApiNonAdmin(TestCommandsAPI):
+    def setUp(self):
+        # Organisation to manage devices
+        org1 = self._get_org()
+        # Admin for this organisation
+        self.administrator = self._create_administrator(organizations=[org1])
+        self.client.force_login(self.administrator)
+        # Credentials for the same organisation
+        cred1 = self._create_credentials(organization=org1)
+        # Connection to a device with these credentials
+        self.device_conn = self._create_device_connection(credentials=cred1)
+        self.device_id = self.device_conn.device.id
+
+    def test_bearer_authentication(self):
+        self.client.logout()
+        command_obj = self._create_command(device_conn=self.device_conn)
+        token = self._obtain_auth_token(username="administrator", password="tester")
+
+        with self.subTest("Test creating command"):
+            url = self._get_path("device_command_list", self.device_id)
+            payload = {
+                "type": "custom",
+                "input": {"command": "echo test"},
+            }
+            response = self.client.post(
+                url,
+                data=payload,
+                content_type="application/json",
+                HTTP_AUTHORIZATION=f"Bearer {token}",
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertIn("id", response.data)
+
+        with self.subTest("Test retrieving command"):
+            url = self._get_path(
+                "device_command_details", self.device_id, command_obj.id
+            )
+            response = self.client.get(
+                url,
+                HTTP_AUTHORIZATION=f"Bearer {token}",
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertIn("id", response.data)
+
+        with self.subTest("Test listing command"):
+            url = self._get_path("device_command_list", self.device_id)
+            response = self.client.get(
+                url,
+                HTTP_AUTHORIZATION=f"Bearer {token}",
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(len(response.data["results"]), 2)
+
+
 class TestConnectionApi(
     TestAdminMixin, AuthenticationMixin, TestCase, CreateConnectionsMixin
 ):