django channels WebsocketCommunicator TimeoutError

Question

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='test@test.test', 
                                        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?

The consumer never called `self.accept()` because the `Token` and `User` instances are somehow not accessible in the consumer (empty QuerySets). https://stackoverflow.com/questions/56175352/test-database-not-keeping-data-during-django-channels-pytest-test is a related question. The posted solution there with pytest and `@pytest.mark.django_db(transaction=True)` and `@pytest.mark.asyncio` did not work for me. This is too complicated. I am giving up now. — Call of Guitar, Dec 29 '21 at 12:05

score 0 · Answer 1 · answered May 09 '22 at 13:21

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='test@test.test', 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/

score 0 · Answer 2 · edited Dec 28 '22 at 14:02

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())
]

django channels WebsocketCommunicator TimeoutError

2 Answers2