Jayway Liu
54dc58d2a5
- process_formula_export 失败清理受阻时记录不含敏感数据的 WARNING（任务编号+文件名）， 并保留 job.path 使残留文件仍可追踪，而不是清空后失联 - cleanup_formula_exports 新增孤儿 .xlsx 回收：超过 24h 保留期且不属于任何 未过期成功任务的文件统一回收（覆盖失败任务残留、任务行被级联删除两种来源）， 不触碰仍处于活动任务的文件 - 测试：注入 unlink 权限错误→任务失败状态仍保存+告警可见；恢复权限后回收成功 且其他任务文件不受影响；失败任务残留/任务行已删/未过期成功/近期文件/非 UUID 命名
158 lines
4.4 KiB
Python
158 lines
4.4 KiB
Python
import os
|
|
import uuid
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
from django.contrib.auth.models import User
|
|
from django.core.management import call_command
|
|
from django.utils import timezone
|
|
|
|
from finance.models import Dataset, FormulaExport, Membership, Organization
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture
|
|
def cleanup_root(settings, tmp_path):
|
|
settings.MEDIA_ROOT = tmp_path
|
|
org = Organization.objects.create(name="Cleanup test")
|
|
user = User.objects.create_user("cleanup", password="Cleanup-TestPassword991!")
|
|
Membership.objects.create(user=user, organization=org, role="admin")
|
|
dataset = Dataset.objects.create(
|
|
organization=org,
|
|
created_by=user,
|
|
name="fixture.csv",
|
|
sha256="a" * 64,
|
|
status="ready",
|
|
)
|
|
return tmp_path / "formula-exports", dataset, user
|
|
|
|
|
|
def _aged(path, hours=25):
|
|
stamp = (timezone.now() - timedelta(hours=hours)).timestamp()
|
|
os.utime(path, (stamp, stamp))
|
|
return path
|
|
|
|
|
|
def test_cleanup_expires_jobs_and_removes_files(cleanup_root):
|
|
root, dataset, user = cleanup_root
|
|
path = root / str(user.pk) / "expired.xlsx"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_bytes(b"workbook")
|
|
job = FormulaExport.objects.create(
|
|
dataset=dataset,
|
|
user=user,
|
|
status="succeeded",
|
|
path=str(path),
|
|
expires_at=timezone.now() - timedelta(seconds=1),
|
|
)
|
|
|
|
call_command("cleanup_formula_exports")
|
|
|
|
job.refresh_from_db()
|
|
assert job.status == "expired"
|
|
assert job.path == ""
|
|
assert not path.exists()
|
|
|
|
|
|
def test_cleanup_removes_only_stale_temporary_files(cleanup_root):
|
|
root, _, user = cleanup_root
|
|
user_dir = root / str(user.pk)
|
|
user_dir.mkdir(parents=True)
|
|
stale, recent = user_dir / "stale.tmp", user_dir / "recent.tmp"
|
|
stale.touch()
|
|
recent.touch()
|
|
_aged(stale)
|
|
|
|
call_command("cleanup_formula_exports")
|
|
|
|
assert not stale.exists()
|
|
assert recent.exists()
|
|
|
|
|
|
def test_cleanup_does_not_delete_files_outside_export_root(cleanup_root, tmp_path):
|
|
root, dataset, user = cleanup_root
|
|
outside = tmp_path / "protected.xlsx"
|
|
outside.write_bytes(b"keep")
|
|
job = FormulaExport.objects.create(
|
|
dataset=dataset,
|
|
user=user,
|
|
status="succeeded",
|
|
path=str(outside),
|
|
expires_at=timezone.now() - timedelta(seconds=1),
|
|
)
|
|
|
|
call_command("cleanup_formula_exports")
|
|
|
|
job.refresh_from_db()
|
|
assert job.status == "expired"
|
|
assert outside.read_bytes() == b"keep"
|
|
|
|
|
|
def test_cleanup_reclaims_failed_job_leftover_after_retention(cleanup_root):
|
|
root, dataset, user = cleanup_root
|
|
job = FormulaExport.objects.create(dataset=dataset, user=user, status="failed")
|
|
path = root / str(user.pk) / f"{job.pk}.xlsx"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_bytes(b"leftover")
|
|
job.path = str(path)
|
|
job.save(update_fields=["path"])
|
|
_aged(path)
|
|
|
|
call_command("cleanup_formula_exports")
|
|
|
|
job.refresh_from_db()
|
|
assert not path.exists()
|
|
assert job.path == ""
|
|
|
|
|
|
def test_cleanup_keeps_workbook_of_unexpired_succeeded_job(cleanup_root):
|
|
root, dataset, user = cleanup_root
|
|
job = FormulaExport.objects.create(
|
|
dataset=dataset,
|
|
user=user,
|
|
status="succeeded",
|
|
expires_at=timezone.now() + timedelta(hours=1),
|
|
)
|
|
path = root / str(user.pk) / f"{job.pk}.xlsx"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_bytes(b"active")
|
|
job.path = str(path)
|
|
job.save(update_fields=["path"])
|
|
_aged(path)
|
|
|
|
call_command("cleanup_formula_exports")
|
|
|
|
assert path.read_bytes() == b"active"
|
|
job.refresh_from_db()
|
|
assert job.path == str(path)
|
|
|
|
|
|
def test_cleanup_reclaims_orphan_whose_job_row_was_deleted(cleanup_root):
|
|
root, _, user = cleanup_root
|
|
user_dir = root / str(user.pk)
|
|
user_dir.mkdir(parents=True)
|
|
orphan = user_dir / f"{uuid.uuid4()}.xlsx"
|
|
orphan.write_bytes(b"orphan")
|
|
recent = user_dir / f"{uuid.uuid4()}.xlsx"
|
|
recent.write_bytes(b"recent orphan")
|
|
_aged(orphan)
|
|
|
|
call_command("cleanup_formula_exports")
|
|
|
|
assert not orphan.exists()
|
|
assert recent.exists()
|
|
|
|
|
|
def test_cleanup_ignores_non_uuid_workbook_names(cleanup_root):
|
|
root, _, user = cleanup_root
|
|
user_dir = root / str(user.pk)
|
|
user_dir.mkdir(parents=True)
|
|
odd = user_dir / "notes.xlsx"
|
|
odd.write_bytes(b"keep")
|
|
_aged(odd)
|
|
|
|
call_command("cleanup_formula_exports")
|
|
|
|
assert odd.exists()
|