Question 121788 in Redis Question posted in
The official documentation can be found here.

Redis – django channels WebsocketCommunicator TimeoutError

I am trying to run the following test:

tests.py

from rest_framework.test import APITestCase
from myapp.routing import application
from channels.testing import WebsocketCommunicator
from account.models import User
from rest_framework.authtoken.models import Token

class Tests(APITestCase):
    def setUp(self):
        self.user = User.objects.create(email='[email protected]', 
                                        password='a password')
        self.token, created = Token.objects.get_or_create(user=self.user)

    async def test_connect(self):
        communicator = WebsocketCommunicator(application, f"/ws/user/{self.token}/")
        connected, subprotocol = await communicator.connect()
        self.assertTrue(connected)
        await communicator.disconnect()

application is a boilerplate instance of channels.routing.ProtocolTypeRouter (like in here: https://channels.readthedocs.io/en/latest/topics/routing.html). Everything works fine in production. The test exits with the following error:

Traceback (most recent call last):
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/testing.py", line 74, in receive_output
    return await self.output_queue.get()
  File "/usr/lib/python3.7/asyncio/queues.py", line 159, in get
    await getter
concurrent.futures._base.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/sync.py", line 223, in __call__
    return call_result.result()
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 428, in result
    return self.__get_result()
  File "/usr/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/sync.py", line 292, in main_wrap
    result = await self.awaitable(*args, **kwargs)
  File "/home/projects/myapp/myapp-api/app/tests.py", line 35, in test_connect
    connected, subprotocol = await communicator.connect()
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/channels/testing/websocket.py", line 36, in connect
    response = await self.receive_output(timeout)
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/testing.py", line 85, in receive_output
    raise e
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/testing.py", line 74, in receive_output
    return await self.output_queue.get()
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/timeout.py", line 66, in __aexit__
    self._do_exit(exc_type)
  File "/home/projects/myapp/myapp-env/lib/python3.7/site-packages/asgiref/timeout.py", line 103, in _do_exit
    raise asyncio.TimeoutError
concurrent.futures._base.TimeoutError

----------------------------------------------------------------------
Ran 1 test in 1.026s

I have tried python versions 3.7.5, 3.8.0 and 3.9.9 using channels 3.0.4 with django 3.2.10 and channels-redis 3.3.1 ('BACKEND': 'channels_redis.core.RedisChannelLayer' in settings.py). The error persists. What am I doing wrong?

Tags:

2

Answers


  1. 0

    I had the same problem. APITestCase or TestCase dont allow transactions, you have to use SimpleTestCase from django test, and set databases to all. Just with that difference i think it will work.
    Note that the transactions will be saved between test, and not rolled back after the test.

    from django.test import SimpleTestCase
from myapp.routing import application
from channels.testing import WebsocketCommunicator
from account.models import User
from rest_framework.authtoken.models import Token


class Tests(SimpleTestCase):
    databases = '__all__'
    def setUp(self):
        self.user = User.objects.create(email='[email protected]', password='a password')
        self.token, created = Token.objects.get_or_create(user=self.user)

    async def test_connect(self):
        communicator = WebsocketCommunicator(application, f"/ws/user/{self.token}/")
        connected, subprotocol = await communicator.connect()
        self.assertTrue(connected)
        await communicator.disconnect()

    here are the information of SimpleTestCase
    https://docs.djangoproject.com/en/4.0/topics/testing/tools/

    Login or Signup to reply.
  2. 0

    I faced a similar issue and the solution is usually to mimic your production router for the tests too, i.e whatever middleware or additional component used in production should also be added when imstantiating your Communicator. for example in my asgi.py I have:

    application = ProtocolTypeRouter(
{
    "http": get_asgi_application(),
    "websocket": AllowedHostsOriginValidator(
        jwt_auth_middleware_stack(URLRouter(chat.routing.websocket_urlpatterns)),
    ),
}
)

    My communicator is instantiated as follows:

    communicator = WebsocketCommunicator(jwt_auth_middleware_stack(URLRouter(websocket_urlpatterns)),
                                     f"/ws/chat/{chat.id}/?token={token}")

    And my url is:

    websocket_urlpatterns = [
    path("ws/chat/<str:chat_id>/", consumers.AsyncChatConsumer.as_asgi())
]
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search