[tests] Add test for common ws endpoint

Author: dee077

openwisp_controller/geo/channels/consumers.py

@@ -1,5 +1,5 @@
 import swapper
-from django_loci.channels.base import BaseAllLocationBroadcast, BaseLocationBroadcast
+from django_loci.channels.base import BaseCommonLocationBroadcast, BaseLocationBroadcast
 
 Location = swapper.load_model("geo", "Location")
 
@@ -19,7 +19,7 @@ def is_authorized(self, user, location):
         return result
 
 
-class AllLocationBroadcast(BaseAllLocationBroadcast):
+class CommonLocationBroadcast(BaseCommonLocationBroadcast):
     model = Location
 
     def is_autherized(self, user, location):

openwisp_controller/geo/channels/routing.py

@@ -3,14 +3,14 @@
 from channels.security.websocket import AllowedHostsOriginValidator
 from django.urls import path
 from django_loci.channels.base import (
-    all_location_broadcast_path,
+    common_location_broadcast_path,
     location_broadcast_path,
 )
 from openwisp_notifications.websockets.routing import (
     get_routes as get_notification_routes,
 )
 
-from .consumers import AllLocationBroadcast, LocationBroadcast
+from .consumers import CommonLocationBroadcast, LocationBroadcast
 
 
 def get_routes():
@@ -19,8 +19,8 @@ def get_routes():
             location_broadcast_path, LocationBroadcast.as_asgi(), name="LocationChannel"
         ),
         path(
-            all_location_broadcast_path,
-            AllLocationBroadcast.as_asgi(),
+            common_location_broadcast_path,
+            CommonLocationBroadcast.as_asgi(),
             name="AllLocationChannel",
         ),
     ]

openwisp_controller/geo/tests/pytest.py

@@ -1,16 +1,14 @@
-import importlib
 import os
 from unittest import skipIf
 
 import pytest
 from channels.db import database_sync_to_async
 from channels.routing import ProtocolTypeRouter
-from channels.testing import WebsocketCommunicator
 from django.conf import settings
-from django.contrib.auth import get_user_model, login
+from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Permission
-from django.http.request import HttpRequest
 from django.utils.module_loading import import_string
+from django_loci.tests.base.test_channels import BaseTestChannels
 from swapper import load_model
 
 from .utils import TestGeoMixin
@@ -23,48 +21,13 @@
 
 
 @skipIf(os.environ.get("SAMPLE_APP", False), "Running tests on SAMPLE_APP")
-class TestChannels(TestGeoMixin):
+class TestChannels(TestGeoMixin, BaseTestChannels):
     application = import_string(getattr(settings, "ASGI_APPLICATION"))
     object_model = Device
     location_model = Location
     object_location_model = DeviceLocation
     user_model = get_user_model()
 
-    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()
-        return request.session
-
-    async def _get_request_dict(self, pk=None, user=None):
-        if not pk:
-            location = await database_sync_to_async(self._create_location)(
-                is_mobile=True
-            )
-            await database_sync_to_async(self._create_object_location)(
-                location=location
-            )
-            pk = location.pk
-        path = "/ws/loci/location/{0}/".format(pk)
-        session = None
-        if user:
-            session = await database_sync_to_async(self._force_login)(user)
-        return {"pk": pk, "path": path, "session": session}
-
-    def _get_communicator(self, request_vars, user=None):
-        communicator = WebsocketCommunicator(self.application, request_vars["path"])
-        if user:
-            communicator.scope.update(
-                {
-                    "user": user,
-                    "session": request_vars["session"],
-                    "url_route": {"kwargs": {"pk": request_vars["pk"]}},
-                }
-            )
-        return communicator
-
     @pytest.mark.asyncio
     @pytest.mark.django_db(transaction=True)
     async def test_consumer_staff_but_no_change_permission(self):
@@ -74,8 +37,51 @@ async def test_consumer_staff_but_no_change_permission(self):
         location = await database_sync_to_async(self._create_location)(is_mobile=True)
         await database_sync_to_async(self._create_object_location)(location=location)
         pk = location.pk
