升级至Django 2与Channels 2后，测试中数据库访问异常求助

咱们先把核心问题捋清楚：pytest-django 默认会用数据库事务包裹每个测试，测试结束后自动回滚，而 Channels 的异步消费者运行在独立的事件循环上下文里，根本看不到事务里还没提交的对象。这就是为啥你在 fixture、setup 或者测试函数里创建了用户，消费者却认不出来的根源。下面给你几个落地可行的解决思路：

1. 改用事务性测试标记/Fixture #

放弃默认的事务隔离机制，让测试中的数据库操作直接提交到数据库，这样消费者就能读到真实存在的数据了。

你可以用 @pytest.mark.django_db(transaction=True) 标记测试函数，或者直接使用 transactional_db fixture：

import pytest
from channels.testing import WebsocketCommunicator
from django.contrib.auth.models import User
from myapp.consumers import ChatConsumer
from asgiref.sync import async_to_sync

@pytest.mark.django_db(transaction=True)
def test_chat_consumer_recognizes_user():
    # 创建用户并保存（此时会直接提交到数据库）
    user = User.objects.create_user(username="testuser", password="testpass")
    
    # 初始化Communicator，手动注入用户到scope中
    communicator = WebsocketCommunicator(ChatConsumer.as_asgi(), "/ws/chat/")
    communicator.scope["user"] = user
    
    # 处理连接
    connected, subprotocol = async_to_sync(communicator.connect)()
    
    # 这里写你的断言逻辑，比如检查用户是否被正确识别
    assert connected
    # ... 后续测试步骤
    
    async_to_sync(communicator.disconnect)()

2. 在测试中手动提交事务 #

如果你不想全局修改测试标记，也可以在创建对象后手动提交事务，强制数据持久化：

from django.db import transaction
from channels.testing import WebsocketCommunicator
from django.contrib.auth.models import User
from asgiref.sync import async_to_sync

def test_chat_consumer_with_manual_commit():
    user = User.objects.create_user(username="testuser", password="testpass")
    # 手动提交事务，让消费者能读到这条数据
    transaction.commit()
    
    # 初始化Communicator并注入用户
    communicator = WebsocketCommunicator(ChatConsumer.as_asgi(), "/ws/chat/")
    communicator.scope["user"] = user
    
    connected, _ = async_to_sync(communicator.connect)()
    # ... 测试逻辑

3. 确保消费者的认证逻辑匹配测试场景 #

有时候不是数据库的问题，而是消费者的认证逻辑没处理好。比如你的消费者依赖 session 认证，那测试里要正确设置 scope 的 session 信息：

def test_consumer_with_session_auth():
    user = User.objects.create_user(username="testuser", password="testpass")
    transaction.commit()
    
    communicator = WebsocketCommunicator(ChatConsumer.as_asgi(), "/ws/chat/")
    # 模拟登录后的session数据
    communicator.scope["session"]["_auth_user_id"] = str(user.id)
    communicator.scope["session"].save()
    
    connected, _ = async_to_sync(communicator.connect)()
    # ... 测试逻辑

如果是 Token 认证，记得在连接时带上 Token 参数：

from rest_framework.authtoken.models import Token

def test_consumer_with_token_auth():
    user = User.objects.create_user(username="testuser", password="testpass")
    token = Token.objects.create(user=user)
    transaction.commit()
    
    # 连接时把Token放到query参数里
    communicator = WebsocketCommunicator(ChatConsumer.as_asgi(), f"/ws/chat/?token={token.key}")
    
    connected, _ = async_to_sync(communicator.connect)()
    # ... 测试逻辑

另外要注意：尽量不要在 autouse 的 fixture 里处理事务，避免不同测试之间的上下文干扰，每个测试里明确处理事务会更稳妥。

内容的提问来源于stack exchange，提问作者Quba