django - django channel 单元测试的假数据

Question

bounty 6 天后到期。这个问题的答案有资格获得 +50 声望奖励。
A.Mohammadi正在寻找规范的答案:

I just want you to tell me how to generate fake data for my websocket consumers. I've tried hard but got nowhere.

The team is struggling and we really need your help! The official docs of django channels don't explain all aspects of testing in detail. The app is growing larger and larger everyday and manual testing becomes less and less efficient with each passing day.

我之前问过一个关于这个话题的问题，但最后我放弃了，因为似乎没有办法......
但是现在我真的真的真的需要为我的 django channel 使用者编写单元测试，因为应用程序的大小越来越大，手动测试不再有效。所以我决定再问一个问题，这次我会尽力解释情况。
主要问题是 “生成虚假数据” .我正在使用 factory_boy和 faker一起为我的测试生成假数据。当我生成假数据时，可以从 TestCase 内部访问它本身，但不能在消费者内部访问。让我通过一个例子向你展示，考虑下面的代码:
test_consumers.py

from chat.models import PersonalChatRoom
from users.models import User
from django.test import TestCase
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(TestCase, 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)

消费者.py

...
class PersonalChatConsumer(
        ChatRoomManagementMixin,
        MessageManagementMixin,
        JsonWebsocketConsumer
    ):
    message_serializer_class = PersonalMessageSerializer
    chat_room_class = PersonalChatRoom

    def connect(self):
     -> print("consumer (user): ", User.objects.all())
     -> print("consumer (personal chat room): ", PersonalChatRoom.objects.all())
        return super().connect() # some magic here
    
    ...

输出
我在代码的 3 个不同部分打印出数据库的内容:

内setUp TestPersonalChatRoomConsumer 中的方法类(class)

内test_personal_chat_room_connection TestPersonalChatRoomConsumer 中的方法类(class)

内connect消费者的方法

我希望结果是相同的，但这是运行测试时的真实输出:

setup: (user):  <QuerySet [<User: joshua03@cervantes.net>, <User: pgolden@cummings.com>]>
setup: (personal chat room):  <QuerySet [<PersonalChatRoom: PV joshua03@cervantes.net and pgolden@cummings.com>]>
test (user):  <QuerySet [<User: joshua03@cervantes.net>, <User: pgolden@cummings.com>]>
test (personal chat room):  <QuerySet [<PersonalChatRoom: PV joshua03@cervantes.net and pgolden@cummings.com>]>

...

consumer (user):  <QuerySet []> # There are no users in the database
consumer (personal chat room):  <QuerySet []> # There are no chat rooms in the database

如您所见，第一部分包含由 factory_boy 生成的假数据。但第二节 包含一个空的查询集
如何重现问题
这真的很简单:

创建一个简单的模型

创建消费者

创建一个测试用例并在测试中创建该模型的一个实例

尝试访问消费者内部新创建的实例，你会惊讶于一个空的查询集。

为什么里面没有生成数据 setUp消费者内部可访问的方法？
以下是我个人认为导致问题的原因:

数据库事务可能有问题

也许是关于数据库连接

也许消费者正在使用另一个数据库

如果您需要更多信息，请在下方留言。我会尽快提供所有这些。谢谢你们。
Github 最小示例
我提供了演示问题的最基本的项目，这样您就不必自己重现它。
这是链接: REPO
额外信息:

数据库:postgresql v13(psycopg2 后端)

操作系统:适用于 Linux 的 Windows 子系统 (WSL) ubuntu 20.04

python解释器版本:3.8.10

编辑
这是 PersonalChatRoomFactory的代码:

class PersonalChatRoomFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = PersonalChatRoom

    user_1 = factory.SubFactory(UserFactory)
    user_2 = factory.SubFactory(UserFactory)

它没有做任何事情，它的目的只是为 user_1 创建 2 个用户。和 user_2领域。这是 UserFactory的代码:

user_pass = 'somepassword'

