Запустите команду django в течение установленного времени внутри теста

Обновление 21.04.23

Ниже приведен рабочий тест благодаря ответу Пола Гилмартина (создатель django-pgpubsub). Я переименовал свою модель уведомлений при импорте в EmailNotification, чтобы избежать путаницы, поскольку это то же имя, что и у модели уведомлений в django-pgpubsub:

@pytest.mark.django_db(transaction=True)  # needed to work
def test_shipment_status_notificiations(testuser, pg_connection):
    user = testuser
    # should already be made by our signal
    notification_settings = NotificationSettings.objects.get(user=user)
    notification_settings.failed_delivery_attempt = True
    notification_settings.available_for_pickup = True
    notification_settings.delivery_issue = True
    notification_settings.save()
    cust_email = "[email protected]"
    customer = Customer.objects.create(email=cust_email)
    order = Order.objects.create(customer=customer)
    shipment = Shipment.objects.create(user=user, order=order)
    pg_connection.poll()
    shipment.status = Shipment.FAILED_ATTEMPT
    shipment.save()
    shipment.status = Shipment.HELD
    shipment.save()
    shipment.status = Shipment.ISSUE
    shipment.save()
    assert 3 == len(pg_connection.notifies)

    # process_notifications() is needed for the listener to run and EmailNotifications to be created
    process_notifications(pg_connection)
    print(f"all emailnotifications: {EmailNotification.objects.all()}")
    existing_notifications = EmailNotification.objects.filter(
        shipment=shipment,
        email=cust_email,
    )
    assert existing_notifications.count() == 3

Оригинальные разделы сообщений ниже (теперь устарели)

Проблема

У меня есть тест, который запускает команду django-admin в потоке с тайм-аутом. Тест проходит, но зависает и приходится вручную останавливать его выполнение. Я не знаком с многопоточностью, но я думаю, что поток продолжает работать, что приводит к «зависанию» теста после его прохождения. Я ищу решение, которое запускает команду, поскольку это скорее интеграционный тест, а не модульный тест с макетами. Любое решение, которое запускает команду в течение заданного времени, должно работать, пока есть надлежащий разрыв теста и это не вызывает проблем для других тестов во время CI.

Фон

Django-pgpubsub прослушивает триггеры PostgresSQL, запуская python manage.py listen. Я использую это как облегченную альтернативу сигналам Django и celery для отправки электронных писем, когда поле состояния экземпляра модели отгрузки изменяется на определенные значения.

Тест

# transaction=True is needed listen command to work for django-pgpubsub
@pytest.mark.django_db(transaction=True)
def test_shipment_status_notifications_with_listen_command(testuser):
    user = testuser
    notification_settings = NotificationSettings.objects.get(user=user)
    notification_settings.failed_attempt = True
    notification_settings.save()
    cust_email = "[email protected]"
    customer = Customer.objects.create(email=cust_email)
    order = Order.objects.create(customer=customer)

    def run_listen_command():
        call_command("listen")

    listen_thread = threading.Thread(target=run_listen_command, daemon=True)
    listen_thread.start()
    # Change Shipment status to trigger the notification
    shipment = Shipment.objects.create(user=user, order=order)
    shipment.status = Shipment.FAILED_ATTEMPT
    shipment.save()
    sleep(1)  # required to allow listen command to process the notification
    # Ensure the function that sends the notification is called and the notification is created
    existing_notifications = Notification.objects.filter(
        type=Notification.Type.FAILED_DELIVERY_ATTEMPT,
        shipment=shipment,
        email=cust_email,
    )
    assert existing_notifications.count() == 1
    # Wait for the listen_thread to complete, with a timeout
    # todo: this doesn't end the test and it hangs
    listen_thread.join(timeout=3)

Другие попытки

Я не могу настроить команду Django, потому что она из внешней библиотеки, django-pgpubsub. Я не уверен, что многопоточность является лучшим решением, поскольку я нашел это предложение с использованием GPT-4 и не смог найти существующие примеры. Другая попытка включала ThreadPoolExecutor с try/except/finally с TimeoutError. Это тоже прошло, но зависло. Другой связан с использованием модуля подпроцесса. Эта версия дает сбой, так как уведомления не отправляются, или она зависает, если subprocess.PIPE передается в параметры stdout и stderr subprocess.Popen() для устранения неполадок. Вот код для этой версии теста:

@pytest.mark.django_db(transaction=True)
def test_shipment_status_notifications_with_listen_command_subprocess(testuser):
    print("starting test")
    user = testuser
    notification_settings = NotificationSettings.objects.get(user=user)
    notification_settings.failed_attempt = True
    notification_settings.save()
    cust_email = "[email protected]"
    customer = Customer.objects.create(email=cust_email)
    order = Order.objects.create(customer=customer)

    print(f"settings.ROOT_DIR: {settings.ROOT_DIR}")
    listen_command = [
        "python",
        str(settings.ROOT_DIR / "manage.py"),
        "listen",
    ]
    print("about to call listen command: ", listen_command)

    # todo: this version doesn't hang the test, but no notifications are sent
    # listen_process = subprocess.Popen(listen_command)
    # todo: using this version with stdout and stderr hangs the test
    listen_process = subprocess.Popen(listen_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0, universal_newlines=True)
    # Change Shipment status to trigger the notification
    shipment = Shipment.objects.create(user=user, order=order)
    shipment.status = Shipment.FAILED_ATTEMPT
    shipment.save()

    time.sleep(5)

    # Stop the listen command
    listen_process.send_signal(signal.SIGTERM)
    listen_process.wait()

    stdout, stderr = listen_process.communicate()

    print("Output from listen command:")
    print(stdout)
    print("Error output from listen command:")
    print(stderr)

    existing_notifications = Notification.objects.filter(
        type=Notification.Type.FAILED_DELIVERY_ATTEMPT,
        shipment=shipment,
        email=cust_email,
    )
    # todo: No existing notifications found when using listen_process = subprocess.Popen(listen_command)
    print(f"existing_notifications: {existing_notifications}")
    assert existing_notifications.count() == 1