Django Channels单元测试setUp生成的假数据在consumer中无法访问

问题成因 #

• Django 原生 TestCase 会将每个测试用例的所有数据库操作包裹在独立的事务中，测试运行过程中事务不会对外提交，测试结束后直接回滚，避免测试数据污染正式库。
• Django Channels 的 Websocket 消费者运行在独立的异步事件循环/线程中，使用单独的数据库连接，无法读取到测试事务中未提交的数据，因此查询返回空结果。

解决方案 #

最简便的方案是改用 Channels 官方提供的 TransactionTestCase 替代原生的 TestCase，这个测试基类会自动将测试数据提交到数据库，确保消费者的数据库连接可以正常读取到测试数据，修改方式如下：

# 替换原有导入
# from django.test import TestCase
from channels.testing import TransactionTestCase
from chat.models import PersonalChatRoom
from users.models import User
from channels.testing import WebsocketCommunicator
from asgiref.sync import sync_to_async
from users.tests.test_setup import TestUtilsMixin
from astra_backend.asgi import application
from chat.tests.model_factory import PersonalChatRoomFactory


# 修改父类继承
class TestPersonalChatRoomConsumer(TransactionTestCase, TestUtilsMixin):
    # 原有测试逻辑完全不需要调整
    def setUp(self) -> None:
        super().setUp()
        self.chat_room = PersonalChatRoomFactory()
        self.u1 = self.chat_room.user_1
        self.u2 = self.chat_room.user_2
        print("setup: (user): ", User.objects.all())
        print("setup: (personal chat room): ", PersonalChatRoom.objects.all())

    async def test_personal_chat_room_connection(self):
        await sync_to_async(print)("test (user): ", User.objects.all())
        await sync_to_async(print)("test (personal chat room): ", PersonalChatRoom.objects.all())

        com = WebsocketCommunicator(application, f'chat/personal/{self.chat_room.pk}/')
        connected, _ = await com.connect()
        self.assertTrue(connected)

补充说明 #

• TransactionTestCase 会在测试结束后自动清空测试数据库的所有表，不需要手动清理测试数据。
• 如果多个测试用例依赖同一份基础数据，可以将数据创建逻辑放到 setUpTestData 方法中，提升测试运行效率。

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