Сигнал Django connection_created вызывает проблемы при тестировании

В моем приложении django у меня есть список типов уведомлений, и я хочу разрешить клиентам подписываться на один или несколько типов уведомлений.

У каждого типа уведомлений есть своя собственная логика, поэтому код каждого уведомления должен быть в другом классе, но я создал одноэлементный класс, который собирает все имена типов уведомлений и другие метаданные, такие как описание и т.д.

Я хочу иметь в базе данных список всех поддерживаемых типов уведомлений, чтобы взаимосвязь между клиентами и типами уведомлений могла храниться в базе данных, пока клиенты подписываются на типы уведомлений. Я хочу иметь таблицу типов уведомлений, чтобы я мог хранить метаданные, и отдельную таблицу для хранения взаимосвязи "многие ко многим" между клиентами и уведомлениями.

Именно здесь поступает сигнал connection_created. Я создал следующий сигнал, который создает элементы типа уведомления в базе данных при получении сигнала connection_created, чтобы они автоматически обновлялись, когда я меняю код:

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

Это, кажется, работает нормально, когда я запускаю свое приложение с помощью python manage.py runserver, но при тестировании с помощью pytest:

  • При использовании postgres сигнал выдается, как и ожидалось, но я получаю 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., и если я закомментирую код, который выполняет запросы в сигнале, ошибка исчезнет.
  • При использовании sqlite сигнал вообще не подается, и я должен вызвать его в тестовой настройке

В то же время, согласно документации, сигнал connection_created должен использоваться для

...любые команды post-подключения к серверной части SQL.

итак, я думаю, что у меня есть предполагаемые варианты использования, но, похоже, есть побочные эффекты? У кого-нибудь еще были подобные проблемы с тестированием?

Я думаю, что это распространенная проблема при использовании сигнала connection_created в тестах. Это происходит по нескольким причинам, таким как:

  • Ранний запуск, поэтому connection_created запускается очень рано при настройке базы данных Django.

  • Создание тестовой базы данных таким образом, при запуске pytest Django сначала подключается к базе данных администратора postgres, чтобы запустить тестовые базы данных.

  • ORM запрашивает слишком рано, поэтому ваш обработчик пытается запустить NotificationType.objects.update_or_create() до того, как база данных приложений будет полностью готова.

Чтобы исправить это, вы можете использовать, post_migrate, что-то вроде этого:

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)
    }
Вернуться на верх