[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:
@@ -77,8 +81,10 @@ jobs:
           pip install ${{ matrix.django-version }}
 
       - name: Start InfluxDB and Redis container
-        run: docker compose up -d influxdb redis
-
+        run: |
+          docker compose up -d influxdb redis
+          until curl -sI "http://localhost:8086/ping"; do sleep 1; done
+          docker compose exec -T influxdb influx -execute 'CREATE DATABASE openwisp2_test'
       - name: QA checks
         run: |
           ./run-qa-checks

docker-compose.yml

@@ -3,6 +3,7 @@ version: "3"
 services:
   influxdb:
     image: influxdb:1.8-alpine
+    platform: linux/amd64
     volumes:
       - influxdb-data:/var/lib/influxdb
     ports:

openwisp_radius/consumers.py

@@ -1,6 +1,5 @@
 from asgiref.sync import sync_to_async
 from channels.generic.websocket import AsyncJsonWebsocketConsumer
-from django.core.exceptions import ObjectDoesNotExist
 
 from .utils import load_model
 
@@ -12,13 +11,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,149 @@
+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._create_admin(password=self.TEST_PASSWORD)
+        batch = self._create_radius_batch(
+            name="test-batch",
+            strategy="prefix",
+            prefix="test-",
+            organization=org,
+        )
+        return org, user, batch
+
+    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_administrator(
+            organizations=[org], password=self.TEST_PASSWORD
+        )
+
+        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_user(is_staff=False, password=self.TEST_PASSWORD)
+
+        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 = self._create_user(is_staff=True, password=self.TEST_PASSWORD)
+
+        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)()