diff --git a/backend/drive/management/commands/cleanup_uploads.py b/backend/drive/management/commands/cleanup_uploads.py
deleted file mode 100644
index bc9eac2..0000000
--- a/backend/drive/management/commands/cleanup_uploads.py
+++ /dev/null
@@ -1,24 +0,0 @@
-from datetime import timedelta
-from pathlib import Path
-from django.conf import settings
-from django.core.management.base import BaseCommand
-from django.utils import timezone
-from drive.models import UploadSession
-
-class Command(BaseCommand):
- help='Purge les sessions de transferts interrompus depuis plus de N jours.'
-
- def add_arguments(self,parser):
- parser.add_argument('--days',type=int,default=14)
-
- def handle(self,*args,**opts):
- if opts['days']<1:
- raise ValueError('Le nombre de jours doit être positif.')
- rows=UploadSession.objects.filter(updated_at__lt=timezone.now()-timedelta(days=opts['days']))
- count=0
- for job in rows.iterator():
- path=Path(settings.FILE_UPLOAD_TEMP_DIR)/f'drive-{job.pk}.part'
- job.delete()
- path.unlink(missing_ok=True)
- count+=1
- self.stdout.write(self.style.SUCCESS(f'{count} import(s) ancien(s) supprimé(s).'))
diff --git a/backend/drive/management/commands/flatten_drive_root.py b/backend/drive/management/commands/flatten_drive_root.py
deleted file mode 100644
index 2235b5b..0000000
--- a/backend/drive/management/commands/flatten_drive_root.py
+++ /dev/null
@@ -1,172 +0,0 @@
-"""Move the legacy physical /storage/root layer into /storage.
-
-This is deliberately an operator-triggered maintenance operation: the application
-must be stopped and both the database and the storage must be backed up first.
-A journal permits safely retrying if the process is interrupted between filesystem
-renames and the atomic database update.
-"""
-import json
-import os
-from pathlib import Path
-
-from django.conf import settings
-from django.core.management.base import BaseCommand, CommandError
-from django.db import transaction
-from django.db.models import F
-from django.db.models.functions import Substr
-
-from drive.models import DriveItem, DriveSpace
-
-JOURNAL_NAME = '.gooddrive-flatten-root.json'
-PREFIX = 'root/'
-
-
-def present(path):
- """Like exists(), but also detect dangling symlinks so they cannot be overwritten."""
- return path.exists() or path.is_symlink()
-
-
-class Command(BaseCommand):
- help = ('Supprime le dossier physique /root hérité des anciennes versions. '
- 'Sans --apply, effectue uniquement les vérifications.')
-
- def add_arguments(self, parser):
- parser.add_argument('--apply', action='store_true',
- help='Effectue les déplacements et met à jour les chemins SQL.')
-
- def handle(self, *args, **options):
- storage = Path(settings.DRIVE_STORAGE_ROOT).resolve()
- old_root = storage / 'root'
- journal = storage / JOURNAL_NAME
- apply = options['apply']
-
- root = DriveSpace.objects.filter(parent__isnull=True, private_owner__isnull=True).order_by('pk').first()
- if root is None:
- self.stdout.write('Aucun espace racine initialisé : rien à modifier.')
- return
- if root.relative_path not in ('root', ''):
- raise CommandError(f'Chemin racine inattendu : {root.relative_path!r}.')
- if old_root.is_symlink() or (present(old_root) and not old_root.is_dir()):
- raise CommandError('Le chemin root est un lien symbolique ou n’est pas un dossier.')
-
- if present(journal):
- try:
- state = json.loads(journal.read_text(encoding='utf-8'))
- except (OSError, ValueError) as exc:
- raise CommandError('Journal de migration illisible : ne pas le supprimer sans vérification.') from exc
- names = state.get('entries')
- if (state.get('version') != 1 or state.get('root_id') != root.pk
- or not isinstance(names, list) or len(names) != len(set(names))
- or any(not isinstance(name, str) or not name or name in ('.', '..')
- or '/' in name or '\\' in name or '\x00' in name for name in names)):
- raise CommandError('Journal de migration invalide : intervention manuelle nécessaire.')
- resumed = True
- else:
- names = sorted(path.name for path in old_root.iterdir()) if old_root.exists() else []
- resumed = False
-
- if root.relative_path == '':
- # Previous attempt may have committed SQL but been interrupted before
- # removing the journal. Do not move untracked content in this state.
- if journal.exists():
- for name in names:
- if present(old_root / name) or not present(storage / name):
- raise CommandError('La base est migrée, mais les déplacements sont incomplets : contrôle manuel nécessaire.')
- if old_root.exists() and any(old_root.iterdir()):
- raise CommandError('Le dossier root contient encore des éléments : intervention manuelle nécessaire.')
- if apply:
- if old_root.exists():
- old_root.rmdir()
- journal.unlink()
- self.stdout.write(self.style.SUCCESS('Migration déjà appliquée ; journal nettoyé.'))
- else:
- self.stdout.write('Base déjà migrée. --apply nettoiera le journal restant.')
- elif old_root.exists():
- if any(old_root.iterdir()):
- raise CommandError('La base utilise déjà /storage, mais /storage/root contient des données non suivies. Aucun déplacement automatique.')
- if apply:
- old_root.rmdir()
- self.stdout.write(self.style.SUCCESS('Dossier root vide supprimé.'))
- else:
- self.stdout.write('Base déjà migrée. --apply supprimera le dossier root vide.')
- else:
- self.stdout.write(self.style.SUCCESS('Déjà configuré : la racine est directement /storage.'))
- return
-
- # Only normalize the legacy root and descendants. Other spaces and
- # existing physical paths are never overwritten or merged implicitly.
- spaces = list(DriveSpace.objects.all().order_by('pk'))
- changing = [s for s in spaces if s.pk == root.pk or s.relative_path.startswith(PREFIX)]
- other_paths = {s.relative_path.casefold() for s in spaces if s not in changing}
- new_paths = ['' if s.pk == root.pk else s.relative_path[len(PREFIX):] for s in changing]
- if (any(p.startswith('/') or not p or p.startswith('../') or '/..' in p
- for p in new_paths if p != '')
- or len({p.casefold() for p in new_paths}) != len(new_paths)
- or any(p.casefold() in other_paths for p in new_paths)):
- raise CommandError('Collision ou chemin invalide parmi les espaces SQL : aucune modification.')
-
- # A referenced physical file cannot be lost due to a typo or an
- # already-missing root. Empty storage_location is legacy and is left as-is.
- if not resumed:
- for location in DriveItem.objects.filter(is_folder=False, storage_location__startswith=PREFIX).values_list('storage_location', flat=True).iterator():
- if not location.startswith(PREFIX) or '\\' in location or '..' in Path(location).parts:
- raise CommandError(f'Chemin de fichier invalide en base : {location!r}.')
- if not (storage / location).is_file():
- raise CommandError(f'Fichier référencé manquant : {location!r}. Aucun déplacement.')
-
- for name in names:
- source, target = old_root / name, storage / name
- if present(source) and source.is_symlink():
- raise CommandError(f'Lien symbolique dans root : {name!r}. Migration refusée.')
- if present(source) and present(target):
- raise CommandError(f'Collision : {source} et {target} existent déjà. Aucun écrasement.')
- if not present(source) and (not resumed or not present(target)):
- raise CommandError(f'Élément introuvable : {source}. Migration interrompue.')
-
- count_files = DriveItem.objects.filter(storage_location__startswith=PREFIX).count()
- self.stdout.write(f'Racine actuelle : {storage}/root -> {storage}')
- self.stdout.write(f'{len(names)} élément(s) physiques ; {len(changing)} espace(s) ; {count_files} chemin(s) de fichier à modifier.')
- if not apply:
- self.stdout.write(self.style.WARNING('Simulation uniquement. Arrête GoodDrive, sauvegarde SQL et /storage, puis lance --apply.'))
- return
-
- if not resumed:
- # Journal persisted before the first rename, for retry after a crash.
- state = {'version': 1, 'root_id': root.pk, 'entries': names}
- try:
- with journal.open('x', encoding='utf-8') as stream:
- json.dump(state, stream, ensure_ascii=False)
- stream.flush()
- os.fsync(stream.fileno())
- except FileExistsError as exc:
- raise CommandError('Une autre migration est déjà en cours.') from exc
-
- try:
- for name in names:
- source, target = old_root / name, storage / name
- if present(source):
- if present(target):
- raise CommandError(f'Destination occupée : {target}. Migration interrompue.')
- source.rename(target) # same filesystem: does not copy large files
- if old_root.exists():
- old_root.rmdir() # refuses unexpected new files; do not erase content
- # Verify the new physical destinations before committing the SQL paths.
- for location in DriveItem.objects.filter(
- is_folder=False, storage_location__startswith=PREFIX
- ).values_list('storage_location', flat=True).iterator():
- if not (storage / location[len(PREFIX):]).is_file():
- raise CommandError(f'Fichier inaccessible après déplacement : {location!r}.')
- with transaction.atomic():
- # Leave descendants coherent with the root; update paths with
- # explicit IDs, not by renaming actual DriveItems (logical folders).
- for space, new_path in zip(changing, new_paths):
- DriveSpace.objects.filter(pk=space.pk).update(relative_path=new_path)
- DriveItem.objects.filter(storage_location__startswith=PREFIX).update(
- storage_location=Substr(F('storage_location'), len(PREFIX) + 1))
- journal.unlink()
- except Exception as exc:
- raise CommandError(
- 'Conversion interrompue : journal conservé. Laisse GoodDrive arrêté '
- f'et relance la même commande --apply après résolution. Détail : {exc}'
- ) from exc
- self.stdout.write(self.style.SUCCESS('Terminé : /storage est la racine physique, sans dossier root.'))
diff --git a/backend/drive/test_empty_trash.py b/backend/drive/test_empty_trash.py
deleted file mode 100644
index 5c75883..0000000
--- a/backend/drive/test_empty_trash.py
+++ /dev/null
@@ -1,83 +0,0 @@
-"""USE_SQLITE=true python manage.py test drive.test_empty_trash"""
-import tempfile
-from pathlib import Path
-from uuid import uuid4
-
-from django.contrib.auth import get_user_model
-from django.test import TestCase, override_settings
-from django.utils import timezone
-from rest_framework.test import APIClient
-
-from .models import DriveItem, SpaceGrant
-from .spaces import ensure_private_space, ensure_space
-from .storage_layout import disk_path
-
-
-class EmptyTrashTests(TestCase):
- def setUp(self):
- temp = tempfile.TemporaryDirectory()
- self.addCleanup(temp.cleanup)
- root = Path(temp.name)
- config = override_settings(DRIVE_STORAGE_ROOT=root, FILE_UPLOAD_TEMP_DIR=str(root / '.tmp'))
- config.enable()
- self.addCleanup(config.disable)
- self.admin = get_user_model().objects.create_superuser('admin-trash', 'admin@example.org', 'pass')
- self.other = get_user_model().objects.create_user('private-owner', password='pass')
- self.shared = ensure_space(self.admin)
- SpaceGrant.objects.create(user=self.other, space=self.shared, permissions=['view', 'download'])
- self.private = ensure_private_space(self.other)
- self.client = APIClient()
- self.client.force_authenticate(self.admin)
-
- def file(self, space, owner, filename, relative_location):
- location = '/'.join(p for p in (space.relative_path, relative_location) if p)
- path = disk_path(location)
- path.parent.mkdir(parents=True, exist_ok=True)
- path.write_bytes(filename.encode())
- record = DriveItem.objects.create(
- space=space, owner=owner, name=filename, mime_type='text/plain',
- storage_key=str(uuid4()), storage_location=location,
- size=len(filename), trashed_at=timezone.now(),
- )
- return record, path
-
- def test_empty_trash_removes_accessible_files_but_not_other_private_files(self):
- item1, path1 = self.file(self.shared, self.admin, 'duplicata.txt', '.gooddrive-trash/id1/duplicata.txt')
- item2, path2 = self.file(self.shared, self.admin, 'duplicata.txt', '.gooddrive-trash/id2/duplicata.txt')
- secret, secret_path = self.file(self.private, self.other, 'prive.txt', '.gooddrive-trash/private/prive.txt')
- with self.captureOnCommitCallbacks(execute=True):
- response = self.client.delete('/api/drive/trash/')
- self.assertEqual(response.status_code, 200, response.data)
- self.assertEqual(response.data['deleted'], 2)
- self.assertFalse(DriveItem.objects.filter(pk=item1.pk).exists())
- self.assertFalse(DriveItem.objects.filter(pk=item2.pk).exists())
- self.assertTrue(DriveItem.objects.filter(pk=secret.pk).exists())
- self.assertFalse(path1.exists())
- self.assertFalse(path2.exists())
- self.assertTrue(secret_path.exists())
-
- def test_empty_trash_requires_authentication(self):
- self.client.force_authenticate(user=None)
- response = self.client.delete('/api/drive/trash/')
- self.assertEqual(response.status_code, 401)
-
- def test_empty_trash_never_deletes_active_files(self):
- active, path = self.file(self.shared, self.admin, 'actif.txt', 'actif.txt')
- DriveItem.objects.filter(pk=active.pk).update(trashed_at=None)
- response = self.client.delete('/api/drive/trash/')
- self.assertEqual(response.status_code, 200)
- self.assertEqual(response.data['deleted'], 0)
- self.assertTrue(path.exists())
- self.assertTrue(DriveItem.objects.filter(pk=active.pk).exists())
-
- def test_reader_cannot_empty_trash(self):
- reader = get_user_model().objects.create_user('reader', password='pass')
- SpaceGrant.objects.create(user=reader, space=self.shared, permissions=['view', 'download'])
- item, path = self.file(self.shared, self.admin, 'lecture.txt', '.gooddrive-trash/reader/lecture.txt')
- self.client.force_authenticate(reader)
- response = self.client.delete('/api/drive/trash/')
- self.assertEqual(response.status_code, 200, response.data)
- self.assertEqual(response.data['deleted'], 0)
- self.assertEqual(response.data['skipped'], 1)
- self.assertTrue(path.exists())
- self.assertTrue(DriveItem.objects.filter(pk=item.pk).exists())
diff --git a/backend/drive/test_word_preview.py b/backend/drive/test_word_preview.py
deleted file mode 100644
index b3086c2..0000000
--- a/backend/drive/test_word_preview.py
+++ /dev/null
@@ -1,22 +0,0 @@
-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('