625 lines
35 KiB
Python
625 lines
35 KiB
Python
"""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)
|
|
|
|
|
|
class SharedSpaceDeletionTests(TestCase):
|
|
"""Deleting managed spaces may never erase an attached user's data."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
override = override_settings(DRIVE_STORAGE_ROOT=Path(self.tmp.name),
|
|
FILE_UPLOAD_TEMP_DIR=str(Path(self.tmp.name) / 'tmp'))
|
|
override.enable()
|
|
self.addCleanup(override.disable)
|
|
self.admin = get_user_model().objects.create_superuser('admin', 'admin@example.test', 'Strong-pass123!')
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(user=self.admin)
|
|
self.root = self.client.get('/api/auth/me/').data['space_id']
|
|
|
|
def create_space(self, name, parent=None):
|
|
reply = self.client.post('/api/drive/spaces/create/',
|
|
{'display_name': name, 'parent_space': parent or self.root}, format='json')
|
|
self.assertEqual(reply.status_code, 201, reply.data)
|
|
return reply.data['id']
|
|
|
|
def delete(self, space_id, confirmed):
|
|
return self.client.delete(f'/api/drive/spaces/{space_id}/',
|
|
{'confirm_name': confirmed}, format='json')
|
|
|
|
def test_deleting_managed_subtree_removes_files_and_uploads(self):
|
|
from .models import UploadSession
|
|
parent = self.create_space('Projet éphémère')
|
|
child = self.create_space('Sous-projet', parent)
|
|
uploaded = self.client.post('/api/drive/uploads/',
|
|
{'space': child, 'files': [SimpleUploadedFile('notes.txt', b'bonjour')]}, format='multipart')
|
|
self.assertEqual(uploaded.status_code, 201, uploaded.data)
|
|
item = DriveItem.objects.get(pk=uploaded.data['items'][0]['id'])
|
|
path = Path(self.tmp.name) / item.storage_location
|
|
self.assertTrue(path.exists())
|
|
UploadSession.objects.create(owner=self.admin, space_id=child,
|
|
name='inachevé.zip', total_size=1024)
|
|
parent_path = Path(self.tmp.name) / DriveSpace.objects.get(pk=parent).relative_path
|
|
with self.captureOnCommitCallbacks(execute=True):
|
|
result = self.delete(parent, 'Projet éphémère')
|
|
self.assertEqual(result.status_code, 200, result.data)
|
|
self.assertFalse(DriveSpace.objects.filter(pk__in=[parent, child]).exists())
|
|
self.assertFalse(DriveItem.objects.filter(pk=item.pk).exists())
|
|
self.assertFalse(parent_path.exists())
|
|
|
|
def test_refuses_wrong_confirmation_root_and_attached_user_space(self):
|
|
managed = self.create_space('Confidentiel')
|
|
self.assertEqual(self.delete(managed, 'autre').status_code, 400)
|
|
self.assertEqual(self.delete(self.root, 'Racine').status_code, 403)
|
|
from .spaces import ensure_private_space, create_space
|
|
assigned = create_space(get_user_model().objects.create_user('alice', password='Strong-pass123!'),
|
|
DriveSpace.objects.get(pk=managed), 'Alice')
|
|
ensure_private_space(assigned.user)
|
|
response = self.delete(managed, 'Confidentiel')
|
|
self.assertEqual(response.status_code, 409, response.data)
|
|
self.assertTrue(DriveSpace.objects.filter(pk=managed).exists())
|
|
self.assertTrue(DriveSpace.objects.filter(pk=assigned.pk).exists())
|
|
|
|
def test_non_manager_cannot_delete_managed_space(self):
|
|
managed = self.create_space('Team')
|
|
alice = get_user_model().objects.create_user('alice', password='Strong-pass123!')
|
|
self.client.force_authenticate(user=alice)
|
|
self.assertEqual(self.delete(managed, 'Team').status_code, 403)
|
|
|
|
|
|
class BootstrapAdminCommandTests(TestCase):
|
|
def test_bootstraps_from_env_only_when_no_superuser_exists(self):
|
|
from unittest.mock import patch
|
|
from django.core.management import call_command
|
|
env = {'ADMIN_USERNAME': 'rootuser', 'ADMIN_PASSWORD': 'Safe-pass123!',
|
|
'ADMIN_FIRST_NAME': 'Ada', 'ADMIN_LAST_NAME': 'Test',
|
|
'ADMIN_EMAIL': 'root@example.test'}
|
|
with patch.dict('os.environ', env):
|
|
call_command('ensure_admin')
|
|
account = get_user_model().objects.get(username='rootuser')
|
|
self.assertTrue(account.is_superuser)
|
|
self.assertTrue(account.is_staff)
|
|
self.assertEqual(account.first_name, 'Ada')
|
|
self.assertTrue(account.check_password('Safe-pass123!'))
|
|
with patch.dict('os.environ', {'ADMIN_PASSWORD': 'Different-pass!'}):
|
|
call_command('ensure_admin')
|
|
account.refresh_from_db()
|
|
self.assertTrue(account.check_password('Safe-pass123!'))
|
|
self.assertEqual(get_user_model().objects.filter(is_superuser=True).count(), 1)
|
|
|
|
def test_existing_username_is_not_promoted_without_permission(self):
|
|
from unittest.mock import patch
|
|
from django.core.management import call_command
|
|
user = get_user_model().objects.create_user('rootuser', password='Current-pass123!')
|
|
with patch.dict('os.environ', {'ADMIN_USERNAME': 'rootuser', 'ADMIN_PASSWORD': 'Other-pass123!'}):
|
|
call_command('ensure_admin')
|
|
user.refresh_from_db()
|
|
self.assertFalse(user.is_superuser)
|
|
self.assertTrue(user.check_password('Current-pass123!'))
|