(backend) add command to clean pending and deleted files

Since we have added custom backgrounds, they could be
soft deleted but were never really deleted.
This commits introduces a command to actually delete
this files.
In introduces also a command to purge pending upload
.
This commit is contained in:
Florent Chehab
2026-06-03 17:38:09 +02:00
parent 898bc9a0f8
commit 5c15140783
8 changed files with 274 additions and 2 deletions
@@ -0,0 +1,90 @@
"""Tests for the clean_pending_files management command."""
from datetime import timedelta
from django.core.files.storage import default_storage
from django.core.management import call_command
from django.utils import timezone
import pytest
from core import factories, models
pytestmark = pytest.mark.django_db
def test_clean_pending_files_no_stale_files():
"""Nothing happens when there are no stale pending files."""
call_command("clean_pending_files")
def test_clean_pending_files_recent_pending_not_deleted():
"""Recent pending files (within threshold) should not be deleted."""
file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
update_upload_state=models.FileUploadStateChoices.PENDING,
upload_bytes=b"hello",
)
call_command("clean_pending_files")
file.refresh_from_db()
assert file.deleted_at is None
assert default_storage.exists(file.file_key)
def test_clean_pending_files_old_pending_deleted():
"""Pending files older than the threshold should be deleted."""
old_date = timezone.now() - timedelta(hours=49)
file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
update_upload_state=models.FileUploadStateChoices.PENDING,
upload_bytes=b"hello",
)
assert default_storage.exists(file.file_key)
models.File.objects.filter(pk=file.pk).update(created_at=old_date)
call_command("clean_pending_files")
assert not models.File.objects.filter(pk=file.pk).exists()
assert not default_storage.exists(file.file_key)
def test_clean_pending_files_old_non_pending_not_deleted():
"""Old files that are not pending should not be deleted."""
old_date = timezone.now() - timedelta(hours=49)
file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
update_upload_state=models.FileUploadStateChoices.READY,
)
models.File.objects.filter(pk=file.pk).update(created_at=old_date)
call_command("clean_pending_files")
file.refresh_from_db()
assert file.deleted_at is None
assert file.hard_deleted_at is None
def test_clean_pending_files_custom_hours():
"""The --hours argument controls the age threshold."""
old_date = timezone.now() - timedelta(hours=10)
file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
update_upload_state=models.FileUploadStateChoices.PENDING,
upload_bytes=b"hello",
)
models.File.objects.filter(pk=file.pk).update(created_at=old_date)
# Default 24h threshold -> file not deleted
call_command("clean_pending_files")
file.refresh_from_db()
assert file.deleted_at is None
assert default_storage.exists(file.file_key)
# 8h threshold -> file deleted
call_command("clean_pending_files", "--hours=8")
assert not models.File.objects.filter(pk=file.pk).exists()
assert not default_storage.exists(file.file_key)
@@ -0,0 +1,85 @@
"""Tests for the purge_deleted_files management command."""
from datetime import timedelta
from io import StringIO
from random import randint
from unittest.mock import patch
from django.core.files.storage import default_storage
from django.core.management import call_command
from django.utils import timezone
import pytest
from core import factories, models
from core.tasks.file import process_file_deletion
pytestmark = pytest.mark.django_db
def test_purge_deleted_files_no_deleted_files(django_assert_num_queries):
"""Nothing happens when there are no purgeable files."""
with django_assert_num_queries(1):
call_command("purge_deleted_files")
@pytest.mark.django_db(transaction=True)
def test_purge_deleted_files_success(settings):
"""
Queue deletion for:
- hard-deleted files
- soft-deleted files past retention period + grace period.
"""
out = StringIO()
settings.FILE_PURGE_GRACE_DAYS = grace = randint(1, 20)
now = timezone.now()
purge_now = now - timedelta(days=grace)
not_deleted_file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
upload_bytes=b"hello",
)
with patch("django.utils.timezone.now", return_value=now):
not_purgeable_file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
upload_bytes=b"hello",
)
not_purgeable_file.soft_delete()
with patch("django.utils.timezone.now", return_value=purge_now):
purgeable_file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
upload_bytes=b"hello",
)
purgeable_file.soft_delete()
hard_deleted_file = factories.FileFactory(
type=models.FileTypeChoices.BACKGROUND_IMAGE,
upload_bytes=b"hello",
)
hard_deleted_file.soft_delete()
hard_deleted_file.hard_delete()
with patch(
"core.management.commands.purge_deleted_files.process_file_deletion.delay",
side_effect=process_file_deletion,
) as mock_delay:
call_command("purge_deleted_files", stdout=out)
assert "Purged 2 deleted file(s)." in out.getvalue()
assert mock_delay.call_count == 2
called_ids = {call.args[0] for call in mock_delay.call_args_list}
assert called_ids == {purgeable_file.id, hard_deleted_file.id}
assert models.File.objects.filter(id=not_deleted_file.id).exists()
assert models.File.objects.filter(id=not_purgeable_file.id).exists()
assert not models.File.objects.filter(id=purgeable_file.id).exists()
assert not models.File.objects.filter(id=hard_deleted_file.id).exists()
assert default_storage.exists(not_deleted_file.file_key)
assert default_storage.exists(not_purgeable_file.file_key)
assert not default_storage.exists(purgeable_file.file_key)
assert not default_storage.exists(hard_deleted_file.file_key)