centralize channels==4.2 for django_loci

django_loci/channels/asgi.py

@@ -7,21 +7,21 @@
 from django_loci.channels.base import location_broadcast_path
 from django_loci.channels.consumers import LocationBroadcast
 
-channel_routing = ProtocolTypeRouter(
+application = ProtocolTypeRouter(
     {
-        'websocket': AllowedHostsOriginValidator(
+        "websocket": AllowedHostsOriginValidator(
             AuthMiddlewareStack(
                 URLRouter(
                     [
                         path(
                             location_broadcast_path,
                             LocationBroadcast.as_asgi(),
-                            name='LocationChannel',
+                            name="LocationChannel",
                         )
                     ]
                 )
             )
         ),
-        'http': get_asgi_application(),
+        "http": get_asgi_application(),
     }
-)
+)

django_loci/channels/base.py

@@ -18,8 +18,6 @@ class BaseLocationBroadcast(JsonWebsocketConsumer):
     to authorized users (superusers or organization operators)
     """
 
-    http_user = True
-
     def connect(self):
         self.pk = None
         try:
@@ -28,19 +26,24 @@ def connect(self):
         except KeyError:
             # Will fall here when the scope does not have
             # one of the variables, most commonly, user
-            # (When a user tries to access without loggin in)
             self.close()
+            return
+
         location = _get_object_or_none(self.model, pk=self.pk)
         if not location or not self.is_authorized(user, location):
             self.close()
             return
+
         self.accept()
+        # Create group name once
+        self.group_name = f'loci.mobile-location.{self.pk}'
         async_to_sync(self.channel_layer.group_add)(
-            'loci.mobile-location.{0}'.format(self.pk), self.channel_name
+            self.group_name,
+            self.channel_name
         )
 
     def is_authorized(self, user, location):
-        perm = '{0}.change_location'.format(self.model._meta.app_label)
+        perm = f'{self.model._meta.app_label}.change_location'
         authenticated = user.is_authenticated
         return authenticated and (
             user.is_superuser or (user.is_staff and user.has_perm(perm))
@@ -49,10 +52,12 @@ def is_authorized(self, user, location):
     def send_message(self, event):
         self.send_json(event['message'])
 
-    def disconnect(self, message=None):
+    def disconnect(self, close_code):
         """
         Perform things on connection close
         """
-        async_to_sync(self.channel_layer.group_discard)(
-            'loci.mobile-location.{0}'.format(self.pk), self.channel_name
-        )
+        if hasattr(self, 'group_name'):
+            async_to_sync(self.channel_layer.group_discard)(
+                self.group_name,
+                self.channel_name
+            )

django_loci/channels/receivers.py

@@ -7,12 +7,12 @@
 
 
 def update_mobile_location(sender, instance, **kwargs):
-    if not kwargs.get('created') and instance.geometry:
-        group_name = 'loci.mobile-location.{0}'.format(str(instance.pk))
+    if not kwargs.get("created") and instance.geometry:
+        group_name = f"loci.mobile-location.{str(instance.pk)}"
         channel_layer = channels.layers.get_channel_layer()
         message = json.loads(instance.geometry.geojson)
         async_to_sync(channel_layer.group_send)(
-            group_name, {'type': 'send_message', 'message': message}
+            group_name, {"type": "send_message", "message": message}
         )
 
 
@@ -21,8 +21,6 @@ def load_location_receivers(sender):
     enables signal listening when called
     designed to be called in AppConfig subclasses
     """
-    # using decorator pattern with old syntax
-    # in order to decorate an existing function
-    receiver(post_save, sender=sender, dispatch_uid='ws_update_mobile_location')(
+    receiver(post_save, sender=sender, dispatch_uid="ws_update_mobile_location")(
         update_mobile_location
-    )
+    )

django_loci/tests/base/test_channels.py

@@ -11,23 +11,21 @@
 from django.http.request import HttpRequest
 
 from django_loci.channels.consumers import LocationBroadcast
-
 from ...channels.base import _get_object_or_none
 from .. import TestAdminMixin, TestLociMixin
 
 
 class BaseTestChannels(TestAdminMixin, TestLociMixin):
     """
-    In channels 2.x, Websockets can only be tested
-    asynchronously, hence, pytest is used for these tests.
+    Test class for WebSocket functionality using pytest for async testing
     """
 
-    def _force_login(self, user, backend=None):
+    async def _force_login(self, user, backend=None):
         engine = importlib.import_module(settings.SESSION_ENGINE)
         request = HttpRequest()
         request.session = engine.SessionStore()
-        login(request, user, backend)
-        request.session.save
+        await database_sync_to_async(login)(request, user, backend)
+        await database_sync_to_async(request.session.save)()
         return request.session
 
     async def _get_request_dict(self, pk=None, user=None):
@@ -39,41 +37,45 @@ async def _get_request_dict(self, pk=None, user=None):
                 location=location
             )
             pk = location.pk
