84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
import os
|
|
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 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()
|
|
old = timezone.now().timestamp() - 25 * 3600
|
|
os.utime(stale, (old, old))
|
|
|
|
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"
|