I have recently started learning Django Channels and found this very nice and detailed project: Real-Time Taxi App With Django Channels and React. However, this project uses Django Channels v2.3.1 and I have v3.0.3 so I made the changes to make it work the latest version. The code below is supposed to be working in the latest version but it doesn't.
What am I doing wrong?
test_websocket.py
import pytest
from channels.db import database_sync_to_async
from channels.layers import get_channel_layer
from channels.testing import WebsocketCommunicator
from django.contrib.auth import get_user_model
from rest_framework_simplejwt.tokens import AccessToken
from taxi.asgi import application
TEST_CHANNEL_LAYERS = {
'default': {
'BACKEND': 'channels.layers.InMemoryChannelLayer'
}
}
@database_sync_to_async
def create_user(username, password):
user = get_user_model().objects.create_user(
username=username,
password=password
)
access = AccessToken.for_user(user)
return user, access
@pytest.mark.asyncio
@pytest.mark.django_db(transaction=True)
class TestWebSocket:
async def test_can_connect_to_server(self, settings):
settings.CHANNEL_LAYERS = TEST_CHANNEL_LAYERS
_, access = await create_user('test.user@example.com', 'pAssw0rd')
communicator = WebsocketCommunicator(application=application, path=f'/taxi/?token={access}')
connected, _ = await communicator.connect()
assert connected is True
await communicator.disconnect()
async def test_can_send_and_receive_messagges(self, settings):
settings.CHANNEL_LAYERS = TEST_CHANNEL_LAYERS
communicator = WebsocketCommunicator(
application=application,
path='/taxi/'
)
connected, _ = await communicator.connect()
message = {
'type': 'echo.message',
'data': 'This is a test message.'
}
await communicator.send_json_to(message)
response = await communicator.receive_json_from()
assert response == message
await communicator.disconnect()
async def test_can_send_and_receive_broadcast_messages(self, settings):
settings.CHANNEL_LAYERS = TEST_CHANNEL_LAYERS
communicator = WebsocketCommunicator(
application=application,
path='/taxi/'
)
connected, _ = await communicator.connect()
message = {
'type': 'echo.message',
'data': 'This is a test message.',
}
channel_layer = get_channel_layer()
await channel_layer.group_send('test', message=message)
response = await communicator.receive_json_from()
assert response == message
await communicator.disconnect()
async def test_cannot_connect_to_socket(self, settings):
settings.CHANNEL_LAYERS = TEST_CHANNEL_LAYERS
communicator = WebsocketCommunicator(
application=application,
path='/taxi/'
)
connected, _ = await communicator.connect()
assert connected is False
await communicator.disconnect()
middleware.py
from urllib.parse import parse_qs
from channels.auth import AuthMiddlewareStack
from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.db import close_old_connections
from rest_framework_simplejwt.tokens import AccessToken
User = get_user_model()
@database_sync_to_async
def get_user(user_id):
try:
return User.objects.get(id=user_id)
except User.DoesNotExist:
return AnonymousUser()
class TokenAuthMiddleware(BaseMiddleware):
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
close_old_connections()
query_string = parse_qs(scope['query_string'].decode())
token = query_string.get('token')
if not token:
scope['user'] = AnonymousUser()
return await self.app(scope, receive, send)
scope['user'] = await get_user(AccessToken(token[0])['id'])
return await self.app(scope, receive, send)
# return await super().__call__(scope, receive, send)
def TokenAuthMiddlewareStack(inner):
return TokenAuthMiddleware(AuthMiddlewareStack(inner))
The error:
====================================================== FAILURES =======================================================
_____________________________ TestWebSocket.test_can_send_and_receive_broadcast_messages ______________________________
self = <channels.testing.websocket.WebsocketCommunicator object at 0x0000017CAA4B6AF0>, timeout = 1
async def receive_output(self, timeout=1):
"""
Receives a single message from the application, with optional timeout.
"""
# Make sure there's not an exception to raise from the task
if self.future.done():
self.future.result()
# Wait and receive the message
try:
async with async_timeout(timeout):
> return await self.output_queue.get()
..\venv\lib\site-packages\asgiref\testing.py:74:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Queue at 0x17caa4b62e0 maxsize=0 tasks=1>
async def get(self):
"""Remove and return an item from the queue.
If queue is empty, wait until an item is available.
"""
while self.empty():
getter = self._loop.create_future()
self._getters.append(getter)
try:
> await getter
E asyncio.exceptions.CancelledError
E:\ProgramFiles\Python\lib\asyncio\queues.py:163: CancelledError
During handling of the above exception, another exception occurred:
self = <trips.tests.test_websocket.TestWebSocket object at 0x0000017CAA461FA0>
settings = <pytest_django.fixtures.SettingsWrapper object at 0x0000017CAA4CAEB0>
async def test_can_send_and_receive_broadcast_messages(self, settings):
settings.CHANNEL_LAYERS = TEST_CHANNEL_LAYERS
communicator = WebsocketCommunicator(
application=application,
path='/taxi/'
)
connected, _ = await communicator.connect()
message = {
'type': 'echo.message',
'data': 'This is a test message.',
}
channel_layer = get_channel_layer()
await channel_layer.group_send('test', message=message)
> response = await communicator.receive_json_from()
trips\tests\test_websocket.py:67:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\venv\lib\site-packages\channels\testing\websocket.py:93: in receive_json_from
payload = await self.receive_from(timeout)
..\venv\lib\site-packages\channels\testing\websocket.py:72: in receive_from
response = await self.receive_output(timeout)
..\venv\lib\site-packages\asgiref\testing.py:85: in receive_output
raise e
..\venv\lib\site-packages\asgiref\testing.py:74: in receive_output
return await self.output_queue.get()
..\venv\lib\site-packages\asgiref\timeout.py:66: in __aexit__
self._do_exit(exc_type)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <asgiref.timeout.timeout object at 0x0000017CAA4B6310>, exc_type = <class 'asyncio.exceptions.CancelledError'>
def _do_exit(self, exc_type: Type[BaseException]) -> None:
if exc_type is asyncio.CancelledError and self._cancelled:
self._cancel_handler = None
self._task = None
> raise asyncio.TimeoutError
E asyncio.exceptions.TimeoutError
..\venv\lib\site-packages\asgiref\timeout.py:103: TimeoutError
=============================================== short test summary info ===============================================
FAILED trips/tests/test_websocket.py::TestWebSocket::test_can_send_and_receive_broadcast_messages - asyncio.exception...
============================================= 1 failed, 7 passed in 2.58s =============================================