Files
fa-agent/tests/test_worker_paths.py
T
Jayway Liu dfaecde8a3
Verify / verify (push) Failing after 1s
Verify / verify (pull_request) Failing after 0s
test(worker): 适配 PostgreSQL 咨询锁下的立即重认领语义
PG 上 worker 持咨询锁会立即重认领 processing 任务,第二次 --once 被导入任务
抢占导致导出断言失败(该测试此前从未在 PG 上真正运行过)。第一次调用后把导入
任务置为 ready 模拟完成,两种后端行为一致
2026-09-26 16:20:40 +08:00

151 lines
4.9 KiB
Python

from datetime import timedelta
from pathlib import Path
from unittest.mock import patch
import pytest
from django.core.management import call_command
from django.utils import timezone
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"
def _export_job(ds, **kwargs):
return FormulaExport.objects.create(
dataset=ds,
user=ds.created_by,
context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
**kwargs,
)
def test_worker_refuses_export_after_role_revocation(workspace):
from finance.exporting import process_formula_export
ds, member, *_ = workspace
job = _export_job(ds, status="processing")
member.role = "viewer"
member.save(update_fields=["role"])
with pytest.raises(PermissionError, match="角色"):
process_formula_export(job.pk)
job.refresh_from_db()
assert job.status == "failed"
assert not job.path
assert not list(Path(ds.file.storage.location).rglob(f"{job.pk}*"))
def test_worker_refuses_export_after_account_deactivation(workspace):
from finance.exporting import process_formula_export
ds, _, _, _, _ = workspace
user = ds.created_by
user.is_active = False
user.save(update_fields=["is_active"])
job = _export_job(ds, status="processing")
with pytest.raises(PermissionError, match="停用"):
process_formula_export(job.pk)
job.refresh_from_db()
assert job.status == "failed"
def test_status_and_download_deny_revoked_viewer(workspace, settings):
ds, member, client, *_ = workspace
job = _export_job(
ds,
status="succeeded",
finished_at=timezone.now(),
expires_at=timezone.now() + timedelta(hours=1),
)
target = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id) / f"{job.pk}.xlsx"
target.parent.mkdir(parents=True)
target.write_bytes(b"PK workbook")
job.path = str(target)
job.save(update_fields=["path"])
assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}").status_code == 200
assert (
client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == 200
)
member.role = "viewer"
member.save(update_fields=["role"])
assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}").status_code == 403
assert (
client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == 403
)
def test_half_finished_export_cannot_be_downloaded(workspace, settings):
ds, _, client, *_ = workspace
for status, code in (("processing", 409), ("failed", 409), ("queued", 409)):
job = _export_job(ds, status=status)
assert (
client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code
== code
), status
succeeded = _export_job(
ds,
status="succeeded",
path=str(Path(settings.MEDIA_ROOT) / "nowhere.xlsx"),
finished_at=timezone.now(),
expires_at=timezone.now() + timedelta(hours=1),
)
assert (
client.get(f"/api/datasets/{ds.pk}/formula-exports/{succeeded.pk}/download").status_code
== 410
)
def test_stale_processing_export_is_reclaimed_after_timeout(workspace):
ds, *_ = workspace
job = _export_job(ds, status="processing")
FormulaExport.objects.filter(pk=job.pk).update(
created_at=timezone.now() - timedelta(minutes=20)
)
call_command("import_worker", once=True)
job.refresh_from_db()
assert job.status == "succeeded"
assert Path(job.path).read_bytes()[:2] == b"PK"
assert not list(Path(job.path).parent.glob(f"{job.pk}.*.tmp"))
def test_recent_processing_export_is_not_reclaimed(workspace):
ds, *_ = workspace
job = _export_job(ds, status="processing")
call_command("import_worker", once=True)
job.refresh_from_db()
assert job.status == "processing"