131 lines
7.8 KiB
Python
131 lines
7.8 KiB
Python
"""New invitation and file-sharing semantics.
|
|
|
|
Run: USE_SQLITE=true python manage.py test drive.test_item_sharing
|
|
"""
|
|
import tempfile
|
|
import uuid
|
|
from pathlib import Path
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase, override_settings
|
|
from rest_framework.test import APIClient
|
|
from .models import DriveItem, DriveSpace, SpaceGrant
|
|
from .spaces import ensure_space, ensure_private_space
|
|
|
|
|
|
class SharesTests(TestCase):
|
|
def setUp(self):
|
|
tmp=tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.storage=Path(tmp.name)
|
|
overrides=override_settings(DRIVE_STORAGE_ROOT=self.storage,
|
|
FILE_UPLOAD_TEMP_DIR=str(self.storage/'.uploads'))
|
|
overrides.enable()
|
|
self.addCleanup(overrides.disable)
|
|
User=get_user_model()
|
|
self.admin=User.objects.create_superuser('admin-shares','admin@example.org','ComplexPassword123!')
|
|
self.root=ensure_space(self.admin)
|
|
ensure_private_space(self.admin)
|
|
self.client=APIClient()
|
|
self.client.force_authenticate(user=self.admin)
|
|
|
|
def invite_user(self, name, **rights):
|
|
self.client.force_authenticate(user=self.admin)
|
|
created=self.client.post('/api/drive/invitations/',{'parent_space':self.root.pk,**rights},format='json')
|
|
self.assertEqual(created.status_code,201,created.data)
|
|
client=APIClient()
|
|
accepted=client.post(f"/api/invitations/{created.data['token']}/accept/",
|
|
{'username':name,'password':'ComplexPassword123!','first_name':'Test','last_name':name},format='json')
|
|
self.assertEqual(accepted.status_code,201,accepted.data)
|
|
return get_user_model().objects.get(username=name)
|
|
|
|
def test_new_invitee_uses_existing_space_without_new_shared_directory(self):
|
|
alice=self.invite_user('alice-share',can_write_parent=False,can_invite=True)
|
|
self.assertFalse(DriveSpace.objects.filter(user=alice).exists())
|
|
self.assertTrue(SpaceGrant.objects.filter(user=alice,space=self.root,can_invite=True,can_write=False).exists())
|
|
self.assertEqual(ensure_space(alice).pk,self.root.pk)
|
|
self.assertEqual(ensure_private_space(alice).parent_id,self.root.pk)
|
|
self.client.force_authenticate(user=alice)
|
|
self.assertEqual(self.client.post('/api/drive/folders/',{'space':self.root.pk,'name':'not-writable'},format='json').status_code,403)
|
|
too_much=self.client.post('/api/drive/invitations/',{
|
|
'parent_space':self.root.pk,'can_view_parent':True,'can_write_parent':True},format='json')
|
|
self.assertEqual(too_much.status_code,403)
|
|
allowed=self.client.post('/api/drive/invitations/',{
|
|
'parent_space':self.root.pk,'can_view_parent':True,'can_invite':True},format='json')
|
|
self.assertEqual(allowed.status_code,201,allowed.data)
|
|
# Creating a subspace is an explicit operation for users with can_invite.
|
|
sub=self.client.post('/api/drive/spaces/create/',{
|
|
'parent_space':self.root.pk,'display_name':'Equipe','quota_bytes':None},format='json')
|
|
self.assertEqual(sub.status_code,201,sub.data)
|
|
|
|
def test_private_item_share_is_read_only_and_revocable(self):
|
|
alice=self.invite_user('alice-files')
|
|
bob=self.invite_user('bob-files')
|
|
private=ensure_private_space(alice)
|
|
folder=DriveItem.objects.create(owner=alice,space=private,name='Documents',is_folder=True)
|
|
child=DriveItem.objects.create(owner=alice,space=private,parent=folder,name='photo.jpg',
|
|
storage_key=str(uuid.uuid4()),storage_location=f'{private.relative_path}/Documents/photo.jpg',size=3)
|
|
secret=DriveItem.objects.create(owner=alice,space=private,name='secret.jpg',
|
|
storage_key=str(uuid.uuid4()),storage_location=f'{private.relative_path}/secret.jpg',size=3)
|
|
self.client.force_authenticate(user=alice)
|
|
shared=self.client.post(f'/api/drive/items/{folder.pk}/shares/',{'user_id':bob.pk},format='json')
|
|
self.assertEqual(shared.status_code,201,shared.data)
|
|
self.client.force_authenticate(user=bob)
|
|
listing=self.client.get('/api/drive/items/',{'view':'shared'})
|
|
self.assertEqual(listing.status_code,200,listing.data)
|
|
self.assertEqual([item['name'] for item in listing.data['items']],['Documents'])
|
|
children=self.client.get('/api/drive/items/',{'view':'shared','parent':str(folder.pk)})
|
|
self.assertEqual(children.status_code,200,children.data)
|
|
self.assertEqual(children.data['items'][0]['id'],str(child.pk))
|
|
self.assertFalse(children.data['items'][0]['can_write'])
|
|
self.assertEqual(self.client.post(f'/api/drive/items/{child.pk}/link/').status_code,200)
|
|
self.assertEqual(self.client.post(f'/api/drive/items/{secret.pk}/link/').status_code,403)
|
|
self.assertEqual(self.client.patch(f'/api/drive/items/{child.pk}/',{'name':'change.jpg'},format='json').status_code,403)
|
|
self.client.force_authenticate(user=alice)
|
|
self.assertEqual(self.client.delete(f'/api/drive/items/{folder.pk}/shares/{shared.data["id"]}/').status_code,204)
|
|
self.client.force_authenticate(user=bob)
|
|
self.assertEqual(self.client.post(f'/api/drive/items/{child.pk}/link/').status_code,403)
|
|
|
|
def test_public_link_access_and_revocation(self):
|
|
alice=self.invite_user('alice-public')
|
|
private=ensure_private_space(alice)
|
|
folder=DriveItem.objects.create(owner=alice,space=private,name='Photos',is_folder=True)
|
|
image=DriveItem.objects.create(owner=alice,space=private,parent=folder,name='pic.png',
|
|
storage_key=str(uuid.uuid4()),storage_location=f'{private.relative_path}/Photos/pic.png',size=3)
|
|
path=self.storage/image.storage_location
|
|
path.parent.mkdir(parents=True,exist_ok=True)
|
|
path.write_bytes(b'PNG')
|
|
self.client.force_authenticate(user=alice)
|
|
public=self.client.post(f'/api/drive/items/{folder.pk}/public-links/',{},format='json')
|
|
self.assertEqual(public.status_code,201,public.data)
|
|
token=public.data['token']
|
|
anonymous=APIClient()
|
|
res=anonymous.get(f'/api/public/{token}/')
|
|
self.assertEqual(res.status_code,200,res.data)
|
|
self.assertEqual(res.data['items'][0]['name'],'pic.png')
|
|
content=anonymous.get(f'/api/public/{token}/files/{image.pk}/')
|
|
self.assertEqual(content.status_code,200)
|
|
self.assertEqual(b''.join(content.streaming_content),b'PNG')
|
|
links=self.client.get(f'/api/drive/items/{folder.pk}/shares/').data['links']
|
|
self.assertEqual(self.client.delete(f'/api/drive/items/{folder.pk}/public-links/{links[0]["id"]}/').status_code,204)
|
|
self.assertEqual(anonymous.get(f'/api/public/{token}/').status_code,404)
|
|
|
|
# A directly shared file must download without either a login or JWT.
|
|
direct=self.client.post(f'/api/drive/items/{image.pk}/public-links/',{},format='json')
|
|
self.assertEqual(direct.status_code,201,direct.data)
|
|
direct_token=direct.data['token']
|
|
direct_preview=anonymous.get(f'/api/public/{direct_token}/')
|
|
self.assertEqual(direct_preview.status_code,200,direct_preview.data)
|
|
self.assertFalse(direct_preview.data['root']['is_folder'])
|
|
direct_file=anonymous.get(f'/api/public/{direct_token}/files/{image.pk}/')
|
|
self.assertEqual(direct_file.status_code,200)
|
|
self.assertEqual(b''.join(direct_file.streaming_content),b'PNG')
|
|
|
|
def test_admin_user_list_is_private_to_admin(self):
|
|
alice=self.invite_user('alice-admin-list')
|
|
self.client.force_authenticate(user=alice)
|
|
self.assertEqual(self.client.get('/api/drive/admin/users/').status_code,403)
|
|
self.client.force_authenticate(user=self.admin)
|
|
users=self.client.get('/api/drive/admin/users/',{'q':'alice-admin-list'})
|
|
self.assertEqual(users.status_code,200,users.data)
|
|
self.assertEqual(users.data['users'][0]['space_name'],self.root.display_name)
|