Jayway Liu
f9f9ad751d
Verify / verify (push) Failing after 4m39s
PG 上 worker 持咨询锁会立即重认领 processing 任务，第二次 --once 被导入任务 抢占导致导出断言失败（该测试此前从未在 PG 上真正运行过）。第一次调用后把导入 任务置为 ready 模拟完成，两种后端行为一致
33 lines
1.2 KiB
Python
33 lines
1.2 KiB
Python
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
|
|
from finance.models import Dataset, FormulaExport
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_worker_processes_queued_import_and_reports_export_failure(workspace):
|
|
ds, *_ = workspace
|
|
ds.status = "queued"
|
|
ds.save(update_fields=["status"])
|
|
with patch(
|
|
"finance.management.commands.import_worker.process_dataset", return_value=ds
|
|
) as process_import:
|
|
call_command("import_worker", once=True)
|
|
process_import.assert_called_once_with(ds.pk)
|
|
# PostgreSQL 上 worker 持咨询锁会立即重认领 processing 任务：模拟导入完成，
|
|
# 否则第二次 --once 又被它抢占，轮不到下面的导出任务（SQLite 无咨询锁，
|
|
# 不走立即恢复路径，两种后端行为一致）。
|
|
Dataset.objects.filter(pk=ds.pk).update(status="ready")
|
|
|
|
export = FormulaExport.objects.create(dataset=ds, user=ds.created_by, context={})
|
|
with patch(
|
|
"finance.management.commands.import_worker.process_formula_export",
|
|
side_effect=ValueError("invalid export"),
|
|
):
|
|
call_command("import_worker", once=True)
|
|
export.refresh_from_db()
|
|
assert export.status == "processing"
|