From 1ca1219424f46d8885af12ae198892157848a201 Mon Sep 17 00:00:00 2001 From: Louis Mazin Date: Wed, 14 Oct 2026 19:01:31 +0200 Subject: [PATCH] test --- .../management/commands/cleanup_uploads.py | 24 --- .../management/commands/flatten_drive_root.py | 172 ------------------ backend/drive/test_empty_trash.py | 83 --------- backend/drive/test_word_preview.py | 22 --- backend/drive/trash_cleanup.py | 61 ------- .../drive/dialogs/PreviewDialog.jsx | 14 +- frontend/src/index.css | 19 ++ 7 files changed, 28 insertions(+), 367 deletions(-) delete mode 100644 backend/drive/management/commands/cleanup_uploads.py delete mode 100644 backend/drive/management/commands/flatten_drive_root.py delete mode 100644 backend/drive/test_empty_trash.py delete mode 100644 backend/drive/test_word_preview.py delete mode 100644 backend/drive/trash_cleanup.py 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('