"""Regression tests for private spaces, delegation and legacy Drive operations. Run with: USE_SQLITE=true python manage.py test drive """ import tempfile from pathlib import Path from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase, override_settings from rest_framework.test import APIClient from .models import DriveItem, DriveSpace, SpaceInvitation, SpaceGrant class HierarchicalDriveTests(TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.overrides=override_settings(DRIVE_STORAGE_ROOT=Path(self.tmp.name), FILE_UPLOAD_TEMP_DIR=str(Path(self.tmp.name)/'tmp')) self.overrides.enable();self.addCleanup(self.overrides.disable) User=get_user_model() self.admin=User.objects.create_superuser('admin', 'admin@example.test','Safe-password123!') self.client=APIClient() self.as_user(self.admin) self.root=self.client.get('/api/auth/me/').data['space_id'] def as_user(self,user): self.client.force_authenticate(user=user) def invite(self,**rights): result=self.client.post('/api/drive/invitations/',rights,format='json') self.assertEqual(result.status_code,201,result.data) return result.data['token'] def accept(self,token,username): self.as_user(None) result=self.client.post(f'/api/invitations/{token}/accept/', {'username':username,'password':'Safe-password123!', 'display_name':username+' perso', 'first_name':username.title(), 'last_name':'Test'},format='json') self.assertEqual(result.status_code,201,result.data) return get_user_model().objects.get(username=username) def folder(self,space_id=None,parent=None,name='Documents'): data={'name':name,'parent':parent} if space_id:data['space']=space_id return self.client.post('/api/drive/folders/',data,format='json') def test_global_trash_across_visible_spaces_without_private_leaks(self): self.as_user(self.admin) token=self.invite(can_view_parent=True) alice=self.accept(token,'alice') token=self.invite() bob=self.accept(token,'bob') self.as_user(self.admin) root_item=self.folder(space_id=self.root,name='root deleted').data['id'] self.client.post(f'/api/drive/items/{root_item}/trash/') self.as_user(alice) personal_item=self.folder(name='alice deleted').data['id'] self.client.post(f'/api/drive/items/{personal_item}/trash/') self.as_user(bob) secret_item=self.folder(name='bob secret').data['id'] self.client.post(f'/api/drive/items/{secret_item}/trash/') self.as_user(alice) # Selecting any space still shows the entire readable trash. result=self.client.get('/api/drive/items/',{'view':'trash','space':self.root}) self.assertEqual(result.status_code,200,result.data) names={item['name'] for item in result.data['items']} self.assertIn('root deleted',names) self.assertIn('alice deleted',names) self.assertNotIn('bob secret',names) self.as_user(self.admin) names={item['name'] for item in self.client.get('/api/drive/items/',{'view':'trash'}).data['items']} self.assertNotIn('alice deleted',names) self.assertNotIn('bob secret',names) def test_standalone_shared_space_creation_and_management(self): self.as_user(self.admin) created=self.client.post('/api/drive/spaces/create/', {'display_name':'Projets communs','parent_space':self.root,'quota_bytes':None},format='json') self.assertEqual(created.status_code,201,created.data) space_id=created.data['id'] space=DriveSpace.objects.get(pk=space_id) self.assertEqual(space.managed_by,self.admin) self.assertRegex(space.directory_name,r'^\d{4,}$') self.assertTrue((Path(self.tmp.name)/space.relative_path).is_dir()) self.assertEqual(self.folder(space_id,name='Équipe').status_code,201) updated=self.client.patch(f'/api/drive/spaces/{space_id}/',{'display_name':'Archives équipe'},format='json') self.assertEqual(updated.status_code,200,updated.data) self.assertEqual(updated.data['display_name'],'Archives équipe') alice=self.accept(self.invite(),'alice') self.as_user(alice) self.assertEqual(self.client.get('/api/drive/items/',{'space':space_id}).status_code,403) self.assertEqual(self.client.patch(f'/api/drive/spaces/{space_id}/',{'display_name':'Piraté'},format='json').status_code,403) def test_signed_preview_supports_byte_ranges_and_checks_revoked_permissions(self): self.as_user(self.admin) new=self.client.post('/api/drive/spaces/create/',{'display_name':'Vidéos'},format='json').data created=self.client.post('/api/drive/uploads/',{'space':new['id'],'files':[SimpleUploadedFile('clip.mp4',b'0123456789')]},format='multipart') self.assertEqual(created.status_code,201,created.data) item_id=created.data['items'][0]['id'] alice=self.accept(self.invite(),'alice') self.as_user(self.admin) SpaceGrant.objects.create(user=alice,space_id=new['id'],can_write=False) self.as_user(alice) token_url=self.client.post(f'/api/drive/items/{item_id}/link/',{'inline':True},format='json') self.assertEqual(token_url.status_code,200,token_url.data) response=self.client.get(token_url.data['url'],HTTP_RANGE='bytes=2-5') self.assertEqual(response.status_code,206) self.assertEqual(response['Content-Range'],'bytes 2-5/10') self.assertEqual(b''.join(response.streaming_content),b'2345') grant=SpaceGrant.objects.get(user=alice,space_id=new['id']) grant.delete() self.assertEqual(self.client.get(token_url.data['url']).status_code,404) def test_numbered_directories_and_parent_readonly(self): token=self.invite(can_view_parent=True,can_write_parent=False) child=self.accept(token,'alice') space=DriveSpace.objects.get(user=child) self.assertRegex(space.directory_name, r'^\d{4,}$') self.assertEqual(space.relative_path, f'root/{space.directory_name}') self.assertTrue((Path(self.tmp.name)/space.relative_path).is_dir()) self.as_user(self.admin) folder=self.folder(name='Racine').data['id'] self.as_user(child) self.assertEqual(self.client.get(f'/api/drive/items/?space={self.root}').status_code,200) self.assertEqual(self.folder(self.root,name='Interdit').status_code,403) self.assertEqual(self.client.patch(f'/api/drive/items/{folder}/',{'name':'Fail'},format='json').status_code,403) self.assertEqual(self.folder(space.pk,name='Partagé').status_code,201) def test_private_sibling_cannot_be_read_or_listed(self): self.as_user(self.admin);a=self.invite() alice=self.accept(a,'alice') self.as_user(self.admin);b=self.invite() bob=self.accept(b,'bob') self.as_user(alice) folder=self.folder(name='Secret') self.assertEqual(folder.status_code,201) alice_space=DriveSpace.objects.get(user=alice) self.as_user(bob) self.assertEqual(self.client.get(f'/api/drive/items/?space={alice_space.pk}').status_code,403) self.assertEqual(self.client.get(f'/api/drive/items/?parent={folder.data["id"]}').status_code,403) self.as_user(self.admin) self.assertEqual(self.client.get(f'/api/drive/items/?space={alice_space.pk}').status_code,200) def test_delegate_invitations_and_revocation(self): self.as_user(self.admin) first=self.invite(can_view_parent=True,can_invite=True) alice=self.accept(first,'alice') self.as_user(alice) second=self.invite(can_view_parent=True) bob=self.accept(second,'bob') bspace=DriveSpace.objects.get(user=bob) self.assertEqual(bspace.parent.user,alice) self.assertEqual(bspace.relative_path, f'{DriveSpace.objects.get(user=alice).relative_path}/{bspace.directory_name}') self.as_user(bob) self.assertEqual(self.client.post('/api/drive/invitations/',{},format='json').status_code,403) self.as_user(alice) pending=self.invite() self.as_user(self.admin) space=DriveSpace.objects.get(user=alice) denied=self.client.patch(f'/api/drive/spaces/members/{space.pk}/',{'can_invite':False},format='json') self.assertEqual(denied.status_code,200) self.as_user(None) self.assertEqual(self.client.get(f'/api/invitations/{pending}/').status_code,404) def test_parent_edit_permission_and_no_cross_space_move(self): self.as_user(self.admin) token=self.invite(can_view_parent=True,can_write_parent=True) child=self.accept(token,'alice') self.as_user(child) folder=self.folder(self.root,name='Sur racine') self.assertEqual(folder.status_code,201) own=self.folder(name='Perso') self.assertEqual(own.status_code,201) moved=self.client.patch(f'/api/drive/items/{own.data["id"]}/',{'parent':folder.data['id']},format='json') self.assertEqual(moved.status_code,403) def test_upload_storage_and_signed_access(self): self.as_user(self.admin) token=self.invite() alice=self.accept(token,'alice') self.as_user(alice) res=self.client.post('/api/drive/uploads/',{'files':[SimpleUploadedFile('a.txt',b'hello')]},format='multipart') self.assertEqual(res.status_code,201,res.data) item=DriveItem.objects.get(pk=res.data['items'][0]['id']) self.assertTrue(item.storage_location.startswith(DriveSpace.objects.get(private_owner=alice).relative_path + '/')) self.assertTrue((Path(self.tmp.name)/item.storage_location).exists()) url=self.client.post(f'/api/drive/items/{item.pk}/link/',{},format='json').data['url'] self.assertEqual(b''.join(self.client.get(url).streaming_content),b'hello') self.as_user(self.admin) self.assertEqual(self.client.post(f'/api/drive/items/{item.pk}/link/').status_code,403) def test_invitation_one_time_only_and_rename_does_not_move_directory(self): self.as_user(self.admin);token=self.invite() alice=self.accept(token,'alice') self.as_user(None) self.assertEqual(self.client.get(f'/api/invitations/{token}/').status_code,404) self.assertEqual(self.client.post(f'/api/invitations/{token}/accept/',{},format='json').status_code,400) self.as_user(alice) space=DriveSpace.objects.get(user=alice) original=space.relative_path changed=self.client.patch('/api/drive/spaces/me/',{'display_name':'Vacances'},format='json') self.assertEqual(changed.status_code,200) space.refresh_from_db() self.assertEqual(space.display_name,'Vacances') self.assertEqual(space.relative_path,original) def test_cannot_grant_parent_write_without_read(self): self.as_user(self.admin) res=self.client.post('/api/drive/invitations/',{'can_view_parent':False,'can_write_parent':True},format='json') self.assertEqual(res.status_code,400) def test_legacy_starred_and_recent_views_not_available(self): self.as_user(self.admin) for view in ('starred', 'recent'): with self.subTest(view=view): result = self.client.get('/api/drive/items/', {'view': view}) self.assertEqual(result.status_code, 400) def test_personal_files_are_private_even_when_shared_space_is_writable(self): self.as_user(self.admin) alice = self.accept(self.invite(can_view_parent=True,can_write_parent=True),'alice') self.as_user(self.admin) bob = self.accept(self.invite(can_view_parent=True,can_write_parent=True),'bob') private_alice=DriveSpace.objects.get(private_owner=alice) assigned_alice=DriveSpace.objects.get(user=alice) self.as_user(alice) secret=self.folder(name='Secret personnel') self.assertEqual(secret.status_code,201,secret.data) self.assertEqual(DriveItem.objects.get(pk=secret.data['id']).space,private_alice) self.assertEqual(self.folder(assigned_alice.pk,name='Dossier partagé').status_code,201) self.as_user(bob) self.assertEqual(self.client.get('/api/drive/items/',{'space':private_alice.pk}).status_code,403) self.assertEqual(self.client.get('/api/drive/items/',{'space':assigned_alice.pk}).status_code,403) self.assertEqual(self.client.patch(f'/api/drive/items/{secret.data["id"]}/',{'name':'Intrusion'},format='json').status_code,403) self.as_user(self.admin) self.assertEqual(self.client.get('/api/drive/items/',{'space':private_alice.pk}).status_code,403) self.assertEqual(self.client.get('/api/drive/stats/',{'space':private_alice.pk}).status_code,403) self.assertEqual(self.client.post(f'/api/drive/items/{secret.data["id"]}/link/',{},format='json').status_code,403) self.assertEqual(self.client.patch(f'/api/drive/items/{secret.data["id"]}/',{'name':'Intrusion admin'},format='json').status_code,403) self.assertEqual(self.client.post(f'/api/drive/items/{secret.data["id"]}/trash/').status_code,403) self.assertEqual(self.client.patch(f'/api/drive/spaces/{private_alice.pk}/quota/',{'quota_bytes':100},format='json').status_code,403) self.assertNotIn(private_alice.pk,[s['id'] for s in self.client.get('/api/drive/spaces/').data['spaces']]) self.as_user(alice) self.assertEqual(self.client.get('/api/drive/items/',{'space':private_alice.pk}).status_code,200) self.assertEqual(self.client.post(f'/api/drive/items/{secret.data["id"]}/trash/').status_code,200) def test_direct_user_access_can_be_granted_and_revoked(self): self.as_user(self.admin) alice=self.accept(self.invite(can_invite=True),'alice') self.as_user(self.admin) bob=self.accept(self.invite(),'bob') alice_shared=DriveSpace.objects.get(user=alice) alice_private=DriveSpace.objects.get(private_owner=alice) self.as_user(alice) search=self.client.get('/api/drive/spaces/users/search/',{'q':'bo'}) self.assertEqual(search.status_code,200,search.data) self.assertEqual(search.data['users'][0]['first_name'],'Bob') self.assertEqual(search.data['users'][0]['last_name'],'Test') grant=self.client.post('/api/drive/spaces/users/',{'user_id':bob.pk,'can_write':False},format='json') self.assertEqual(grant.status_code,201,grant.data) self.as_user(bob) self.assertEqual(self.client.get('/api/drive/items/',{'space':alice_shared.pk}).status_code,200) self.assertEqual(self.folder(alice_shared.pk,name='No write').status_code,403) self.assertEqual(self.client.get('/api/drive/items/',{'space':alice_private.pk}).status_code,403) self.as_user(alice) changed=self.client.patch(f'/api/drive/spaces/users/{grant.data["id"]}/',{'can_write':True},format='json') self.assertEqual(changed.status_code,200,changed.data) self.as_user(bob) self.assertEqual(self.folder(alice_shared.pk,name='Now writable').status_code,201) self.as_user(alice) self.assertEqual(self.client.delete(f'/api/drive/spaces/users/{grant.data["id"]}/').status_code,204) self.as_user(bob) self.assertEqual(self.client.get('/api/drive/items/',{'space':alice_shared.pk}).status_code,403) def test_registration_names_and_virtual_names_only(self): self.as_user(self.admin) token=self.invite() self.as_user(None) incomplete=self.client.post(f'/api/invitations/{token}/accept/',{ 'username':'no_name','password':'Safe-password123!', 'display_name':'Mon espace'},format='json') self.assertEqual(incomplete.status_code,400) person=self.accept(token,'alice') self.as_user(person) me=self.client.get('/api/auth/me/').data self.assertEqual(me['first_name'],'Alice') self.assertEqual(me['last_name'],'Test') all_spaces=self.client.get('/api/drive/spaces/').data self.assertEqual(all_spaces['mine'],me['private_space_id']) self.assertEqual(all_spaces['assigned'],me['space_id']) self.assertNotEqual(all_spaces['mine'],all_spaces['assigned']) self.assertTrue(all('directory' not in space and 'relative_path' not in space for space in all_spaces['spaces'])) class QuotaTests(HierarchicalDriveTests): """Quotas apply to descendants, trash and reservations, not just visible files.""" def upload_small(self, contents, space_id=None): data={'files':[SimpleUploadedFile('data.bin',contents)]} if space_id is not None:data['space']=space_id return self.client.post('/api/drive/uploads/',data,format='multipart') def test_invitation_quota_enforced_direct_and_trash(self): token=self.invite(quota_bytes=5) person=self.accept(token,'limited') space=DriveSpace.objects.get(user=person) self.assertEqual(space.quota_bytes,5) self.as_user(person) first=self.upload_small(b'1234') self.assertEqual(first.status_code,201,first.data) stat=self.client.get('/api/drive/stats/').data self.assertEqual(stat['used_bytes'],4) self.assertEqual(stat['available_bytes'],1) self.assertEqual(self.upload_small(b'89').status_code,413) item_id=first.data['items'][0]['id'] self.assertEqual(self.client.post(f'/api/drive/items/{item_id}/trash/').status_code,200) # Trashed files are still physically stored and consume quota. self.assertEqual(self.client.get('/api/drive/stats/').data['used_bytes'],4) self.assertEqual(self.upload_small(b'89').status_code,413) self.assertEqual(self.client.delete(f'/api/drive/items/{item_id}/delete/').status_code,204) self.assertEqual(self.client.get('/api/drive/stats/').data['available_bytes'],5) self.assertEqual(self.upload_small(b'89').status_code,201) def test_resumable_upload_reserves_entire_file(self): user=self.accept(self.invite(quota_bytes=10),'uploader') self.as_user(user) begin=self.client.post('/api/drive/uploads/start/',{'name':'video.bin','size':8},format='json') self.assertEqual(begin.status_code,201,begin.data) upload_id=begin.data['id'] self.assertEqual(self.client.get('/api/drive/stats/').data['reserved_bytes'],8) self.assertEqual(self.upload_small(b'abc').status_code,413) self.assertEqual(self.client.post('/api/drive/uploads/start/',{'name':'second','size':3},format='json').status_code,413) self.assertEqual(self.client.delete(f'/api/drive/uploads/{upload_id}/cancel/').status_code,204) self.assertEqual(self.client.get('/api/drive/stats/').data['reserved_bytes'],0) self.assertEqual(self.upload_small(b'abc').status_code,201) def test_parent_quota_includes_nested_spaces(self): alice=self.accept(self.invite(quota_bytes=9,can_invite=True),'alice') self.as_user(alice) bob=self.accept(self.invite(quota_bytes=20),'bob') self.as_user(bob) self.assertEqual(self.upload_small(b'x'*7).status_code,201) data=self.client.get('/api/drive/stats/').data self.assertEqual(data['quota_bytes'],20) self.assertEqual(data['available_bytes'],2) # parent's 9-byte cap self.assertEqual(self.upload_small(b'y'*3).status_code,413) self.as_user(alice) self.assertEqual(self.client.get('/api/drive/stats/',{'space':DriveSpace.objects.get(user=alice).pk}).data['used_bytes'],7) self.assertEqual(self.upload_small(b'z'*2).status_code,201) self.assertEqual(self.upload_small(b'z').status_code,413) def test_manager_can_change_child_quota_but_not_below_used(self): alice=self.accept(self.invite(can_invite=True),'alice') self.as_user(alice) bob=self.accept(self.invite(quota_bytes=5),'bob') space=DriveSpace.objects.get(user=bob) self.as_user(bob) self.assertEqual(self.upload_small(b'1234').status_code,201) self.as_user(alice) endpoint=f'/api/drive/spaces/members/{space.pk}/' self.assertEqual(self.client.patch(endpoint,{'quota_bytes':3},format='json').status_code,400) self.assertEqual(self.client.patch(endpoint,{'quota_bytes':8},format='json').status_code,200) space.refresh_from_db();self.assertEqual(space.quota_bytes,8) self.assertEqual(self.client.patch(endpoint,{'quota_bytes':None},format='json').status_code,200) self.assertEqual(self.client.patch(endpoint,{'quota_bytes':-1},format='json').status_code,400) self.as_user(bob) self.assertEqual(self.client.patch(endpoint,{'quota_bytes':0},format='json').status_code,403) def test_admin_root_quota_caps_everyone(self): self.as_user(self.admin) endpoint=f'/api/drive/spaces/{self.root}/quota/' self.assertEqual(self.client.patch(endpoint,{'quota_bytes':4},format='json').status_code,200) person=self.accept(self.invite(quota_bytes=None),'someone') self.as_user(person) self.assertEqual(self.upload_small(b'1234').status_code,201) self.assertEqual(self.upload_small(b'5').status_code,413) self.as_user(self.admin) self.assertEqual(self.client.patch(endpoint,{'quota_bytes':2},format='json').status_code,400) class TransferAndSharedRefreshTests(HierarchicalDriveTests): """Uploads can be aborted without leaving a quota reservation or a file.""" def test_start_chunk_cancel_frees_quota(self): self.as_user(self.admin) started=self.client.post('/api/drive/uploads/start/', {'name':'gros.zip','size':12,'space':self.root},format='json') self.assertEqual(started.status_code,201,started.data) job_id=started.data['id'] chunk=self.client.post(f'/api/drive/uploads/{job_id}/chunk/', {'offset':'0','part':SimpleUploadedFile('bloc.bin',b'abc')},format='multipart') self.assertEqual(chunk.status_code,200,chunk.data) self.assertEqual(chunk.data['offset'],3) progress=self.client.get(f'/api/drive/uploads/{job_id}/') self.assertEqual(progress.data['offset'],3) with self.captureOnCommitCallbacks(execute=True): canceled=self.client.delete(f'/api/drive/uploads/{job_id}/cancel/') self.assertEqual(canceled.status_code,204) self.assertFalse((Path(self.tmp.name)/'tmp'/f'drive-{job_id}.part').exists()) self.assertEqual(self.client.get(f'/api/drive/uploads/{job_id}/').status_code,404) self.assertEqual(self.client.get(f'/api/drive/stats/?space={self.root}').data['reserved_bytes'],0) self.assertFalse(DriveItem.objects.filter(name='gros.zip').exists()) def test_shared_folder_listing_reflects_other_users_trash(self): self.as_user(self.admin) token=self.invite(can_view_parent=True,can_write_parent=True) child=self.accept(token,'member') self.as_user(self.admin) folder=self.folder(space_id=self.root,name='Commun') self.assertEqual(folder.status_code,201) folder_id=folder.data['id'] self.as_user(child) self.assertIn(folder_id, [x['id'] for x in self.client.get(f'/api/drive/items/?space={self.root}').data['items']]) self.as_user(self.admin) self.assertEqual(self.client.post(f'/api/drive/items/{folder_id}/trash/').status_code,200) self.as_user(child) self.assertNotIn(folder_id, [x['id'] for x in self.client.get(f'/api/drive/items/?space={self.root}').data['items']]) class PrivacyAndStorageTests(HierarchicalDriveTests): def test_admin_cannot_read_private_upload_content_or_metadata(self): alice = self.accept(self.invite(), 'alice') self.as_user(alice) uploaded = self.client.post('/api/drive/uploads/', { 'files': [SimpleUploadedFile('private.txt', b'secret-data')] }, format='multipart') self.assertEqual(uploaded.status_code, 201, uploaded.data) pk = uploaded.data['items'][0]['id'] private_id = DriveSpace.objects.get(private_owner=alice).pk self.as_user(self.admin) self.assertEqual(self.client.get('/api/drive/items/', {'space': private_id}).status_code, 403) self.assertEqual(self.client.post(f'/api/drive/items/{pk}/link/', {}, format='json').status_code, 403) self.assertEqual(self.client.delete(f'/api/drive/items/{pk}/delete/').status_code, 403) # Direct Django-admin listing of private DriveItem/UploadSession models is disabled. from django.contrib import admin from .models import UploadSession self.assertFalse(admin.site.is_registered(DriveItem)) self.assertFalse(admin.site.is_registered(UploadSession)) def test_quota_details_distinguish_live_trash_and_pending(self): alice = self.accept(self.invite(quota_bytes=200), 'alice') self.as_user(alice) uploaded = self.client.post('/api/drive/uploads/', { 'files': [SimpleUploadedFile('file.txt', b'1234567')] }, format='multipart') pk = uploaded.data['items'][0]['id'] stats = lambda: self.client.get('/api/drive/stats/').data self.assertEqual((stats()['used_bytes'], stats()['active_bytes'], stats()['trashed_bytes']), (7, 7, 0)) self.assertEqual(self.client.post(f'/api/drive/items/{pk}/trash/').status_code, 200) self.assertEqual((stats()['used_bytes'], stats()['active_bytes'], stats()['trashed_bytes']), (7, 0, 7)) self.assertEqual(self.client.delete(f'/api/drive/items/{pk}/delete/').status_code, 204) self.assertEqual(stats()['used_bytes'], 0) self.assertEqual(stats()['trashed_bytes'], 0) started = self.client.post('/api/drive/uploads/start/', {'name':'incomplete.bin','size':70}, format='json') self.assertEqual(started.status_code, 201) self.assertEqual(stats()['reserved_bytes'], 70) self.assertEqual(self.client.delete(f'/api/drive/uploads/{started.data["id"]}/cancel/').status_code, 204) self.assertEqual(stats()['reserved_bytes'], 0) def test_file_inside_trashed_folder_is_counted_as_trash(self): alice = self.accept(self.invite(), 'alice') self.as_user(alice) folder = self.folder(name='À supprimer') uploaded = self.client.post('/api/drive/uploads/', { 'parent': folder.data['id'], 'files': [SimpleUploadedFile('nested.txt', b'hello')] }, format='multipart') self.assertEqual(uploaded.status_code, 201, uploaded.data) self.client.post(f'/api/drive/items/{folder.data["id"]}/trash/') result = self.client.get('/api/drive/stats/').data self.assertEqual(result['used_bytes'], 5) self.assertEqual(result['active_bytes'], 0) self.assertEqual(result['trashed_bytes'], 5) class InterruptedUploadTests(HierarchicalDriveTests): def test_pending_uploads_belong_only_to_owner(self): alice = self.accept(self.invite(), 'alice') self.as_user(alice) started = self.client.post('/api/drive/uploads/start/', { 'name': 'unfinished.txt', 'size': 71_680 }, format='json') self.assertEqual(started.status_code, 201, started.data) pending = self.client.get('/api/drive/uploads/pending/') self.assertEqual(pending.status_code, 200) self.assertEqual(pending.data['sessions'][0]['size'], 71_680) self.as_user(self.admin) self.assertEqual(self.client.get('/api/drive/uploads/pending/').data['sessions'], []) self.as_user(alice) self.assertEqual(self.client.delete(f'/api/drive/uploads/{started.data["id"]}/cancel/').status_code, 204) self.assertEqual(self.client.get('/api/drive/uploads/pending/').data['sessions'], []) def test_sibling_quotas_cannot_overallocate_parent(self): self.as_user(self.admin) self.assertEqual(self.client.patch(f'/api/drive/spaces/{self.root}/quota/', {'quota_bytes':10_000},format='json').status_code,200) first=self.client.post('/api/drive/spaces/create/', {'display_name':'Premier','parent_space':self.root,'quota_bytes':7_000},format='json') self.assertEqual(first.status_code,201,first.data) second=self.client.post('/api/drive/spaces/create/', {'display_name':'Deuxième','parent_space':self.root,'quota_bytes':4_000},format='json') self.assertEqual(second.status_code,400,second.data) smaller=self.client.post('/api/drive/spaces/create/', {'display_name':'Deuxième','parent_space':self.root,'quota_bytes':3_000},format='json') self.assertEqual(smaller.status_code,201,smaller.data) rejected=self.client.patch(f'/api/drive/spaces/{first.data["id"]}/', {'quota_bytes':8_000},format='json') self.assertEqual(rejected.status_code,400,rejected.data) def test_unlimited_child_inherits_parent_remaining(self): self.as_user(self.admin) self.client.patch(f'/api/drive/spaces/{self.root}/quota/',{'quota_bytes':10_000},format='json') child=self.client.post('/api/drive/spaces/create/', {'display_name':'Sans limite propre','parent_space':self.root,'quota_bytes':None},format='json') self.assertEqual(child.status_code,201,child.data) self.assertIsNone(child.data['quota_bytes']) self.assertEqual(child.data['available_bytes'],10_000) self.assertEqual(self.client.post('/api/drive/spaces/create/', {'display_name':'Réservation','parent_space':self.root,'quota_bytes':4_000},format='json').status_code,201) stats=self.client.get('/api/drive/stats/',{'space':child.data['id']}) self.assertEqual(stats.status_code,200,stats.data) self.assertEqual(stats.data['available_bytes'],6_000) self.assertEqual(self.client.post('/api/drive/uploads/',{'space':child.data['id'], 'files':[SimpleUploadedFile('too-big.bin',b'x'*6_001)]},format='multipart').status_code,413) def test_parent_cannot_be_reduced_below_allocated_children(self): self.as_user(self.admin) self.client.post('/api/drive/spaces/create/', {'display_name':'Réservation','parent_space':self.root,'quota_bytes':1000},format='json') result=self.client.patch(f'/api/drive/spaces/{self.root}/quota/', {'quota_bytes':999},format='json') self.assertEqual(result.status_code,400,result.data)