"""Focused regressions for separate private storage and its 50 Go quota. Run with: USE_SQLITE=true python manage.py test drive.test_private_layout """ import tempfile import uuid from pathlib import Path from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.core.management import call_command from django.core.management.base import CommandError from django.db import transaction from django.test import TestCase, override_settings from rest_framework.test import APIClient from .models import DriveItem, DriveSpace from .quotas import PRIVATE_SPACE_QUOTA_BYTES, QuotaExceeded, assert_space_capacity, locked_quota_tree, occupancy, quota_state from .spaces import ensure_private_space, ensure_space, create_space, storage_path class PrivateStorageTests(TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.rootdir = Path(self.temp.name) overrides = override_settings( DRIVE_STORAGE_ROOT=self.rootdir, FILE_UPLOAD_TEMP_DIR=str(self.rootdir / '.uploads'), ) overrides.enable() self.addCleanup(overrides.disable) User = get_user_model() self.admin = User.objects.create_superuser('root-admin', 'admin@example.org', 'TestPassword123!') self.shared = ensure_space(self.admin) self.private = ensure_private_space(self.admin) self.client = APIClient() self.client.force_authenticate(user=self.admin) def test_private_directories_are_separate_from_shared_spaces(self): self.assertIsNone(self.private.parent_id) self.assertEqual(self.private.relative_path, f'privates/{self.private.directory_name}') self.assertEqual(self.private.quota_bytes, PRIVATE_SPACE_QUOTA_BYTES) self.assertTrue((self.rootdir / self.private.relative_path).is_dir()) person = get_user_model().objects.create_user('member', password='TestPassword123!') shared = create_space(person, self.shared, 'Compte membre') private = ensure_private_space(person) self.assertTrue(shared.relative_path.startswith(f'{shared.directory_name}')) self.assertEqual(private.relative_path, f'privates/{private.directory_name}') self.assertIsNone(private.parent_id) self.assertNotEqual(shared.directory_name, private.directory_name) def test_private_quota_does_not_consume_shared_parent(self): self.shared.quota_bytes = 1 self.shared.save(update_fields=['quota_bytes']) created = self.client.post('/api/drive/uploads/', { 'files': [SimpleUploadedFile('photo.jpg', b'abcdef')] }, format='multipart') self.assertEqual(created.status_code, 201, created.data) node = DriveItem.objects.get(pk=created.data['items'][0]['id']) self.assertEqual(node.storage_location, f'{self.private.relative_path}/photo.jpg') self.assertEqual(storage_path(node).read_bytes(), b'abcdef') self.assertEqual(occupancy(self.shared)[0], 0) self.assertEqual(quota_state(self.private)['quota_bytes'], PRIVATE_SPACE_QUOTA_BYTES) self.assertEqual(quota_state(self.private)['used_bytes'], 6) rejected = self.client.patch(f'/api/drive/spaces/{self.private.pk}/quota/', { 'quota_bytes': None }, format='json') self.assertEqual(rejected.status_code, 403) def test_private_50_go_upload_limit_is_enforced(self): DriveItem.objects.create( owner=self.admin, space=self.private, name='large.bin', storage_key=str(uuid.uuid4()), storage_location=f'{self.private.relative_path}/large.bin', size=PRIVATE_SPACE_QUOTA_BYTES - 1, ) with transaction.atomic(): with locked_quota_tree(self.private): with self.assertRaises(QuotaExceeded): assert_space_capacity(self.private, 2) assert_space_capacity(self.private, 1) def simulate_legacy_private(self): """Relocate the empty modern directory to the historical nested location.""" new = self.rootdir / self.private.relative_path new.rmdir() old_name = self.private.directory_name old = f'{old_name}' src = self.rootdir / old (src / 'Documents').mkdir(parents=True) (src / 'Documents' / 'facture.pdf').write_bytes(b'PDF CONTENT') self.private.parent = self.shared self.private.relative_path = old self.private.quota_bytes = None self.private.save(update_fields=['parent', 'relative_path', 'quota_bytes']) folder = DriveItem.objects.create( owner=self.admin, space=self.private, name='Documents', is_folder=True, ) item = DriveItem.objects.create( owner=self.admin, space=self.private, parent=folder, name='facture.pdf', storage_key=str(uuid.uuid4()), storage_location=f'{old}/Documents/facture.pdf', size=11, ) return old, item def test_migrate_private_storage_dry_run_and_apply(self): old, item = self.simulate_legacy_private() call_command('move_private_storage') self.private.refresh_from_db() self.assertEqual(self.private.relative_path, old) self.assertTrue((self.rootdir / old / 'Documents' / 'facture.pdf').exists()) call_command('move_private_storage', '--apply') self.private.refresh_from_db() item.refresh_from_db() self.assertEqual(self.private.relative_path, f'privates/{self.private.directory_name}') self.assertIsNone(self.private.parent_id) self.assertEqual(self.private.quota_bytes, PRIVATE_SPACE_QUOTA_BYTES) self.assertEqual(item.storage_location, f'{self.private.relative_path}/Documents/facture.pdf') self.assertEqual(storage_path(item).read_bytes(), b'PDF CONTENT') self.assertFalse((self.rootdir / old).exists()) self.assertFalse((self.rootdir / '.gooddrive-move-privates.json').exists()) call_command('move_private_storage', '--apply') # Idempotent. def test_interrupted_migration_can_resume_after_directory_move(self): from .management.commands.move_private_storage import Command, write_journal old, item = self.simulate_legacy_private() plan = Command().plan() journal = self.rootdir / '.gooddrive-move-privates.json' write_journal(journal, plan) new = self.rootdir / 'privates' / self.private.directory_name (self.rootdir / old).rename(new) # Simulate a crash before SQL update. call_command('move_private_storage', '--apply') item.refresh_from_db() self.assertEqual(storage_path(item).read_bytes(), b'PDF CONTENT') self.assertFalse(journal.exists()) def test_migration_never_overwrites_existing_destination(self): old, item = self.simulate_legacy_private() target = self.rootdir / 'privates' / self.private.directory_name target.mkdir(parents=True) (target / 'unrelated.txt').write_text('KEEP ME') with self.assertRaises(CommandError): call_command('move_private_storage', '--apply') self.assertEqual((target / 'unrelated.txt').read_text(), 'KEEP ME') self.assertTrue((self.rootdir / old / 'Documents' / 'facture.pdf').exists()) item.refresh_from_db() self.assertEqual(item.storage_location, f'{old}/Documents/facture.pdf')