Django Channels测试套件与Consumer使用不同数据库问题求助

Django 默认的 TestCase 测试类会将每个测试方法的逻辑包裹在独立的数据库事务中，测试过程中产生的所有数据都不会实际提交到测试数据库，仅对当前事务所在的线程可见。而 Channels Consumer 运行在独立的异步上下文/线程中，默认使用的数据库连接和测试线程的事务隔离，因此只能读到原始数据库的内容，看不到测试生成的临时数据。

方案1：更换测试基类（最便捷，优先推荐） #

将测试类继承的 django.test.TestCase 替换为 django.test.TransactionTestCase：

from django.test import TransactionTestCase

class TestPersonalChatConsumer(TransactionTestCase):
    # 剩下的代码不需要改动
    def setUp(self) -> None:
        self.chat_room = PersonalChatRoomFactory()
        self.user_1 = self.chat_room.user_1
        self.user_2 = self.chat_room.user_2
        return super().setUp()

    async def test_we_can_get_messsages(self):
        # 原有逻辑不变

TransactionTestCase 不会用事务包裹测试逻辑，你在测试中创建的数据会直接提交到测试数据库，Consumer 就能正常访问到这些数据，测试执行完成后框架会自动清空测试数据库的所有数据，不会产生脏数据。

方案2：强制同步数据库操作共享线程 #

如果你需要继续使用 TestCase 的事务特性，可以将所有涉及数据库操作的 sync_to_async 调用的 thread_sensitive 参数设置为 True，强制所有同步代码运行在同一个线程，共享测试线程的数据库连接：

• 测试代码中调用数据库的 sync_to_async 增加参数：

  @sync_to_async(thread_sensitive=True)
  def generate_messages(self):
      # 原有逻辑不变
  

• Consumer 中所有封装数据库操作的同步方法调用，同样增加 thread_sensitive=True 参数：

  # 比如你原有调用sync的地方修改为
  from asgiref.sync import sync_to_async
  contact = await sync_to_async(self.get_contact_user, thread_sensitive=True)(contact_id)
  

你当前的测试代码没有给 Websocket 请求传入认证信息，Consumer 从 scope['user'] 拿到的会是匿名用户，建议构造 WebsocketCommunicator 时手动注入测试用户信息，跳过不必要的认证逻辑校验：

com = WebsocketCommunicator(application, f'/chat/personal/{self.user_1.pk}/')
# 直接注入用户到scope
com.scope['user'] = self.user_1
connected, _ = await com.connect()

内容的提问来源于stack exchange，提问作者A.Mohammadi