88 lines
4.8 KiB
Python
88 lines
4.8 KiB
Python
"""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)
|