Django Channels Set Session When Testing With AsyncWebsocketConsumer

user_5844532 0
Viewed 147

I am trying to build a simple chat app using Django Channels 2.4.0, here is my consumer code:

class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
    if "session_str" in self.scope["session"]:
        # get chat history 
    else:
        #create a new chat room
    await self.accept()

I am not sure how I can set the session string when I'm test with AsyncWebsocketConsumer. I have tried below method

@pytest.mark.asyncio
async def test_ws_connection(self):
""" Test initial websocket connection
"""
start_datetime = datetime.datetime.now()
test_room = Room(create_datetime=start_datetime)
session_str = test_room.session_str
client = Client()
client.session["session_str"] = session_str
application = URLRouter([
    url(r'ws/chat/$', ChatConsumer),
])
communicator = WebsocketCommunicator(application, "ws/chat/")
connected, _ = await communicator.connect()
assert connected
await communicator.disconnect()

I got the following error on line: client.session["session_str"] = session_str:

django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async

django-channels testing django
1 Answers

This is probably too late for you but for everyone else:

I found that you can set the session by hand. In my case I first do a normal HTTP request and expect the session of this request to be used in the consumer.

@pytest.mark.asyncio
@pytest.mark.django_db
async def test_socket_connecting(async_client):
    # do a HTTP call
    response = await async_client.get(f"/game/")
    assert response.status_code == 200

    # Build webscocket communicator
    application = URLRouter([
        path("ws/game/", GameConsumer.as_asgi()),
    ])
    communicator = WebsocketCommunicator(application, f"/ws/game/")
    # Set session of communicator
    communicator.scope["session"] = async_client.session
    # Do actual test
    # ...
user_12244670
Related