class UserFactory(factory.django.DjangoModelFactory):
    class Meta:
        model = User

    username = factory.Sequence(lambda n: faker.unique.first_name())
    email = factory.Sequence(lambda n: faker.unique.email())
    full_name = factory.Sequence(lambda n: faker.name())
    bio = factory.Sequence(lambda n: faker.text())
    date_verified = timezone.now()
    
    @classmethod
    def _create(cls, model_class, *args, **kwargs):
        """ Just hashes the raw password when creating the user """
        user =  super()._create(model_class, *args, **kwargs)
        user.set_password(kwargs.get('password', user_pass))
        user.save()
        return user

编辑 2
这是我的 User 的代码模型:

class User(AbstractBaseUser, PermissionsMixin):
    class Meta:
        verbose_name = 'user'
        verbose_name_plural = 'users'

    email = models.EmailField(unique=True)
    username = CidField()
    full_name = models.CharField(max_length=255, blank=True, null=True)
    bio = models.CharField(max_length=255, blank=True, null=True)
    flagged_by = models.ManyToManyField('User', blank=True, related_name='flagged_users')
    profile_image = models.ImageField(upload_to='profile-images/', blank=True, null=True)
    date_birth = models.DateField(blank=True, null=True)
    is_woman = models.BooleanField(default=None, null=True, blank=True)
    major = models.ForeignKey(Major, on_delete=models.SET_NULL, blank=True, null=True, related_name='users')
    date_verified = models.DateTimeField(blank=True, null=True)
    verification_code = models.CharField(max_length=255, blank=True, null=True)
    date_verification_sent = models.DateTimeField(blank=True, null=True)
    date_joined = models.DateTimeField(auto_now_add=True)
    can_own_movement = models.BooleanField('can user own a movemnet',default=False)
    online_devices = models.IntegerField(default=0)

    USERNAME_FIELD = 'email'

    objects = UserManager()
    # ... (some convenience methods and dynamic properties)

它有一个自定义管理器:

class UserManager(BaseUserManager):
    use_in_migrations = True

    def _create_user(self, email, password, **extra_fields):
        """
        Creates and saves a User with the given email and password.
        """
        if not email:
            raise ValueError('email field is required')
        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.password = password
        user.save(using=self._db)
        return user

    def create_user(self, email, password=None, **extra_fields):
        extra_fields.setdefault('is_superuser', False)
        return self._create_user(email, password, **extra_fields)

    def create_superuser(self, email, password, **extra_fields):
        extra_fields.setdefault('is_superuser', True)

        if extra_fields.get('is_superuser') is not True:
            raise ValueError('Superuser must have is_superuser=True.')
        password = make_password(password)

        return self._create_user(email, password, **extra_fields)

这是 PersonalChatRoom 的代码模型:

class PersonalChatRoom(models.Model):
    user_1 = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='personal_chat_rooms')
    user_2 = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, related_name='personal_chat_room_contacts')
    visibility_stat = models.IntegerField(default=ChatRoomVisibilityStat.BOTH)

    def clean(self) -> None:
        if (not self.pk and self.are_users_connected(self.user_1, self.user_2)):
            raise ValidationError(
                "a chat room already exists between these two users", code="room_already_exists")
        return super().clean()
    # ... (some convenience methods)

Answer 1

问题:由于 ，您面临此数据丢失问题异步 调用。
解决方案:在 django.test 有一个名为 的测试类交易测试案例 .通过使用它，我们可以克服 异步 数据缺失问题。
进行以下更改，您就可以开始了:
测试文件
替换 TestCase与 TransactionTestCase你准备好了。

from django.test import TransactionTestCase
class TestTheTestConsumer(TransactionTestCase):

输出:

python manage.py test
Creating test database for alias 'default'...
System check identified no issues (0 silenced).
User (test.setUp):  <QuerySet [<User: someuser1>, <User: someuser2>]>
User (test.test_users_are_listed_correctly):  <QuerySet [<User: someuser1>, <User: someuser2>]>
User (consumer.connect):  <QuerySet [<User: someuser1>, <User: someuser2>]>
User (consumer.send_user_list):  <QuerySet [<User: someuser1>, <User: someuser2>]>
User (results):  [{'username': 'someuser1', 'id': 1}, {'username': 'someuser2', 'id': 2}]