-        request_vars = await self._get_request_dict(user=user, pk=pk)
-        communicator = self._get_communicator(request_vars, user)
+        request_vars = await self._get_specific_location_request_dict(user=user, pk=pk)
+        communicator = self._get_specific_location_communicator(request_vars, user)
+        connected, _ = await communicator.connect()
+        assert not connected
+        await communicator.disconnect()
+        # add permission to change location and repeat
+        perm = await database_sync_to_async(
+            (
+                await database_sync_to_async(Permission.objects.filter)(
+                    name="Can change location"
+                )
+            ).first
+        )()
+        await database_sync_to_async(user.user_permissions.add)(perm)
+        user = await database_sync_to_async(User.objects.get)(pk=user.pk)
+        request_vars = await self._get_specific_location_request_dict(user=user, pk=pk)
+        communicator = self._get_specific_location_communicator(request_vars, user)
+        connected, _ = await communicator.connect()
+        assert not connected
+        await communicator.disconnect()
+        # add user to organization
+        await database_sync_to_async(OrganizationUser.objects.create)(
+            organization=location.organization, user=user, is_admin=True
+        )
+        await database_sync_to_async(location.organization.save)()
+        user = await database_sync_to_async(User.objects.get)(pk=user.pk)
+        request_vars = await self._ge_get_specific_location_request_dictt_request_dict(
+            user=user, pk=pk
+        )
+        communicator = self._get_specific_location_communicator(request_vars, user)
+        connected, _ = await communicator.connect()
+        assert connected
+        await communicator.disconnect()
+
+    @pytest.mark.asyncio
+    @pytest.mark.django_db(transaction=True)
+    async def test_common_location_consumer_staff_but_no_change_permission(self):
+        user = await database_sync_to_async(User.objects.create_user)(
+            username="user", password="password", email="[email protected]", is_staff=True
+        )
+        location = await database_sync_to_async(self._create_location)(is_mobile=True)
+        await database_sync_to_async(self._create_object_location)(location=location)
+        pk = location.pk
+        request_vars = await self._get_common_location_request_dict(user=user, pk=pk)
+        communicator = self._get_common_location_communicator(request_vars, user)
         connected, _ = await communicator.connect()
         assert not connected
         await communicator.disconnect()
@@ -89,8 +95,8 @@ async def test_consumer_staff_but_no_change_permission(self):
         )()
         await database_sync_to_async(user.user_permissions.add)(perm)
         user = await database_sync_to_async(User.objects.get)(pk=user.pk)
-        request_vars = await self._get_request_dict(user=user, pk=pk)
-        communicator = self._get_communicator(request_vars, user)
+        request_vars = await self._get_common_location_request_dict(user=user, pk=pk)
+        communicator = self._get_common_location_communicator(request_vars, user)
         connected, _ = await communicator.connect()
         assert not connected
         await communicator.disconnect()
@@ -100,8 +106,8 @@ async def test_consumer_staff_but_no_change_permission(self):
         )
         await database_sync_to_async(location.organization.save)()
         user = await database_sync_to_async(User.objects.get)(pk=user.pk)
-        request_vars = await self._get_request_dict(user=user, pk=pk)
-        communicator = self._get_communicator(request_vars, user)
+        request_vars = await self._get_common_location_request_dict(user=user, pk=pk)
+        communicator = self._get_common_location_communicator(request_vars, user)
         connected, _ = await communicator.connect()
         assert connected
         await communicator.disconnect()

requirements.txt

@@ -3,6 +3,7 @@ django-reversion~=6.0.0
 django-taggit~=6.1.0
 netjsonconfig @ https://github.com/openwisp/netjsonconfig/archive/refs/heads/1.3.tar.gz
 django-x509 @ https://github.com/openwisp/django-x509/archive/refs/heads/1.4.tar.gz
+# Todo: Make path https://github.com/openwisp/django-loci/archive/refs/heads/1.3.tar.gz before merge
 django-loci @ https://github.com/openwisp/django-loci/tarball/feature/191-new-ws-endpoint-for-all-location
 django-flat-json-widget @ https://github.com/openwisp/django-flat-json-widget/archive/refs/heads/0.5.tar.gz
 openwisp-users @ https://github.com/openwisp/openwisp-users/archive/refs/heads/1.3.tar.gz