django 频道单元测试的假数据

Question

我之前问过一个关于这个话题的问题，但最后我放弃了，因为似乎没有办法......

但是现在我真的真的需要为我的 django 频道消费者编写单元测试，因为应用程序的规模越来越大，而手动测试不再有效。所以我决定再问一个问题，这次我会尽力说明情况。

主要问题是“生成虚假数据”。我将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)

consumers.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 个不同部分打印出数据库的内容：

• 在TestPersonalChatRoomConsumer 类的setUp 方法内
• 在TestPersonalChatRoomConsumer 类的test_personal_chat_room_connection 方法内
• 在消费者的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 生成的假数据，但第二部分包含一个空查询集

如何重现问题

其实很简单：

1. 创建一个简单的模型
2. 创建消费者
3. 创建一个测试用例并在测试中创建该模型的实例
4. 尝试访问消费者内部新创建的实例，您会惊讶于空查询集。

为什么 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 和user_2 字段创建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中有一个名为TransactionTestCase的测试类。通过使用它，我们可以克服 异步 数据丢失问题。

进行以下更改，一切就绪：

test.py

将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}]