test
This commit is contained in:
Louis Mazin
parent
1ca1219424
commit
1b2b2c414d
@ -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)
|
|
||||||
61
backend/drive/trash_cleanup.py
Normal file
61
backend/drive/trash_cleanup.py
Normal file
@ -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)
|
||||||
Loading…
x
Reference in New Issue
Block a user