Repository: openwisp/openwisp-network-topology (language: python)

Issue:
[bug] Websocket doesn't take into account frontend visualizer
There's a non admin view in which the websocket code doesn't work because it doesn't take it into account.
[bug] Websocket doesn't take into account frontend visualizer
There's a non admin view in which the websocket code doesn't work because it doesn't take it into account.


Discussion / hints:

Fix patch:
diff --git a/openwisp_network_topology/routing.py b/openwisp_network_topology/routing.py
index 3015ca03..4d34c022 100644
--- a/openwisp_network_topology/routing.py
+++ b/openwisp_network_topology/routing.py
@@ -3,8 +3,13 @@
 from . import consumers
 
 websocket_urlpatterns = [
+    # For topology admin view
     re_path(
         r'^admin/topology/topology/(?P<pk>[^/]+)/change/$',
         consumers.TopologyConsumer.as_asgi(),
-    )
+    ),
+    # For topology non admin view
+    re_path(
+        r'^topology/topology/(?P<pk>[^/]+)/$', consumers.TopologyConsumer.as_asgi()
+    ),
 ]


Test patch:
diff --git a/openwisp_network_topology/tests/test_websockets.py b/openwisp_network_topology/tests/test_websockets.py
index fd86791b..55a7c6b3 100644
--- a/openwisp_network_topology/tests/test_websockets.py
+++ b/openwisp_network_topology/tests/test_websockets.py
@@ -15,16 +15,24 @@
 @pytest.mark.asyncio
 @pytest.mark.django_db(transaction=True)
 class TestTopologySockets(CreateGraphObjectsMixin, CreateOrgMixin):
-    application = import_string(getattr(settings, 'ASGI_APPLICATION'))
-    topology_model = Topology
     node_model = Node
     link_model = Link
+    topology_model = Topology
+    application = import_string(getattr(settings, 'ASGI_APPLICATION'))
+
+    async def _get_url(self, topology_id, admin_view):
+        non_admin_view_url = f'topology/topology/{topology_id}/'
+        admin_view_url = f'admin/topology/topology/{topology_id}/change/'
+        if not admin_view:
+            return non_admin_view_url
+        return admin_view_url
 
-    async def _get_communicator(self, admin_client, topology_id):
+    async def _get_communicator(self, admin_client, topology_id, admin_view=True):
+        url = await self._get_url(topology_id, admin_view)
         session_id = admin_client.cookies['sessionid'].value
         communicator = WebsocketCommunicator(
             self.application,
-            f'admin/topology/topology/{topology_id}/change/',
+            url,
             headers=[
                 (
                     b'cookie',
@@ -127,3 +135,13 @@ async def test_topology_properties_update(self, admin_user, admin_client):
         assert response['topology'] is not None
         assert response['topology'] == expected_response
         await communicator.disconnect()
+
+    async def test_non_admin_view_consumer_connection(self, admin_user, admin_client):
+        org = await database_sync_to_async(self._create_org)()
+        t = await database_sync_to_async(self._create_topology)(organization=org)
+        communicator = await self._get_communicator(
+            admin_client, t.pk, admin_view=False
+        )
+        connected, _ = await communicator.connect()
+        assert connected is True
+        await communicator.disconnect()