背景

Django-pgpubsub通过运行python manage.py listen监听PostgreSQL触发器。此命令持续运行，并作为Django信号和celery的轻量级替代品，用于在Shipment模型示例的状态字段更改为某些值时发送电子邮件。

问题
更新

这适用于--reuse-db（在this pytest-django issue上找到的解决方法）。
集成测试使用subprocess在一个单独的进程中调用python manage.py listen命令。测试通过，但测试数据库仍然存在，我必须手动删除它，以防止下一个测试因重复数据库而失败。无论django-pytest附带的拆除功能被调用，因为在测试完成时输出：
Destroying test database for alias 'default' ('test_my_project_database')

测试

@pytest.mark.django_db(transaction=True) # needed to work
def test_shipment_status_notifications_with_listen_command_subprocess(testuser):
    user = testuser
    # model instances setup for the test
    notification_settings = NotificationSettings.objects.get(user=user)
    notification_settings.failed_attempt = True
    notification_settings.save()
    cust_email = "test@test.com"
    customer = Customer.objects.create(email=cust_email)
    order = Order.objects.create(customer=customer)
    shipment = Shipment.objects.create(user=user, order=order)

    listen_command = [
        "python",
        str(settings.ROOT_DIR / "manage.py"),
        "listen",
    ]
    env = os.environ.copy()
    # we need to update these env vars to use the actual test database instead of the live one
    # the process uses the live database by default
    env.update(
        {
            "DATABASE_URL":
                f"postgresql://{config('POSTGRES_USER')}:{config('POSTGRES_PASSWORD')}@127.0.0.1:5432/test_{config('POSTGRES_DB')}",
        }
    )

    listen_process = subprocess.Popen(
        listen_command,
        env=env,
    )
    time.sleep(2)  # confirmed needed
    # Change Shipment status to trigger the notification
    shipment.status = Shipment.FAILED_ATTEMPT
    shipment.save()

    time.sleep(2)  # confirmed needed
    listen_process.terminate()

    existing_notifications = Notification.objects.filter(
        type=Notification.Type.FAILED_DELIVERY_ATTEMPT,
        shipment=shipment,
        email=cust_email,
    )
    print(f"existing_notifications: {existing_notifications}")
    assert existing_notifications.count() == 1

我所尝试的

使用pytest fixture删除数据库（在yield之后调用SQL命令，以便将其包含在测试的拆卸中）会导致内置的pytest-django拆卸异常“connection already closed/doesn 't exist”。