Django connection_created signal is causing problems when testing

In my django application I have a list of notification types and I want to allow customers to subscribe to one or more notification types.

Each notification type has somewhat of a custom logic so the code of each notification has to be in a different class but I have created a singleton class that gathers all the notification type names and other metadata like description etc.

I want to have a list in the database of all the supported notification types so that the relationship between customers and notification types can be stored in the database while customers subscribe to notification types. I want to have a notification type table so that I can store the metadata and a separate table to store the many-to-many relationship between customers and notifications.

That is where connection_created signal comes in. I have created the following signal that creates the notification type items in the database when the connection_created signal is received so they get auto-updated when I am changing the code:

from django.db.backends.signals import connection_created
from django.db.backends.postgresql.base import DatabaseWrapper
from django.dispatch import receiver

from notification_type_singleton import NotificationTypeSingleton
from models import NotificationType


@receiver(connection_created, sender=DatabaseWrapper)
def create_or_update_notification_type(sender, **kwargs):
    exiting_ids = []
    for _, notification_type in (
        NotificationTypeSingleton._data.items()
    ):
        notification, _ = NotificationType.objects.update_or_create(
            name=notification_type.name,
            defaults={
                'description': notification_type.description,
                'is_active': True,
            },
        )
        exiting_ids.append(notification.id)

    # Deactivate all notifications in the database that are not used
    NotificationType.objects.exclude(id__in=exiting_ids).update(is_active=False)

    # Update the registry with the created events
    NotificationTypeSingleton._registry = {
        notification.name: notification
        for notification in NotificationType.objects.filter(is_active=True)
    }

That seems to work fine when I bring up my application with python manage.py runserver but I have the following issues when testing with pytest:

  • When using postgres, the signal is raised as expected but I get RuntimeWarning: Normally Django will use a connection to the 'postgres' database to avoid running initialization queries against the production database when it's not needed (for example, when running tests). Django was unable to create a connection to the 'postgres' database and will use the first PostgreSQL database instead. and if I comment out the code that is doing the queries in the signal then the error goes away.
  • When using sqlite the signal is not raised at all and I have to call it in the test setup

At the same time according to the docs the connection_created signal should be used for

...any post connection commands to the SQL backend.

so I think I am in the intended use cases but it seems that there are side effects? Does anybody else have had similar problems with testing?

I think this is a common issue when using the connection_created signal in tests. It happens for several reasons, such as:

  • Early firing so connection_created runs very early in Django’s DB setup.

  • Test DB creation so when pytest runs, Django first connects to the postgres admin DB to spin up test DBs.

  • ORM queries too soon so your handler tries to run NotificationType.objects.update_or_create() before the app DB is fully ready.

To fix that, you can use, post_migrate, something like this:

from django.db.models.signals import post_migrate
from django.dispatch import receiver
from notification_type_singleton import NotificationTypeSingleton
from models import NotificationType

@receiver(post_migrate)
def create_or_update_notification_types(sender, **kwargs):
    # Only run for your app to avoid running multiple times !!!!!!!
    if sender.name != 'your_app_name':
        return
        
    existing_ids = []
    for _, notification_type in NotificationTypeSingleton._data.items():
        notification, _ = NotificationType.objects.update_or_create(
            name=notification_type.name,
            defaults={
                'description': notification_type.description,
                'is_active': True,
            },
        )
        existing_ids.append(notification.id)

    # Deactivate all notifications in the database that are not used

    NotificationType.objects.exclude(id__in=existing_ids).update(is_active=False)

    # Update the registry with the created events
    NotificationTypeSingleton._registry = {
        notification.name: notification
        for notification in NotificationType.objects.filter(is_active=True)
    }
Вернуться на верх