Django 频道在使用 AsyncWebsocketConsumer 进行测试时设置会话

Question

我正在尝试使用 Django Channels 2.4.0 构建一个简单的聊天应用程序，这是我的消费者代码：

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

我不确定当我使用AsyncWebsocketConsumer进行测试时如何设置会话字符串。我尝试了以下方法

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

我在网上收到以下错误：client.session["session_str"] = session_str：

django.core.exceptions.SyncOnlyOperation：你不能从异步上下文调用它 - 使用线程或sync_to_async

Answer 1

这对你来说可能太晚了，但对其他人来说：

我发现您可以手动设置会话。就我而言，我首先执行一个正常的HTTP请求，并期望此请求的会话在消费者中使用。

@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
# ...