Fake data for django channels unit tests

Question

I previously asked a question regarding this topic but finally I gave up on that because there seemed to be no way ...

But now I really really really need to write unit tests for my django channel consumers because the application is growing larger in size and manual testing isn't efficient anymore. So I decided to ask another question and this time I'm going to do my best to explain the situation.

The main problem is "Generating Fake Data". I'm using factory_boy and faker together in order to generate fake data for my tests. When I generate fake data, it is accessible from inside the TestCase itself but is not accessible inside the consumer. Let me show you by an example, consider the code below:

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
    
    ...

output

I'm printing out the contents of the database in 3 different sections of the code:

• inside the setUp method in the TestPersonalChatRoomConsumer class
• inside the test_personal_chat_room_connection method in the TestPersonalChatRoomConsumer class
• inside the connect method of the consumer

I expected the results to be identical but here is the real output when running the test:

setup: (user):  <QuerySet [<User: [email protected]>, <User: [email protected]>]>
setup: (personal chat room):  <QuerySet [<PersonalChatRoom: PV [email protected] and [email protected]>]>
test (user):  <QuerySet [<User: [email protected]>, <User: [email protected]>]>
test (personal chat room):  <QuerySet [<PersonalChatRoom: PV [email protected] and [email protected]>]>

...

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

As you can see, the first section contains the fake data generated by factory_boy but the second section contains an empty queryset

How to reproduce the problem

It is really simple:

1. create a simple model
2. create a consumer
3. create a test case and create an instance of that model inside tests
4. try to access the newly created instance inside the consumer and you'll be surprised with an empty queryset.

Why isn't the data generated inside setUp method accessible inside the consumer?

Here is what I think causes the problem personally:

• there might be something wrong with database transactions
• maybe it is about database connections
• maybe the consumer is using another database

if you need more information please leave a comment down below. I will provide all of them as soon as possible. Thank you all.

Github minimal example

I have provided the most basic project that demonstrates the problem so that you don't have to reproduce it yourself.

Here is the link: REPO

extra info:

• database: postgresql v13 (psycopg2 backend)
• OS: Windows Subsystem for Linux (WSL) ubuntu 20.04
• python interpreter version: 3.8.10

Edit

Here is the code for PersonalChatRoomFactory:

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

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

It doesn't do anything that much, its purpose is to only create 2 users for user_1 and user_2 fields. Here is the code for 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

Edit 2

Here is the code for my User model:

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)

And it has a custom manager:

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)

And here is the code for the PersonalChatRoom model:

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

Issue: You face this data missing issue because of asynchronous calls.

Solution: In django.test there is a test class called TransactionTestCase. By using this we can overcome that asynchronous data missing issue.

Make following changes and you are all set to go:

test.py

Replace TestCase with TransactionTestCase and you are all set to go.

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

Output:

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