"""Tests: USE_SQLITE=true python manage.py test drive.test_share_permissions_and_retention""" import tempfile from datetime import timedelta from pathlib import Path 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, DriveSpace, ItemGrant, UploadSession from .permissions import item_rights from .spaces import ensure_space, ensure_private_space class ShareAndTrashTests(TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) self.root = Path(tmp.name) cfg = override_settings(DRIVE_STORAGE_ROOT=self.root, FILE_UPLOAD_TEMP_DIR=str(self.root/'.uploads')) cfg.enable() self.addCleanup(cfg.disable) User = get_user_model() self.owner = User.objects.create_user('share-owner',password='abc1234567890') self.recipient = User.objects.create_user('share-recipient',password='abc1234567890') self.space = DriveSpace.objects.create( user=self.owner,relative_path='',display_name='Partagé') self.private = ensure_private_space(self.owner) self.client = APIClient() self.client.force_authenticate(user=self.owner) def make_item(self, name, folder=False, parent=None, space=None): space = space or self.private location = '/'.join(filter(None,(space.relative_path, parent.name if parent else '', name))) entry = DriveItem.objects.create(owner=self.owner, space=space, parent=parent, name=name,is_folder=folder, storage_key='' if folder else 'f'*36, storage_location='' if folder else location,size=0 if folder else 4) path = self.root / location path.parent.mkdir(parents=True, exist_ok=True) if folder:path.mkdir(exist_ok=True) else:path.write_bytes(b'data') return entry, path def share(self, item, permissions): return self.client.post(f'/api/drive/items/{item.pk}/shares/', {'user_id':self.recipient.pk,'permissions':permissions},format='json') def test_file_shares_only_allow_view_download_modify(self): file, _ = self.make_item('photo.jpg') for right in ('add','delete','share','manage_access','create_spaces','manage_space'): result = self.share(file,['view',right]) self.assertEqual(result.status_code,403,(right,result.data)) success = self.share(file,['view','modify']) self.assertEqual(success.status_code,201,success.data) self.client.force_authenticate(user=self.recipient) self.assertIn('modify',item_rights(self.recipient,file)) self.assertNotIn('delete',item_rights(self.recipient,file)) self.assertEqual(self.client.post(f'/api/drive/items/{file.pk}/trash/').status_code,403) response=self.client.patch(f'/api/drive/items/{file.pk}/',{'name':'new.jpg'},format='json') self.assertEqual(response.status_code,200,response.data) self.assertEqual(self.client.patch(f'/api/drive/items/{file.pk}/',{'parent':None},format='json').status_code,200) self.assertEqual(self.client.post(f'/api/drive/items/{file.pk}/public-links/').status_code,403) def test_folder_add_inherited_but_no_delete_or_sharing(self): folder, _ = self.make_item('docs',folder=True) child, _ = self.make_item('a.txt',parent=folder) response=self.share(folder,['view','download','modify','add']) self.assertEqual(response.status_code,201,response.data) self.client.force_authenticate(user=self.recipient) self.assertIn('add',item_rights(self.recipient,folder)) self.assertIn('modify',item_rights(self.recipient,child)) self.assertNotIn('share',item_rights(self.recipient,child)) self.assertEqual(self.client.post(f'/api/drive/folders/',{ 'parent':str(folder.pk),'space':self.private.pk,'name':'new'},format='json').status_code,201) self.assertEqual(self.client.post(f'/api/drive/items/{child.pk}/trash/').status_code,403) self.assertEqual(self.client.patch(f'/api/drive/items/{folder.pk}/',{'parent':None},format='json').status_code,200) def test_old_unsafe_share_rights_are_clamped_and_removing_share_preserves_file(self): file,path=self.make_item('old.txt') grant=ItemGrant.objects.create(item=file,user=self.recipient,created_by=self.owner, permissions=['view','download','add','delete','share','manage_access','manage_space']) self.assertEqual(item_rights(self.recipient,file),{'view','download'}) self.client.force_authenticate(user=self.recipient) shared=self.client.get('/api/drive/items/',{'view':'shared'}) self.assertEqual(shared.status_code,200,shared.data) self.assertEqual(shared.data['items'][0]['share_root_id'],str(file.pk)) self.assertEqual(self.client.post(f'/api/drive/items/{file.pk}/trash/').status_code,403) gone=self.client.delete(f'/api/drive/items/{file.pk}/my-share/') self.assertEqual(gone.status_code,204) self.assertFalse(ItemGrant.objects.filter(pk=grant.pk).exists()) self.assertTrue(DriveItem.objects.filter(pk=file.pk).exists()) self.assertEqual(path.read_bytes(),b'data') def test_old_unsafe_folder_grant_does_not_allow_space_administration(self): folder,_=self.make_item('shared',folder=True) child,_=self.make_item('x.txt',parent=folder) ItemGrant.objects.create(item=folder,user=self.recipient,created_by=self.owner, permissions=['view','download','add','delete','modify','share','manage_access','create_spaces','manage_space']) self.assertEqual(item_rights(self.recipient,child),{'view','download','modify','add'}) self.client.force_authenticate(user=self.recipient) self.assertEqual(self.client.delete(f'/api/drive/items/{child.pk}/my-share/').status_code,204) self.assertTrue(DriveItem.objects.filter(pk=child.pk).exists()) def test_cleanup_after_five_days_with_recent_item_preserved(self): stale,path_stale=self.make_item('old.bin') recent,path_recent=self.make_item('recent.bin') DriveItem.objects.filter(pk=stale.pk).update(trashed_at=timezone.now()-timedelta(days=6)) DriveItem.objects.filter(pk=recent.pk).update(trashed_at=timezone.now()-timedelta(days=4)) # TestCase wraps callbacks in a transaction. Flush on_commit manually. with self.captureOnCommitCallbacks(execute=True): call_command('cleanup_trash',days=5) self.assertFalse(DriveItem.objects.filter(pk=stale.pk).exists()) self.assertFalse(path_stale.exists()) self.assertTrue(DriveItem.objects.filter(pk=recent.pk).exists()) self.assertTrue(path_recent.exists()) def test_cleanup_dry_run_and_active_upload_delay(self): folder,path=self.make_item('pending',folder=True) file,_=self.make_item('part.txt',parent=folder) DriveItem.objects.filter(pk=folder.pk).update(trashed_at=timezone.now()-timedelta(days=7)) call_command('cleanup_trash',days=5,dry_run=True) self.assertTrue(DriveItem.objects.filter(pk=folder.pk).exists()) session=UploadSession.objects.create(owner=self.owner,space=self.private,parent=folder, name='upload.txt',total_size=5) call_command('cleanup_trash',days=5) self.assertTrue(DriveItem.objects.filter(pk=folder.pk).exists()) session.delete() with self.captureOnCommitCallbacks(execute=True): call_command('cleanup_trash',days=5) self.assertFalse(DriveItem.objects.filter(pk=folder.pk).exists()) self.assertFalse(DriveItem.objects.filter(pk=file.pk).exists()) self.assertFalse(path.exists())