diff --git a/backend/drive/models.py b/backend/drive/models.py
index 511f0f5..8589f15 100644
--- a/backend/drive/models.py
+++ b/backend/drive/models.py
@@ -82,7 +82,7 @@ class DriveItem(models.Model):
indexes = [models.Index(fields=['owner','parent']), models.Index(fields=['space','trashed_at'])]
def clean(self):
- if not self.name or self.name in {'.', '..'} or any(c in self.name for c in '/\\\x00\r\n'):
+ if not self.name or self.name in {'.', '..', '.gooddrive-trash'} or any(c in self.name for c in '/\\\x00\r\n'):
raise ValidationError('Nom de fichier ou dossier invalide.')
if len(self.name) > 255:
raise ValidationError('Nom trop long.')
diff --git a/backend/drive/services.py b/backend/drive/services.py
index afc1dd6..9bcfe61 100644
--- a/backend/drive/services.py
+++ b/backend/drive/services.py
@@ -60,7 +60,7 @@ def valid_name(name):
if not isinstance(name,str) or not name.strip() or len(name)>255:
raise ValueError('Nom invalide (1 à 255 caractères).')
name=name.strip()
- if name in ('.','..') or any(c in name for c in ('/','\\','\x00','\r','\n')):
+ if name in ('.','..', '.gooddrive-trash') or any(c in name for c in ('/','\\','\x00','\r','\n')):
raise ValueError('Ce nom contient des caractères interdits.')
return name
diff --git a/backend/drive/storage_layout.py b/backend/drive/storage_layout.py
index 7fe1a07..fd02b98 100644
--- a/backend/drive/storage_layout.py
+++ b/backend/drive/storage_layout.py
@@ -48,7 +48,8 @@ def disk_path(location):
def assert_available(space, parent, name, exclude=None):
- siblings = DriveItem.objects.filter(space=space, parent=parent, name=name)
+ # Trashed entries have their own isolated physical location. Names can be reused.
+ siblings = DriveItem.objects.filter(space=space, parent=parent, name=name, trashed_at__isnull=True)
if exclude is not None:
siblings = siblings.exclude(pk=exclude)
if siblings.exists():
@@ -72,8 +73,8 @@ def relocate_item(item, new_parent, new_name):
The on-disk move is reversed if the database transaction fails. The caller
must ensure authorization and that new_parent is not a descendant of item.
"""
- old = (file_location(item.space, item.parent, item.name) if item.is_folder
- else item.storage_location or f'{item.storage_key[:2]}/{item.storage_key}')
+ old = item.storage_location or (file_location(item.space, item.parent, item.name)
+ if item.is_folder else f'{item.storage_key[:2]}/{item.storage_key}')
new = file_location(item.space, new_parent, new_name)
source = disk_path(old)
target = assert_available(item.space, new_parent, new_name, exclude=item.pk)
@@ -92,16 +93,8 @@ def relocate_item(item, new_parent, new_name):
source.rename(target)
moved = True
if item.is_folder:
- prefix = old.rstrip('/') + '/'
- affected = list(DriveItem.objects.filter(
- space=item.space, storage_location__startswith=prefix
- ).only('pk', 'storage_location'))
- for child in affected:
- child.storage_location = new + '/' + child.storage_location[len(prefix):]
- if affected:
- DriveItem.objects.bulk_update(affected, ['storage_location'], batch_size=500)
- else:
- item.storage_location = new
+ _update_descendant_locations(item, old, new)
+ item.storage_location = new
item.name = new_name
item.parent = new_parent
item.full_clean()
@@ -112,3 +105,77 @@ def relocate_item(item, new_parent, new_name):
if target.exists() and not source.exists():
target.rename(source)
raise
+
+
+def _update_descendant_locations(item, old, new):
+ """Keep physical paths in sync, including legacy folders without a stored path.
+
+ Entries already moved independently to trash keep their own location.
+ """
+ old_prefix = old.rstrip('/') + '/'
+ new_prefix = new.rstrip('/') + '/'
+ pending = [item.pk]
+ while pending:
+ batch, pending = pending[:300], pending[300:]
+ children = list(DriveItem.objects.filter(parent_id__in=batch).select_related('parent', 'space'))
+ pending.extend(child.pk for child in children if child.is_folder)
+ changed = []
+ for child in children:
+ location = child.storage_location
+ if not location and child.is_folder:
+ # Historical folders used only their logical name as a disk path.
+ location = file_location(child.space, child.parent, child.name)
+ if location and location.startswith(old_prefix):
+ child.storage_location = new_prefix + location[len(old_prefix):]
+ changed.append(child)
+ if changed:
+ DriveItem.objects.bulk_update(changed, ['storage_location'], batch_size=300)
+
+
+def trash_location(item):
+ """Each deleted object gets a separate directory, regardless of display name."""
+ return f'.gooddrive-trash/{item.pk}/{item.name}'
+
+
+def move_to_trash(item, *, migrate_existing=False):
+ """Atomic disk/SQL move. Preserve the original name/parent for restoration."""
+ from django.utils import timezone
+ from .trash_cleanup import has_active_uploads_in_tree
+ if (item.trashed_at is not None) != migrate_existing:
+ raise ValueError('État de corbeille inattendu.')
+ if migrate_existing and item.storage_location == trash_location(item):
+ return item
+ if has_active_uploads_in_tree(item):
+ raise ValueError('Un transfert est encore en cours dans ce dossier.')
+ old = item.storage_location or (file_location(item.space, item.parent, item.name)
+ if item.is_folder else f'{item.storage_key[:2]}/{item.storage_key}')
+ new = trash_location(item)
+ source, target = disk_path(old), disk_path(new)
+ if not source.exists() or (item.is_folder != source.is_dir()):
+ raise FileNotFoundError('Élément absent du stockage, déplacement impossible.')
+ if target.exists() or target.is_symlink():
+ raise NameConflict('La destination de la corbeille est déjà utilisée.')
+ target.parent.mkdir(parents=True, exist_ok=False)
+ moved = False
+ try:
+ with transaction.atomic():
+ source.rename(target)
+ moved = True
+ if item.is_folder:
+ _update_descendant_locations(item, old, new)
+ item.storage_location = new
+ if migrate_existing:
+ item.save(update_fields=['storage_location'])
+ else:
+ item.trashed_at = timezone.now()
+ item.save(update_fields=['storage_location', 'trashed_at', 'updated_at'])
+ except Exception:
+ if moved and target.exists() and not source.exists():
+ target.rename(source)
+ raise
+ finally:
+ # Leave the isolation folder only while it contains the deleted element.
+ if not target.exists():
+ try: target.parent.rmdir()
+ except OSError: pass
+ return item
diff --git a/backend/drive/test_trash_isolation.py b/backend/drive/test_trash_isolation.py
new file mode 100644
index 0000000..e070265
--- /dev/null
+++ b/backend/drive/test_trash_isolation.py
@@ -0,0 +1,87 @@
+"""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_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/test_word_preview.py b/backend/drive/test_word_preview.py
new file mode 100644
index 0000000..b3086c2
--- /dev/null
+++ b/backend/drive/test_word_preview.py
@@ -0,0 +1,22 @@
+import tempfile
+from pathlib import Path
+from zipfile import ZipFile
+from django.test import SimpleTestCase
+from .word_preview import render_word_preview
+
+
+class WordPreviewTests(SimpleTestCase):
+ def test_docx_text_escaped_and_tables(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp) / 'note.docx'
+ document = b'''
+ <script>alert(1)</script>
+ Tableau
+ '''
+ with ZipFile(path, 'w') as archive:
+ archive.writestr('word/document.xml', document)
+ html = render_word_preview(path)
+ self.assertIn('<script>', html)
+ self.assertNotIn('