背景
Django-pgpubsub 通过运行来监听 PostgresSQL 触发器
python manage.py listen
。此命令持续运行并作为 Django 信号和芹菜的轻量级替代方案,用于在 Shipment 模型实例的状态字段更改为特定值时发送电子邮件。
问题
集成测试使用
python manage.py listen
在单独的进程中调用subprocess
命令。测试通过但测试数据库仍然存在,之后我必须手动删除它以防止下一次测试由于重复数据库而失败。无论 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 = "[email protected]"
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
拆卸异常关于“连接已经关闭/不存在”。