tests/test_worker_paths.py

@@ -6,7 +6,7 @@ import pytest
 from django.core.management import call_command
 from django.utils import timezone
 
-from finance.models import FormulaExport
+from finance.models import Dataset, FormulaExport
 
 pytestmark = pytest.mark.django_db
 
@@ -20,6 +20,10 @@ def test_worker_processes_queued_import_and_reports_export_failure(workspace):
     ) 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(