使用WebSocketCommunicator的Django通道v2测试中的身份验证

from channels.generic.websocket import WebsocketConsumer
from django.contrib.auth import get_user_model

class MyConsumer(WebsocketConsumer):
    def connect(self):
        # 进行身份验证
        user = self.authenticate(self.scope["user"])
        if user is None:
            # 身份验证失败，关闭连接
            self.close()
        else:
            # 身份验证成功，接受连接
            self.accept()

    def authenticate(self, user):
        # 自定义身份验证逻辑
        # 返回经过身份验证的用户对象或None
        UserModel = get_user_model()
        try:
            return UserModel.objects.get(username=user.username)
        except UserModel.DoesNotExist:
            return None

from channels.testing import WebsocketCommunicator
from myapp.consumers import MyConsumer

async def test_websocket_communicator():
    # 创建WebSocketCommunicator对象
    communicator = WebsocketCommunicator(MyConsumer.as_asgi(), "/ws/")

    # 进行身份验证
    user = await authenticate_user()  # 自定义的身份验证方法
    communicator.scope["user"] = user

    # 连接到WebSocket
    connected, _ = await communicator.connect()

    # 断言连接成功
    assert connected

    # 发送和接收消息
    await communicator.send_json({"message": "Hello"})
    response = await communicator.receive_json()

    # 断言接收到正确的响应
    assert response["message"] == "Hello, World!"

    # 关闭连接
    await communicator.disconnect()