diff --git a/backend/drive/test_trash_isolation.py b/backend/drive/test_trash_isolation.py deleted file mode 100644 index f6379b8..0000000 --- a/backend/drive/test_trash_isolation.py +++ /dev/null @@ -1,135 +0,0 @@ -"""Run: USE_SQLITE=true python manage.py test drive.test_trash_isolation""" -import tempfile -from pathlib import Path -from datetime import timedelta -from django.contrib.auth import get_user_model -from django.core.management import call_command -from django.test import TestCase, override_settings -from django.utils import timezone -from rest_framework.test import APIClient -from .models import DriveItem -from .spaces import ensure_space -from .storage_layout import disk_path - - -class TrashIsolationTests(TestCase): - def setUp(self): - temp = tempfile.TemporaryDirectory() - self.addCleanup(temp.cleanup) - self.root = Path(temp.name) - config = override_settings(DRIVE_STORAGE_ROOT=self.root, FILE_UPLOAD_TEMP_DIR=str(self.root / '.tmp')) - config.enable() - self.addCleanup(config.disable) - self.user = get_user_model().objects.create_superuser('trash-tester', 'trash@example.org', 'password') - self.space = ensure_space(self.user) - self.client = APIClient() - self.client.force_authenticate(self.user) - - def new_file(self, name='rapport.txt', content=b'original'): - from uuid import uuid4 - location = f'{self.space.relative_path}/{name}' if self.space.relative_path else name - path = disk_path(location) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_bytes(content) - return DriveItem.objects.create(space=self.space, owner=self.user, name=name, size=len(content), - mime_type='text/plain', storage_key=str(uuid4()), storage_location=location) - - def test_reuse_same_name_multiple_times_and_restore(self): - first = self.new_file(content=b'first') - self.assertEqual(self.client.post(f'/api/drive/items/{first.pk}/trash/').status_code, 200) - first.refresh_from_db() - self.assertEqual(disk_path(first.storage_location).read_bytes(), b'first') - second = self.new_file(content=b'second') - self.assertEqual(self.client.post(f'/api/drive/items/{second.pk}/trash/').status_code, 200) - second.refresh_from_db() - self.assertNotEqual(first.storage_location, second.storage_location) - active = self.new_file(content=b'third') - response = self.client.post(f'/api/drive/items/{first.pk}/restore/') - self.assertEqual(response.status_code, 200, response.data) - self.assertEqual(response.data['name'], 'rapport (restauré 2).txt') - first.refresh_from_db() - self.assertEqual(disk_path(first.storage_location).read_bytes(), b'first') - self.assertEqual(disk_path(second.storage_location).read_bytes(), b'second') - self.assertEqual(disk_path(active.storage_location).read_bytes(), b'third') - self.assertEqual(self.client.delete(f'/api/drive/items/{second.pk}/delete/').status_code, 204) - self.assertTrue(disk_path(active.storage_location).exists()) - - def test_trashed_folder_frees_name_and_restores_contents(self): - folder = self.client.post('/api/drive/folders/', {'name': 'Cours', 'space':self.space.pk}, format='json') - self.assertEqual(folder.status_code, 201, folder.data) - original = DriveItem.objects.get(pk=folder.data['id']) - location = '/'.join(part for part in (self.space.relative_path, 'Cours', 'file.txt') if part) - target = disk_path(location) - target.write_bytes(b'present') - from uuid import uuid4 - child = DriveItem.objects.create(space=self.space, owner=self.user, parent=original, - name='file.txt', storage_key=str(uuid4()), storage_location=location) - self.assertEqual(self.client.post(f'/api/drive/items/{original.pk}/trash/').status_code, 200) - child.refresh_from_db() - self.assertTrue(disk_path(child.storage_location).is_file()) - replacement = self.client.post('/api/drive/folders/', {'name': 'Cours', 'space':self.space.pk}, format='json') - self.assertEqual(replacement.status_code, 201, replacement.data) - restored = self.client.post(f'/api/drive/items/{original.pk}/restore/') - self.assertEqual(restored.status_code, 200, restored.data) - self.assertEqual(restored.data['name'], 'Cours (restauré 2)') - child.refresh_from_db() - self.assertEqual(disk_path(child.storage_location).read_bytes(), b'present') - - def test_file_can_be_trashed_again_after_restore(self): - item = self.new_file(name='cycle.txt', content=b'never lose me') - container = disk_path(f'.gooddrive-trash/{item.pk}') - for _ in range(3): - response = self.client.post(f'/api/drive/items/{item.pk}/trash/') - self.assertEqual(response.status_code, 200, response.data) - item.refresh_from_db() - self.assertTrue(container.is_dir()) - self.assertEqual(disk_path(item.storage_location).read_bytes(), b'never lose me') - response = self.client.post(f'/api/drive/items/{item.pk}/restore/') - self.assertEqual(response.status_code, 200, response.data) - item.refresh_from_db() - self.assertIsNone(item.trashed_at) - self.assertFalse(container.exists(), 'La restauration doit nettoyer le dossier ID vide') - self.assertEqual(disk_path(item.storage_location).read_bytes(), b'never lose me') - - def test_stale_empty_id_folder_from_old_restore_is_reused(self): - item = self.new_file(name='existing.txt') - container = disk_path(f'.gooddrive-trash/{item.pk}') - container.mkdir(parents=True) - response = self.client.post(f'/api/drive/items/{item.pk}/trash/') - self.assertEqual(response.status_code, 200, response.data) - item.refresh_from_db() - self.assertEqual(disk_path(item.storage_location).read_bytes(), b'original') - - def test_nonempty_legacy_container_is_never_overwritten(self): - item = self.new_file(name='important.txt', content=b'original') - container = disk_path(f'.gooddrive-trash/{item.pk}') - container.mkdir(parents=True) - (container / 'untracked.txt').write_bytes(b'KEEP') - response = self.client.post(f'/api/drive/items/{item.pk}/trash/') - self.assertEqual(response.status_code, 409) - self.assertEqual((container / 'untracked.txt').read_bytes(), b'KEEP') - item.refresh_from_db() - self.assertIsNone(item.trashed_at) - self.assertEqual(disk_path(item.storage_location).read_bytes(), b'original') - - def test_folder_can_be_trashed_again_after_restore(self): - response = self.client.post('/api/drive/folders/', {'name': 'Recettes', 'space': self.space.pk}, format='json') - self.assertEqual(response.status_code, 201, response.data) - item = DriveItem.objects.get(pk=response.data['id']) - container = disk_path(f'.gooddrive-trash/{item.pk}') - for _ in range(2): - self.assertEqual(self.client.post(f'/api/drive/items/{item.pk}/trash/').status_code, 200) - self.assertTrue(container.is_dir()) - self.assertEqual(self.client.post(f'/api/drive/items/{item.pk}/restore/').status_code, 200) - self.assertFalse(container.exists()) - - def test_migrate_existing_trashed_without_resetting_retention(self): - item = self.new_file() - when = timezone.now() - timedelta(days=4) - DriveItem.objects.filter(pk=item.pk).update(trashed_at=when) - call_command('migrate_trash_storage', '--apply', verbosity=0) - item.refresh_from_db() - self.assertEqual(item.trashed_at, when) - self.assertEqual(disk_path(item.storage_location).read_bytes(), b'original') - self.new_file(content=b'new copy') - call_command('migrate_trash_storage', '--apply', verbosity=0) diff --git a/backend/drive/trash_cleanup.py b/backend/drive/trash_cleanup.py new file mode 100644 index 0000000..a165ad1 --- /dev/null +++ b/backend/drive/trash_cleanup.py @@ -0,0 +1,61 @@ +"""Permanent deletion shared by the trash API and scheduled retention cleanup.""" +from django.db import transaction +from .models import DriveItem, UploadSession +from .services import descendant_files +from .storage_layout import disk_path, file_location + + +def has_active_uploads_in_tree(item): + """Do not purge directories while a resumable upload targets a descendant.""" + if not item.is_folder: + return False + pending = [item.pk] + while pending: + batch, pending = pending[:250], pending[250:] + if UploadSession.objects.filter(parent_id__in=batch).exists(): + return True + pending.extend(DriveItem.objects.filter(parent_id__in=batch, is_folder=True).values_list('pk', flat=True)) + return False + + +def delete_trashed_item(item): + """Delete the SQL tree, then its tracked files after commit. + + Untracked files and directories are not erased. Failures resolving a stored + path abort before the SQL delete; missing physical files are tolerated. + """ + if item.trashed_at is None: + raise ValueError('L’élément doit être dans la corbeille.') + if has_active_uploads_in_tree(item): + raise ValueError('Un transfert est encore en cours dans ce dossier.') + files = [disk_path(key) for key in descendant_files(item)] + # A child may have been trashed separately, into another isolated directory. + folders = [] + queue = [item.pk] if item.is_folder else [] + if item.is_folder: + folders.append(disk_path(item.storage_location or file_location(item.space, item.parent, item.name))) + while queue: + batch, queue = queue[:300], queue[300:] + children = list(DriveItem.objects.filter(parent_id__in=batch, is_folder=True).select_related('parent','space')) + queue.extend(child.pk for child in children) + for child in children: + folders.append(disk_path(child.storage_location or file_location(child.space, child.parent, child.name))) + with transaction.atomic(): + item.delete() + def remove_files(): + for path in files: + path.unlink(missing_ok=True) + for folder in sorted(set(folders), key=lambda p:len(p.parts), reverse=True): + if folder.is_dir() and not folder.is_symlink(): + # Never recursively erase untracked files. + try: folder.rmdir() + except OSError: pass + # Remove empty UUID trash containers (never the shared .trash root). + trash_root = disk_path('.gooddrive-trash') + for path in [*files, *folders]: + parent = path.parent + while parent != trash_root and parent.is_relative_to(trash_root): + try: parent.rmdir() + except OSError: break + parent = parent.parent + transaction.on_commit(remove_files)