-        path = '/ws/loci/location/{0}/'.format(pk)
+        path = f'/ws/loci/location/{pk}/'
         session = None
         if user:
-            session = await database_sync_to_async(self._force_login)(user)
+            session = await self._force_login(user)
         return {'pk': pk, 'path': path, 'session': session}
 
     def _get_communicator(self, request_vars, user=None):
         communicator = WebsocketCommunicator(
-            LocationBroadcast.as_asgi(), request_vars['path']
+            LocationBroadcast.as_asgi(), 
+            request_vars['path'],
+            subprotocols=['websocket']
         )
         if user:
-            communicator.scope.update(
-                {
-                    "user": user,
-                    "session": request_vars['session'],
-                    "url_route": {"kwargs": {"pk": request_vars['pk']}},
-                }
-            )
+            communicator.scope.update({
+                "user": user,
+                "session": request_vars['session'],
+                "url_route": {"kwargs": {"pk": request_vars['pk']}},
+            })
         return communicator
 
     @pytest.mark.django_db(transaction=True)
-    def test_object_or_none(self):
-        result = _get_object_or_none(self.location_model, pk=1)
+    async def test_object_or_none(self):
+        result = await database_sync_to_async(_get_object_or_none)(
+            self.location_model, 
+            pk=1
+        )
         assert result is None
         plausible_pk = self.location_model().pk
-        result = _get_object_or_none(self.location_model, pk=plausible_pk)
+        result = await database_sync_to_async(_get_object_or_none)(
+            self.location_model, 
+            pk=plausible_pk
+        )
         assert result is None
 
     @pytest.mark.asyncio
     @pytest.mark.django_db(transaction=True)
     async def test_consumer_unauthenticated(self):
         request_vars = await self._get_request_dict()
-        communicator = WebsocketCommunicator(
-            LocationBroadcast.as_asgi(), request_vars['path']
-        )
+        communicator = self._get_communicator(request_vars)
         connected, _ = await communicator.connect()
         assert not connected
         await communicator.disconnect()
@@ -126,13 +128,10 @@ async def test_consumer_staff_but_no_change_permission(self):
         connected, _ = await communicator.connect()
         assert not connected
         await communicator.disconnect()
+
         # add permission to change location and repeat
         loc_perm = await database_sync_to_async(
-            (
-                await database_sync_to_async(Permission.objects.filter)(
-                    name='Can change location'
-                )
-            ).first
+            Permission.objects.filter(name='Can change location').first
         )()
         await database_sync_to_async(test_user.user_permissions.add)(loc_perm)
         test_user = await database_sync_to_async(self.user_model.objects.get)(
@@ -151,17 +150,17 @@ async def test_location_update(self):
         communicator = self._get_communicator(request_vars, test_user)
         connected, _ = await communicator.connect()
         assert connected
-        await database_sync_to_async(self._save_location)(request_vars['pk'])
+        await self._save_location(request_vars['pk'])
         response = await communicator.receive_json_from()
         assert response == {'type': 'Point', 'coordinates': [12.513124, 41.897903]}
         await communicator.disconnect()
 
-    def _save_location(self, pk):
-        loc = self.location_model.objects.get(pk=pk)
+    async def _save_location(self, pk):
+        loc = await database_sync_to_async(self.location_model.objects.get)(pk=pk)
         loc.geometry = 'POINT (12.513124 41.897903)'
-        loc.save()
+        await database_sync_to_async(loc.save)()
 
-    def test_routing(self):
+    @pytest.mark.django_db(transaction=True)
+    async def test_routing(self):
         from django_loci.channels.asgi import channel_routing
-
-        assert isinstance(channel_routing, ProtocolTypeRouter)
+        assert isinstance(channel_routing, ProtocolTypeRouter)

requirements.txt

@@ -1,6 +1,5 @@
 # Does not supports Django 4.0.0
 django>=3.2.18,!=4.0.*,<4.3
-channels~=3.0.4
 django-leaflet~=0.31.0
 Pillow~=10.4.0
 geopy~=2.4.1