[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

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_commands.py

@@ -564,3 +564,27 @@ def test_convert_called_station_id_command_with_slug(self, *args):
                 call_command("convert_called_station_id")
             radius_acc.refresh_from_db()
             self.assertEqual(radius_acc.called_station_id, "CC-CC-CC-CC-CC-0C")
+
+    def test_convert_called_station_id_command_wrapper(self):
+        from ..management.commands.convert_called_station_id import Command
+
+        command = Command()
+        self.assertIsNotNone(command)
+        from ..management.commands.base.convert_called_station_id import (
+            BaseConvertCalledStationIdCommand,
+        )
+
+        self.assertIsInstance(command, BaseConvertCalledStationIdCommand)
+
+    def test_prefix_add_users_command_wrapper(self):
+        from ..management.commands.prefix_add_users import Command
+
+        command = Command()
+        self.assertIsNotNone(command)
+        from ..management.commands.base import BatchAddMixin
+        from ..management.commands.base.prefix_add_users import (
+            BasePrefixAddUsersCommand,
+        )
+
+        self.assertIsInstance(command, BatchAddMixin)
+        self.assertIsInstance(command, BasePrefixAddUsersCommand)

openwisp_radius/tests/test_consumers.py

@@ -0,0 +1,264 @@
+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)()
+
+    def test_batch_status_update(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.send_json_to(
+                {"type": "batch.status.update", "status": "completed"}
+            )
+
+            from channels.layers import get_channel_layer
+
+            channel_layer = get_channel_layer()
+
+            await channel_layer.group_send(
+                f"radius_batch_{batch.pk}",
+                {"type": "batch_status_update", "status": "processing"},
+            )
+
+            response = await communicator.receive_json_from()
+            assert response == {"status": "processing"}
+
+            await communicator.disconnect()
+
+        async_to_sync(test)()
+
+    def test_disconnect_cleanup(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()
+
+            from channels.layers import get_channel_layer
+
+            channel_layer = get_channel_layer()
+
+            await channel_layer.group_send(
+                f"radius_batch_{batch.pk}",
+                {"type": "batch_status_update", "status": "completed"},
+            )
+
+        async_to_sync(test)()
+
+    def test_user_can_access_batch_method(self):
+        _, user, batch = self._create_test_data()
+        consumer = RadiusBatchConsumer()
+
+        self.assertTrue(consumer._user_can_access_batch(user, batch.pk))
+
+        org = self._create_org(name="test-org-2", slug="test-org-2")
+        staff_user = self._create_administrator(
+            organizations=[org],
+            password=self.TEST_PASSWORD,
+            username="staff_user_2",
+            email="staff2@example.com",
+        )
+        batch2 = self._create_radius_batch(
+            name="test2",
+            organization=org,
+            strategy="prefix",
+            prefix="test-prefix-2",
+        )
+        self.assertTrue(consumer._user_can_access_batch(staff_user, batch2.pk))
+
+        other_org = self._create_org(name="other", slug="other")
+        other_user = self._create_administrator(
+            organizations=[other_org],
+            password=self.TEST_PASSWORD,
+            username="other_user",
+            email="other@example.com",
+        )
+        self.assertFalse(consumer._user_can_access_batch(other_user, batch2.pk))
+
+    def test_invalid_batch_id(self):
+        _, user, _ = self._create_test_data()
+
+        async def test():
+            invalid_batch_id = "00000000-0000-0000-0000-000000000000"
+            communicator = WebsocketCommunicator(
+                application,
+                f"/ws/radius/batch/{invalid_batch_id}/",
+            )
+            communicator.scope["user"] = user
+            communicator.scope["url_route"] = {"kwargs": {"batch_id": invalid_batch_id}}
+
+            connected, _ = await communicator.connect()
+            assert connected is False
+
+        async_to_sync(test)()
+
+    def test_user_can_access_batch_with_invalid_uuid(self):
+        _, user, _ = self._create_test_data()
+        consumer = RadiusBatchConsumer()
+        
+        result = consumer._user_can_access_batch(user, "00000000-0000-0000-0000-000000000000")
+        self.assertFalse(result)
+

openwisp_radius/tests/test_counters/test_base_counter.py

@@ -114,6 +114,13 @@ def test_resets(self):
             self.assertEqual(str(datetime.fromtimestamp(start)), "2021-10-22 00:00:00")
             self.assertEqual(str(datetime.fromtimestamp(end)), "2021-11-22 00:00:00")
 
+        with self.subTest("monthly_subscription future start date logic"):
+            user.date_joined = datetime.fromisoformat("2021-07-04 12:34:58")
+            user.save(update_fields=["date_joined"])
+            start, end = resets["monthly_subscription"](user)
+            self.assertEqual(str(datetime.fromtimestamp(start)), "2021-10-04 00:00:00")
+            self.assertEqual(str(datetime.fromtimestamp(end)), "2021-11-04 00:00:00")
+
         with self.subTest("never"):
             start, end = resets["never"]()
             self.assertEqual(start, 0)
@@ -131,5 +138,33 @@ class MaxInputOctetsCounter(BaseDailyCounter):
         self.assertEqual(BaseMontlhyTrafficCounter.get_attribute_type(), "bytes")
         self.assertEqual(MaxInputOctetsCounter.get_attribute_type(), "bytes")
 
+    def test_base_exception_logging(self):
+        from unittest.mock import MagicMock
+
+        from ...counters.exceptions import BaseException
+
+        logger = MagicMock()
+        BaseException("message", "error", logger)
+        logger.error.assert_called_with("message")
+        with self.assertRaises(AssertionError):
+            BaseException("message", "invalid_level", logger)
+
+    def test_consumed_method(self):
+        opts = self._get_kwargs("Max-Daily-Session")
+        from ...counters.sqlite.daily_counter import DailyCounter
+
+        counter = DailyCounter(**opts)
+        consumed = counter.consumed()
+        self.assertEqual(consumed, 0)
+        self.assertIsInstance(consumed, int)
+        
+        from .utils import _acct_data
+
+        self._create_radius_accounting(**_acct_data)
+        consumed = counter.consumed()
+        self.assertEqual(consumed, int(_acct_data["session_time"]))
+        self.assertIsInstance(consumed, int)
+
+
 
 del BaseTransactionTestCase