Jayway Liu
2a83de21d9
Verify / verify (push) Failing after 2m39s
- 参数与口径页新增维度键与显示名对照行（只读） - 使用说明页新增输入字典：每个可编辑输入逐项列出字段、单位、来源、 修改位置与缺失含义；补充固定结构边界说明 - 毛利率参数行改为文本快照并指向计算明细 MarginRate 列 - 对账容差说明改为按指标类型独立容差、调整需财务批准 - 撤权校验区分账号停用与角色变更的报错信息 - 引入 test_formula_workbook_docs 与扩展的 test_worker_paths（F08 撤权、 崩溃恢复语义对齐 PostgreSQL 咨询锁立即重认领）
142 lines
5.0 KiB
Python
142 lines
5.0 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_reclaimed_immediately(workspace):
|
|
# PostgreSQL 咨询锁语义下 worker 立即重认领 processing 任务，不再等待超时。
|
|
ds, *_ = workspace
|
|
job = _export_job(ds, status="processing")
|
|
|
|
call_command("import_worker", once=True)
|
|
|
|
job.refresh_from_db()
|
|
assert job.status == "succeeded"
|