[tests] Add WebSocket consumer tests #677

Implement tests for RadiusBatchConsumer to verify authentication,
authorization, and group messaging logic.
Increases coverage for consumers.py to 84%.

Fixes #677

Author: shivsubh

.github/workflows/ci.yml

@@ -31,6 +31,10 @@ jobs:
           - python-version: "3.13"
             django-version: django~=4.2.0
 
+    permissions:
+      pull-requests: write
+      contents: read
+
     steps:
       - uses: actions/checkout@v6
         with:
@@ -99,6 +103,12 @@ jobs:
         if: ${{ failure() }}
         run: cat geckodriver.log
 
+      - name: Pytest coverage comment
+        if: ${{ success() }}
+        uses: MishaKav/pytest-coverage-comment@v1.2.0
+        with:
+          pytest-coverage-path: coverage.xml
+
       - name: Upload Coverage
         if: ${{ !cancelled() && steps.deps.conclusion == 'success' }}
         uses: coverallsapp/github-action@v2

openwisp_radius/consumers.py

@@ -12,13 +12,9 @@ def _user_can_access_batch(self, user, batch_id):
         if user.is_superuser:
             return RadiusBatch.objects.filter(pk=batch_id).exists()
         # For non-superusers, check their managed organizations
-        try:
-            RadiusBatch.objects.filter(
-                pk=batch_id, organization__in=user.organizations_managed
-            ).exists()
-            return True
-        except ObjectDoesNotExist:
-            return False
+        return RadiusBatch.objects.filter(
+            pk=batch_id, organization__in=user.organizations_managed
+        ).exists()
 
     async def connect(self):
         self.batch_id = self.scope["url_route"]["kwargs"]["batch_id"]

openwisp_radius/tests/test_consumers.py

@@ -0,0 +1,174 @@
+from asgiref.sync import async_to_sync
+from channels.routing import URLRouter
+from channels.testing import WebsocketCommunicator
+from django.contrib.auth import get_user_model
+from django.test import TransactionTestCase
+from django.urls import re_path
+
+from openwisp_users.tests.utils import TestOrganizationMixin
+
+from ..consumers import RadiusBatchConsumer
+from ..utils import load_model
+from . import CreateRadiusObjectsMixin
+
+User = get_user_model()
+RadiusBatch = load_model("RadiusBatch")
+
+application = URLRouter(
+    [
+        re_path(
+            r"^ws/radius/batch/(?P<batch_id>[^/]+)/$",
+            RadiusBatchConsumer.as_asgi(),
+        ),
+    ]
+)
+
+
+class TestRadiusBatchConsumer(
+    CreateRadiusObjectsMixin, TestOrganizationMixin, TransactionTestCase
+):
+
+    TEST_PASSWORD = "test_password"  # noqa: S105
+
+    def _create_test_data(self):
+        org = self._create_org()
+        user = self._get_admin()
+        user.is_staff = True
+        user.is_superuser = True
+        user.save()
+
+        batch = self._create_radius_batch(
+            name="test-batch",
+            strategy="prefix",
+            prefix="test-",
+            organization=org,
+        )
+        return org, user, batch
+
+    def _create_staff_user(self, org):
+        user = User.objects.create_user(
+            username="staff_user",
+            email="staff@test.com",
+            password=self.TEST_PASSWORD,
+            is_staff=True,
+        )
+        org.add_user(user, is_admin=True)
+        return user
+
+    def _create_regular_user(self):
+        return User.objects.create_user(
+            username="regular_user",
+            email="regular@test.com",
+            password=self.TEST_PASSWORD,
+            is_staff=False,
+        )
+
+    def test_websocket_connect_superuser(self):
+        _, user, batch = self._create_test_data()
+
+        async def test():
+            communicator = WebsocketCommunicator(
+                application,
+                f"/ws/radius/batch/{batch.pk}/",
+            )
+            communicator.scope["user"] = user
+            communicator.scope["url_route"] = {"kwargs": {"batch_id": str(batch.pk)}}
+
+            connected, _ = await communicator.connect()
+            assert connected is True
+            await communicator.disconnect()
+
+        async_to_sync(test)()
+
+    def test_websocket_connect_staff_with_permission(self):
+        org, _, batch = self._create_test_data()
+        staff_user = self._create_staff_user(org)
+
+        async def test():
+            communicator = WebsocketCommunicator(
+                application,
+                f"/ws/radius/batch/{batch.pk}/",
+            )
+            communicator.scope["user"] = staff_user
+            communicator.scope["url_route"] = {"kwargs": {"batch_id": str(batch.pk)}}
+
+            connected, _ = await communicator.connect()
+            assert connected is True
+            await communicator.disconnect()
+
+        async_to_sync(test)()
+
+    def test_websocket_reject_unauthenticated(self):
+        _, _, batch = self._create_test_data()
+
+        async def test():
+            communicator = WebsocketCommunicator(
+                application,
+                f"/ws/radius/batch/{batch.pk}/",
+            )
+            from django.contrib.auth.models import AnonymousUser
+
+            communicator.scope["user"] = AnonymousUser()
+            communicator.scope["url_route"] = {"kwargs": {"batch_id": str(batch.pk)}}
+
+            connected, _ = await communicator.connect()
+            assert connected is False
+
+        async_to_sync(test)()
+
+    def test_websocket_reject_non_staff(self):
+        _, _, batch = self._create_test_data()
+        regular_user = self._create_regular_user()
+
+        async def test():
+            communicator = WebsocketCommunicator(
+                application,
+                f"/ws/radius/batch/{batch.pk}/",
+            )
+            communicator.scope["user"] = regular_user
+            communicator.scope["url_route"] = {"kwargs": {"batch_id": str(batch.pk)}}
+
+            connected, _ = await communicator.connect()
+            assert connected is False
+
+        async_to_sync(test)()
+
+    def test_websocket_reject_no_permission(self):
+        _, _, batch = self._create_test_data()
+
+        staff_user = User.objects.create_user(
+            username="staff_no_access",
+            email="staff2@test.com",
+            password=self.TEST_PASSWORD,
+            is_staff=True,
+        )
+
+        async def test():
+            communicator = WebsocketCommunicator(
+                application,
+                f"/ws/radius/batch/{batch.pk}/",
+            )
+            communicator.scope["user"] = staff_user
+            communicator.scope["url_route"] = {"kwargs": {"batch_id": str(batch.pk)}}
+
+            connected, _ = await communicator.connect()
+            assert connected is False
+
+        async_to_sync(test)()
+
+    def test_websocket_group_connection(self):
+        _, user, batch = self._create_test_data()
+
+        async def test():
+            communicator = WebsocketCommunicator(
+                application,
+                f"/ws/radius/batch/{batch.pk}/",
+            )
+            communicator.scope["user"] = user
+            communicator.scope["url_route"] = {"kwargs": {"batch_id": str(batch.pk)}}
+
+            connected, _ = await communicator.connect()
+            assert connected is True
+            await communicator.disconnect()
+
+        async_to_sync(test)()