test

backend/drive/management/commands/cleanup_uploads.py

@@ -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).'))

backend/drive/management/commands/flatten_drive_root.py

@@ -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.'))

backend/drive/test_empty_trash.py

@@ -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())

backend/drive/test_word_preview.py

@@ -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'''<w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main">
-            <w:body><w:p><w:r><w:t>&lt;script&gt;alert(1)&lt;/script&gt;</w:t></w:r></w:p>
-            <w:tbl><w:tr><w:tc><w:p><w:r><w:t>Tableau</w:t></w:r></w:p></w:tc></w:tr></w:tbl>
-            </w:body></w:document>'''
-            with ZipFile(path, 'w') as archive:
-                archive.writestr('word/document.xml', document)
-            html = render_word_preview(path)
-            self.assertIn('&lt;script&gt;', html)
-            self.assertNotIn('<script>', html)
-            self.assertIn('<table>', html)
-            self.assertIn('Tableau', html)

backend/drive/trash_cleanup.py

@@ -1,61 +0,0 @@
-"""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)

frontend/src/components/drive/dialogs/PreviewDialog.jsx

@@ -79,13 +79,17 @@ export default function PreviewDialog({ controller: c }) {
     <DialogContent
       ref={dialogRef}
       showCloseButton={false}
+      data-preview-fullscreen={fullscreen ? 'true' : undefined}
       className={fullscreen
-        ? 'grid grid-rows-[auto_minmax(0,1fr)_auto] gap-3 overflow-hidden border-0 bg-background p-4'
+        ? 'gooddrive-preview-dialog grid grid-rows-[auto_minmax(0,1fr)_auto] gap-3 overflow-hidden border-0 bg-background p-4'
-        : 'grid max-h-[calc(100dvh-1rem)] w-[calc(100vw-1.5rem)] max-w-[min(96vw,1100px)] grid-rows-[auto_minmax(0,1fr)_auto] overflow-hidden'}
+        : 'gooddrive-preview-dialog grid max-h-[calc(100dvh-1rem)] w-[calc(100vw-1.5rem)] max-w-[min(96vw,1100px)] grid-rows-[auto_minmax(0,1fr)_auto] overflow-hidden'}
+      // The dialog's usual centering uses the CSS `translate` property (Tailwind v4),
+      // not `transform`. Reset both or the fullscreen popup stays offset.
       style={fullscreen ? {
-        top: 0, left: 0, right: 0, bottom: 0, inset: 0,
+        position: 'fixed', inset: 0, top: 0, left: 0, right: 0, bottom: 0,
-        transform: 'none', width: '100vw', maxWidth: 'none', maxHeight: 'none',
+        translate: 'none', transform: 'none', margin: 0,
-        height: '100dvh', borderRadius: 0,
+        boxSizing: 'border-box', width: '100vw', height: '100dvh',
+        maxWidth: 'none', maxHeight: 'none', borderRadius: 0,
       } : undefined}
     >
       <DialogHeader className="min-w-0">

frontend/src/index.css

@@ -41,3 +41,22 @@
 .drive-file:hover .row-action{opacity:1}
 .drive-file:focus-within .row-action{opacity:1}
 @media(max-width:767px){.drive-sidebar{position:fixed;z-index:35;inset:0 auto 0 0;transform:translateX(-110%);transition:transform .2s;box-shadow:0 15px 30px #0002}.drive-sidebar[data-open="true"]{transform:translateX(0)}}
+
+/* Le dialogue d'aperçu garde normalement une translation de centrage.
+   En plein écran, cette translation doit être supprimée, y compris
+   avant le rendu React qui suit l'événement fullscreenchange. */
+.gooddrive-preview-dialog:fullscreen,
+.gooddrive-preview-dialog[data-preview-fullscreen="true"] {
+  position: fixed !important;
+  inset: 0 !important;
+  top: 0 !important;
+  left: 0 !important;
+  width: 100vw !important;
+  height: 100dvh !important;
+  max-width: none !important;
+  max-height: none !important;
+  margin: 0 !important;
+  translate: none !important;
+  transform: none !important;
+  border-radius: 0 !important;
+}