parfait
This commit is contained in:
Louis Mazin
parent
d94a36d9a7
commit
f1f7eed9c6
@ -1,112 +0,0 @@
|
||||
"""Run: USE_SQLITE=true python manage.py test drive.test_granular_permissions"""
|
||||
import tempfile
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase, override_settings
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from rest_framework.test import APIClient
|
||||
from .models import DriveItem, DriveSpace, SpaceGrant, ItemGrant
|
||||
from .spaces import ensure_space, ensure_private_space
|
||||
from .quotas import quota_state, occupancy
|
||||
|
||||
|
||||
class GranularPermissionTests(TestCase):
|
||||
def setUp(self):
|
||||
tmp=tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.dir=Path(tmp.name)
|
||||
settings=override_settings(DRIVE_STORAGE_ROOT=self.dir, FILE_UPLOAD_TEMP_DIR=str(self.dir/'.uploads'))
|
||||
settings.enable();self.addCleanup(settings.disable)
|
||||
User=get_user_model()
|
||||
self.admin=User.objects.create_superuser('test-admin','admin@localhost','ComplexPassword123!')
|
||||
self.member=User.objects.create_user('test-member',password='ComplexPassword123!')
|
||||
self.guest=User.objects.create_user('test-guest',password='ComplexPassword123!')
|
||||
self.root=ensure_space(self.admin)
|
||||
self.private=ensure_private_space(self.member)
|
||||
self.client=APIClient()
|
||||
self.client.force_authenticate(self.admin)
|
||||
|
||||
def test_delegation_and_independent_rights(self):
|
||||
rights=['view','download','manage_access']
|
||||
SpaceGrant.objects.create(user=self.member,space=self.root,permissions=rights,can_invite=True)
|
||||
self.client.force_authenticate(self.member)
|
||||
self.assertEqual(self.client.post('/api/drive/folders/',{'space':self.root.pk,'name':'NO'},format='json').status_code,403)
|
||||
created=self.client.post('/api/drive/invitations/',{
|
||||
'parent_space':self.root.pk,'permissions':['view','download']},format='json')
|
||||
self.assertEqual(created.status_code,201,created.data)
|
||||
escalation=self.client.post('/api/drive/invitations/',{
|
||||
'parent_space':self.root.pk,'permissions':['view','modify']},format='json')
|
||||
self.assertEqual(escalation.status_code,403)
|
||||
no_create=self.client.post('/api/drive/spaces/create/',{
|
||||
'parent_space':self.root.pk,'display_name':'Impossible','quota_bytes':None},format='json')
|
||||
self.assertEqual(no_create.status_code,403)
|
||||
self.client.force_authenticate(self.admin)
|
||||
grant=self.client.post('/api/drive/spaces/users/',{
|
||||
'parent_space':self.root.pk,'user_id':self.guest.pk,
|
||||
'permissions':['view','add']},format='json',QUERY_STRING=f'parent_space={self.root.pk}')
|
||||
self.assertEqual(grant.status_code,201,grant.data)
|
||||
self.client.force_authenticate(self.guest)
|
||||
self.assertEqual(self.client.post('/api/drive/folders/',{'space':self.root.pk,'name':'OK'},format='json').status_code,201)
|
||||
self.assertEqual(self.client.post('/api/drive/invitations/',{'parent_space':self.root.pk},format='json').status_code,403)
|
||||
|
||||
def test_granular_folder_share_and_anonymous_preview(self):
|
||||
folder=DriveItem.objects.create(owner=self.member,space=self.private,name='Photos',is_folder=True)
|
||||
image=DriveItem.objects.create(owner=self.member,space=self.private,parent=folder,name='photo.png',
|
||||
storage_key=str(uuid.uuid4()),storage_location=f'{self.private.relative_path}/Photos/photo.png',size=8,mime_type='image/png')
|
||||
filename=self.dir/image.storage_location
|
||||
filename.parent.mkdir(parents=True,exist_ok=True)
|
||||
filename.write_bytes(b'PNGDATA!')
|
||||
self.client.force_authenticate(self.member)
|
||||
share=self.client.post(f'/api/drive/items/{folder.pk}/shares/',{
|
||||
'user_id':self.guest.pk,'permissions':['view','download','add']},format='json')
|
||||
self.assertEqual(share.status_code,201,share.data)
|
||||
self.client.force_authenticate(self.guest)
|
||||
self.assertEqual(self.client.post('/api/drive/folders/',{
|
||||
'name':'Nouveau','parent':str(folder.pk),'space':self.private.pk},format='json').status_code,201)
|
||||
self.assertEqual(self.client.post(f'/api/drive/items/{image.pk}/trash/').status_code,403)
|
||||
self.client.force_authenticate(self.member)
|
||||
result=self.client.post(f'/api/drive/items/{image.pk}/public-links/',{},format='json')
|
||||
self.assertEqual(result.status_code,201,result.data)
|
||||
anonymous=APIClient()
|
||||
token=result.data['token']
|
||||
inline=anonymous.get(f'/api/public/{token}/files/{image.pk}/?inline=1')
|
||||
self.assertEqual(inline.status_code,200)
|
||||
self.assertEqual(inline['Content-Type'],'image/png')
|
||||
self.assertEqual(b''.join(inline.streaming_content),b'PNGDATA!')
|
||||
partial=anonymous.get(f'/api/public/{token}/files/{image.pk}/?inline=1',HTTP_RANGE='bytes=0-2')
|
||||
self.assertEqual(partial.status_code,206)
|
||||
self.assertEqual(b''.join(partial.streaming_content),b'PNG')
|
||||
self.client.force_authenticate(self.member)
|
||||
self.assertEqual(self.client.patch(f'/api/drive/items/{folder.pk}/shares/{share.data["id"]}/permissions/',{
|
||||
'permissions':['view']},format='json').status_code,200)
|
||||
self.client.force_authenticate(self.guest)
|
||||
self.assertEqual(self.client.post('/api/drive/folders/',{
|
||||
'name':'Interdit','parent':str(folder.pk),'space':self.private.pk},format='json').status_code,403)
|
||||
self.assertEqual(self.client.post(f'/api/drive/items/{image.pk}/link/',{'inline':False},format='json').status_code,403)
|
||||
self.assertEqual(self.client.post(f'/api/drive/items/{image.pk}/link/',{'inline':True},format='json').status_code,200)
|
||||
|
||||
def test_unreserved_child_quota_parent_usage_reduces_child_capacity(self):
|
||||
gib=1024**3
|
||||
self.root.quota_bytes=50*gib
|
||||
self.root.save(update_fields=['quota_bytes'])
|
||||
self.client.force_authenticate(self.admin)
|
||||
res=self.client.post('/api/drive/spaces/create/',{
|
||||
'parent_space':self.root.pk,'display_name':'Sous espace','quota_bytes':30*gib},format='json')
|
||||
self.assertEqual(res.status_code,201,res.data)
|
||||
child=DriveSpace.objects.get(pk=res.data['id'])
|
||||
self.assertEqual(child.parent_id,self.root.pk)
|
||||
DriveItem.objects.create(space=self.root,owner=self.admin,name='parent.bin',size=20*gib,
|
||||
storage_key=str(uuid.uuid4()),storage_location='unused')
|
||||
self.assertEqual(quota_state(self.root)['used_bytes'],20*gib)
|
||||
self.assertEqual(quota_state(child)['used_bytes'],0)
|
||||
DriveItem.objects.filter(space=self.root).update(size=30*gib)
|
||||
self.assertEqual(quota_state(self.root)['used_bytes'],30*gib)
|
||||
self.assertEqual(quota_state(child)['used_bytes'],10*gib)
|
||||
self.assertEqual(quota_state(child)['available_bytes'],20*gib)
|
||||
DriveItem.objects.create(space=child,owner=self.admin,name='child.bin',size=10*gib,
|
||||
storage_key=str(uuid.uuid4()),storage_location='unused2')
|
||||
self.assertGreaterEqual(quota_state(self.root)['used_bytes'],10*gib)
|
||||
self.assertEqual(quota_state(self.root)['used_bytes'],40*gib)
|
||||
self.assertEqual(quota_state(child)['used_bytes'],20*gib)
|
||||
self.assertEqual(occupancy(child)[0],10*gib)
|
||||
@ -1,148 +0,0 @@
|
||||
"""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)
|
||||
|
||||
def test_members_never_lists_viewer_as_direct_added_user(self):
|
||||
"""Hide redundant grants for the viewer or space owner, not real guests."""
|
||||
User = get_user_model()
|
||||
delegate = User.objects.create_user('delegate', password='ComplexPassword123!')
|
||||
guest = User.objects.create_user('real-guest', password='ComplexPassword123!')
|
||||
SpaceGrant.objects.create(space=self.root, user=self.admin, can_invite=True)
|
||||
SpaceGrant.objects.create(space=self.root, user=delegate, can_invite=True)
|
||||
SpaceGrant.objects.create(space=self.root, user=guest)
|
||||
|
||||
response = self.client.get('/api/drive/spaces/members/', {'parent_space': self.root.pk})
|
||||
self.assertEqual(response.status_code, 200, response.data)
|
||||
self.assertEqual({g['user_id'] for g in response.data['grants']}, {delegate.pk, guest.pk})
|
||||
|
||||
self.client.force_authenticate(user=delegate)
|
||||
response = self.client.get('/api/drive/spaces/members/', {'parent_space': self.root.pk})
|
||||
self.assertEqual(response.status_code, 200, response.data)
|
||||
self.assertEqual([g['user_id'] for g in response.data['grants']], [guest.pk])
|
||||
@ -1,272 +0,0 @@
|
||||
"""Private disk layout, 50-Go private cap, and real-byte pooled accounting.
|
||||
|
||||
Run with: USE_SQLITE=true python manage.py test drive.test_private_layout
|
||||
"""
|
||||
import tempfile
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from django.core.management import call_command
|
||||
from django.core.management.base import CommandError
|
||||
from django.db import transaction
|
||||
from django.test import TestCase, override_settings
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from .models import DriveItem, DriveSpace, SpaceInvitation, SpaceGrant
|
||||
from .quotas import (PRIVATE_SPACE_QUOTA_BYTES, QuotaExceeded, assert_space_capacity,
|
||||
assert_updated_quota, locked_quota_tree, occupancy, quota_state)
|
||||
from .spaces import ensure_private_space, ensure_space, create_space, storage_path
|
||||
|
||||
|
||||
class PrivateStorageTests(TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.rootdir = Path(self.temp.name)
|
||||
overrides = override_settings(
|
||||
DRIVE_STORAGE_ROOT=self.rootdir,
|
||||
FILE_UPLOAD_TEMP_DIR=str(self.rootdir / '.uploads'),
|
||||
)
|
||||
overrides.enable()
|
||||
self.addCleanup(overrides.disable)
|
||||
User = get_user_model()
|
||||
self.admin = User.objects.create_superuser('root-admin', 'admin@example.org', 'TestPassword123!')
|
||||
self.shared = ensure_space(self.admin)
|
||||
self.private = ensure_private_space(self.admin)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.admin)
|
||||
|
||||
def test_private_directories_are_separate_from_shared_spaces(self):
|
||||
self.assertEqual(self.private.parent_id, self.shared.pk)
|
||||
self.assertEqual(self.private.relative_path, f'privates/{self.private.directory_name}')
|
||||
self.assertEqual(self.private.quota_bytes, PRIVATE_SPACE_QUOTA_BYTES)
|
||||
self.assertTrue((self.rootdir / self.private.relative_path).is_dir())
|
||||
person = get_user_model().objects.create_user('member', password='TestPassword123!')
|
||||
shared = create_space(person, self.shared, 'Compte membre')
|
||||
private = ensure_private_space(person)
|
||||
self.assertTrue(shared.relative_path.startswith(f'{shared.directory_name}'))
|
||||
self.assertEqual(private.relative_path, f'privates/{private.directory_name}')
|
||||
self.assertEqual(private.parent_id, shared.pk)
|
||||
self.assertNotEqual(shared.directory_name, private.directory_name)
|
||||
|
||||
def test_private_usage_counts_toward_assigned_space_without_reserving(self):
|
||||
self.shared.quota_bytes = 100
|
||||
self.shared.save(update_fields=['quota_bytes'])
|
||||
# Private can be 50 Go even when the assigned budget is just 100 bytes.
|
||||
self.assertEqual(quota_state(self.shared)['available_bytes'], 100)
|
||||
self.assertEqual(quota_state(self.private)['available_bytes'], 100)
|
||||
created = self.client.post('/api/drive/uploads/', {
|
||||
'files': [SimpleUploadedFile('photo.jpg', b'abcdef')]
|
||||
}, format='multipart')
|
||||
self.assertEqual(created.status_code, 201, created.data)
|
||||
node = DriveItem.objects.get(pk=created.data['items'][0]['id'])
|
||||
self.assertEqual(node.storage_location, f'{self.private.relative_path}/photo.jpg')
|
||||
self.assertEqual(storage_path(node).read_bytes(), b'abcdef')
|
||||
self.assertEqual(occupancy(self.shared)[0], 6)
|
||||
self.assertEqual(quota_state(self.shared)['available_bytes'], 94)
|
||||
self.assertEqual(quota_state(self.private)['available_bytes'], 94)
|
||||
self.assertEqual(quota_state(self.private)['quota_bytes'], PRIVATE_SPACE_QUOTA_BYTES)
|
||||
self.assertEqual(quota_state(self.private)['used_bytes'], 6)
|
||||
rejected = self.client.patch(f'/api/drive/spaces/{self.private.pk}/quota/', {
|
||||
'quota_bytes': None
|
||||
}, format='json')
|
||||
self.assertEqual(rejected.status_code, 403)
|
||||
|
||||
def test_actual_private_use_is_pooled_in_100_go_assigned_space(self):
|
||||
person = get_user_model().objects.create_user('quota-user', password='TestPassword123!')
|
||||
assigned = create_space(person, self.shared, '100 Go', quota_bytes=100 * 1024 ** 3)
|
||||
private = ensure_private_space(person)
|
||||
gib = 1024 ** 3
|
||||
self.assertEqual(private.parent_id, assigned.pk)
|
||||
self.assertEqual(private.relative_path, f'privates/{private.directory_name}')
|
||||
self.assertEqual(quota_state(assigned)['available_bytes'], 100 * gib)
|
||||
self.assertEqual(quota_state(private)['available_bytes'], 50 * gib)
|
||||
# Existing 20 Go in shared space. This leaves both limits at their max.
|
||||
DriveItem.objects.create(
|
||||
owner=person, space=assigned, name='shared.bin',
|
||||
storage_key=str(uuid.uuid4()), storage_location=f'{assigned.relative_path}/shared.bin',
|
||||
size=20 * gib,
|
||||
)
|
||||
self.assertEqual(quota_state(assigned)['used_bytes'], 20 * gib)
|
||||
self.assertEqual(quota_state(private)['used_bytes'], 0)
|
||||
DriveItem.objects.create(
|
||||
owner=person, space=private, name='private.bin',
|
||||
storage_key=str(uuid.uuid4()), storage_location=f'{private.relative_path}/private.bin',
|
||||
size=20 * gib,
|
||||
)
|
||||
self.assertEqual(occupancy(assigned)[0], 40 * gib)
|
||||
self.assertEqual(quota_state(assigned)['used_bytes'], 40 * gib)
|
||||
self.assertEqual(quota_state(assigned)['available_bytes'], 60 * gib)
|
||||
self.assertEqual(quota_state(private)['used_bytes'], 20 * gib)
|
||||
self.assertEqual(quota_state(private)['available_bytes'], 30 * gib)
|
||||
with transaction.atomic():
|
||||
with locked_quota_tree(assigned):
|
||||
assert_space_capacity(assigned, 60 * gib)
|
||||
with self.assertRaises(QuotaExceeded):
|
||||
assert_space_capacity(assigned, 60 * gib + 1)
|
||||
assert_updated_quota(assigned, 40 * gib)
|
||||
with self.assertRaisesRegex(ValueError, 'fichiers'):
|
||||
assert_updated_quota(assigned, 39 * gib)
|
||||
|
||||
def test_parent_limit_blocks_private_even_before_50_go(self):
|
||||
self.shared.quota_bytes = 20
|
||||
self.shared.save(update_fields=['quota_bytes'])
|
||||
DriveItem.objects.create(
|
||||
owner=self.admin, space=self.private, name='existing.bin',
|
||||
storage_key=str(uuid.uuid4()), storage_location=f'{self.private.relative_path}/existing.bin', size=19,
|
||||
)
|
||||
self.assertEqual(quota_state(self.private)['available_bytes'], 1)
|
||||
with transaction.atomic():
|
||||
with locked_quota_tree(self.private):
|
||||
assert_space_capacity(self.private, 1)
|
||||
with self.assertRaises(QuotaExceeded):
|
||||
assert_space_capacity(self.private, 2)
|
||||
|
||||
def test_invitation_preview_includes_creator_full_name_and_parent_space(self):
|
||||
self.admin.first_name = 'Camille'
|
||||
self.admin.last_name = 'Martin'
|
||||
self.admin.save(update_fields=['first_name', 'last_name'])
|
||||
result = self.client.post('/api/drive/invitations/', {
|
||||
'parent_space': self.shared.pk,
|
||||
}, format='json')
|
||||
self.assertEqual(result.status_code, 201, result.data)
|
||||
preview = APIClient().get(f"/api/invitations/{result.data['token']}/")
|
||||
self.assertEqual(preview.status_code, 200, preview.data)
|
||||
self.assertEqual(preview.data['inviter_name'], 'Camille Martin')
|
||||
self.assertEqual(preview.data['parent_name'], self.shared.display_name)
|
||||
|
||||
self.admin.first_name = ''
|
||||
self.admin.last_name = ''
|
||||
self.admin.save(update_fields=['first_name', 'last_name'])
|
||||
preview = APIClient().get(f"/api/invitations/{result.data['token']}/")
|
||||
self.assertEqual(preview.status_code, 200, preview.data)
|
||||
self.assertEqual(preview.data['inviter_name'], self.admin.username)
|
||||
|
||||
def test_invitation_has_no_quota_even_from_an_old_client(self):
|
||||
self.shared.quota_bytes = 15
|
||||
self.shared.save(update_fields=['quota_bytes'])
|
||||
# Even an old client cannot allocate storage through an invitation.
|
||||
result = self.client.post('/api/drive/invitations/', {
|
||||
'parent_space': self.shared.pk, 'quota_bytes': 10,
|
||||
}, format='json')
|
||||
self.assertEqual(result.status_code, 201, result.data)
|
||||
self.assertNotIn('quota_bytes', result.data)
|
||||
invitation = SpaceInvitation.objects.get(pk=result.data['id'])
|
||||
self.assertIsNone(invitation.quota_bytes)
|
||||
self.assertEqual(quota_state(self.shared)['available_bytes'], 15)
|
||||
|
||||
# Legacy unredeemed invitations may still contain a saved quota.
|
||||
# The quota must not be applied when the invite is accepted.
|
||||
invitation.quota_bytes = 1
|
||||
invitation.save(update_fields=['quota_bytes'])
|
||||
preview = APIClient().get(f"/api/invitations/{result.data['token']}/")
|
||||
self.assertEqual(preview.status_code, 200, preview.data)
|
||||
self.assertNotIn('quota_bytes', preview.data)
|
||||
accepted = APIClient().post(
|
||||
f"/api/invitations/{result.data['token']}/accept/", {
|
||||
'username': 'invitee', 'password': 'TestPassword123!',
|
||||
'first_name': 'Jean', 'last_name': 'Dupont',
|
||||
'display_name': 'Dossier invité',
|
||||
}, format='json',
|
||||
)
|
||||
self.assertEqual(accepted.status_code, 201, accepted.data)
|
||||
user = get_user_model().objects.get(username='invitee')
|
||||
self.assertFalse(DriveSpace.objects.filter(user=user).exists())
|
||||
self.assertTrue(SpaceGrant.objects.filter(user=user, space=self.shared).exists())
|
||||
assigned = ensure_space(user)
|
||||
private = ensure_private_space(user)
|
||||
self.assertEqual(assigned.pk, self.shared.pk)
|
||||
self.assertEqual(private.quota_bytes, PRIVATE_SPACE_QUOTA_BYTES)
|
||||
self.assertEqual(private.parent_id, self.shared.pk)
|
||||
self.assertEqual(quota_state(assigned)['available_bytes'], 15)
|
||||
self.assertEqual(quota_state(private)['available_bytes'], 15)
|
||||
|
||||
def test_private_pending_uploads_count_in_parent(self):
|
||||
from .models import UploadSession
|
||||
UploadSession.objects.create(
|
||||
owner=self.admin, space=self.private, name='movie.bin', total_size=1234
|
||||
)
|
||||
self.assertEqual(quota_state(self.shared)['reserved_bytes'], 1234)
|
||||
self.assertEqual(quota_state(self.shared)['used_bytes'], 0)
|
||||
self.assertEqual(quota_state(self.private)['available_bytes'], PRIVATE_SPACE_QUOTA_BYTES - 1234)
|
||||
|
||||
def test_private_50_go_upload_limit_is_enforced(self):
|
||||
DriveItem.objects.create(
|
||||
owner=self.admin, space=self.private, name='large.bin',
|
||||
storage_key=str(uuid.uuid4()),
|
||||
storage_location=f'{self.private.relative_path}/large.bin',
|
||||
size=PRIVATE_SPACE_QUOTA_BYTES - 1,
|
||||
)
|
||||
with transaction.atomic():
|
||||
with locked_quota_tree(self.private):
|
||||
with self.assertRaises(QuotaExceeded):
|
||||
assert_space_capacity(self.private, 2)
|
||||
assert_space_capacity(self.private, 1)
|
||||
|
||||
def simulate_legacy_private(self):
|
||||
"""Relocate the empty modern directory to the historical nested location."""
|
||||
new = self.rootdir / self.private.relative_path
|
||||
new.rmdir()
|
||||
old_name = self.private.directory_name
|
||||
old = f'{old_name}'
|
||||
src = self.rootdir / old
|
||||
(src / 'Documents').mkdir(parents=True)
|
||||
(src / 'Documents' / 'facture.pdf').write_bytes(b'PDF CONTENT')
|
||||
self.private.parent = self.shared
|
||||
self.private.relative_path = old
|
||||
self.private.quota_bytes = None
|
||||
self.private.save(update_fields=['parent', 'relative_path', 'quota_bytes'])
|
||||
folder = DriveItem.objects.create(
|
||||
owner=self.admin, space=self.private, name='Documents', is_folder=True,
|
||||
)
|
||||
item = DriveItem.objects.create(
|
||||
owner=self.admin, space=self.private, parent=folder, name='facture.pdf',
|
||||
storage_key=str(uuid.uuid4()), storage_location=f'{old}/Documents/facture.pdf',
|
||||
size=11,
|
||||
)
|
||||
return old, item
|
||||
|
||||
def test_migrate_private_storage_dry_run_and_apply(self):
|
||||
old, item = self.simulate_legacy_private()
|
||||
call_command('move_private_storage')
|
||||
self.private.refresh_from_db()
|
||||
self.assertEqual(self.private.relative_path, old)
|
||||
self.assertTrue((self.rootdir / old / 'Documents' / 'facture.pdf').exists())
|
||||
call_command('move_private_storage', '--apply')
|
||||
self.private.refresh_from_db()
|
||||
item.refresh_from_db()
|
||||
self.assertEqual(self.private.relative_path, f'privates/{self.private.directory_name}')
|
||||
self.assertEqual(self.private.parent_id, self.shared.pk)
|
||||
self.assertEqual(self.private.quota_bytes, PRIVATE_SPACE_QUOTA_BYTES)
|
||||
self.assertEqual(item.storage_location, f'{self.private.relative_path}/Documents/facture.pdf')
|
||||
self.assertEqual(storage_path(item).read_bytes(), b'PDF CONTENT')
|
||||
self.assertFalse((self.rootdir / old).exists())
|
||||
self.assertFalse((self.rootdir / '.gooddrive-move-privates.json').exists())
|
||||
call_command('move_private_storage', '--apply') # Idempotent.
|
||||
|
||||
def test_interrupted_migration_can_resume_after_directory_move(self):
|
||||
from .management.commands.move_private_storage import Command, write_journal
|
||||
old, item = self.simulate_legacy_private()
|
||||
plan = Command().plan()
|
||||
journal = self.rootdir / '.gooddrive-move-privates.json'
|
||||
write_journal(journal, plan)
|
||||
new = self.rootdir / 'privates' / self.private.directory_name
|
||||
(self.rootdir / old).rename(new) # Simulate a crash before SQL update.
|
||||
call_command('move_private_storage', '--apply')
|
||||
item.refresh_from_db()
|
||||
self.assertEqual(storage_path(item).read_bytes(), b'PDF CONTENT')
|
||||
self.assertFalse(journal.exists())
|
||||
|
||||
def test_migration_never_overwrites_existing_destination(self):
|
||||
old, item = self.simulate_legacy_private()
|
||||
target = self.rootdir / 'privates' / self.private.directory_name
|
||||
target.mkdir(parents=True)
|
||||
(target / 'unrelated.txt').write_text('KEEP ME')
|
||||
with self.assertRaises(CommandError):
|
||||
call_command('move_private_storage', '--apply')
|
||||
self.assertEqual((target / 'unrelated.txt').read_text(), 'KEEP ME')
|
||||
self.assertTrue((self.rootdir / old / 'Documents' / 'facture.pdf').exists())
|
||||
item.refresh_from_db()
|
||||
self.assertEqual(item.storage_location, f'{old}/Documents/facture.pdf')
|
||||
@ -1,168 +0,0 @@
|
||||
"""Deletion contract: admin may remove shared spaces, never private spaces.
|
||||
|
||||
Run: USE_SQLITE=true python manage.py test drive.test_protected_deletions
|
||||
"""
|
||||
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, ArchivedDriveAccount
|
||||
from .spaces import ensure_space, ensure_private_space, create_space
|
||||
|
||||
|
||||
class ProtectedDeletionTests(TestCase):
|
||||
def setUp(self):
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.storage = Path(tmp.name)
|
||||
override = override_settings(DRIVE_STORAGE_ROOT=self.storage,
|
||||
FILE_UPLOAD_TEMP_DIR=str(self.storage / '.uploads'))
|
||||
override.enable()
|
||||
self.addCleanup(override.disable)
|
||||
User = get_user_model()
|
||||
self.admin = User.objects.create_superuser('test-admin', 'admin@example.org', 'VeryStrongPass123!')
|
||||
self.alice = User.objects.create_user('test-alice', password='VeryStrongPass123!')
|
||||
self.root = ensure_space(self.admin)
|
||||
self.admin_private = ensure_private_space(self.admin)
|
||||
self.assigned = create_space(self.alice, self.root, 'Alice partagé')
|
||||
self.private = ensure_private_space(self.alice)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.admin)
|
||||
|
||||
def private_file(self):
|
||||
item = DriveItem.objects.create(owner=self.alice, space=self.private,
|
||||
name='mon-secret.txt', storage_key=str(uuid.uuid4()),
|
||||
storage_location=f'{self.private.relative_path}/mon-secret.txt', size=6)
|
||||
path = self.storage / item.storage_location
|
||||
path.write_bytes(b'secret')
|
||||
return item, path
|
||||
|
||||
def test_admin_cannot_browse_or_delete_any_other_private_space(self):
|
||||
item, path = self.private_file()
|
||||
result = self.client.get('/api/drive/spaces/')
|
||||
self.assertEqual(result.status_code, 200, result.data)
|
||||
self.assertNotIn(self.private.pk, [s['id'] for s in result.data['spaces']])
|
||||
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.private.pk}).status_code, 403)
|
||||
self.assertEqual(self.client.get('/api/drive/stats/', {'space':self.private.pk}).status_code, 403)
|
||||
self.assertEqual(self.client.post(f'/api/drive/items/{item.pk}/link/').status_code, 403)
|
||||
self.assertEqual(self.client.delete(f'/api/drive/spaces/{self.private.pk}/',
|
||||
{'confirm_name':self.private.display_name}, format='json').status_code, 403)
|
||||
self.client.force_authenticate(user=self.alice)
|
||||
self.assertEqual(self.client.delete(f'/api/drive/spaces/{self.private.pk}/',
|
||||
{'confirm_name':self.private.display_name}, format='json').status_code, 403)
|
||||
self.assertTrue(path.is_file())
|
||||
|
||||
def test_admin_can_delete_assigned_shared_space_without_private_loss(self):
|
||||
secret, path = self.private_file()
|
||||
shared_path = self.storage / self.assigned.relative_path
|
||||
self.assertTrue(shared_path.is_dir())
|
||||
with self.captureOnCommitCallbacks(execute=True):
|
||||
res = self.client.delete(f'/api/drive/spaces/{self.assigned.pk}/',
|
||||
{'confirm_name':self.assigned.display_name}, format='json')
|
||||
self.assertEqual(res.status_code, 200, res.data)
|
||||
self.assertFalse(DriveSpace.objects.filter(pk=self.assigned.pk).exists())
|
||||
self.assertFalse(shared_path.exists())
|
||||
self.private.refresh_from_db()
|
||||
self.assertEqual(self.private.parent_id, self.root.pk)
|
||||
self.assertTrue(path.exists())
|
||||
self.assertTrue(DriveItem.objects.filter(pk=secret.pk).exists())
|
||||
self.client.force_authenticate(user=self.alice)
|
||||
self.assertIsNone(ensure_space(self.alice))
|
||||
self.assertIsNone(self.client.get('/api/auth/me/').data['space_id'])
|
||||
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.private.pk}).status_code, 200)
|
||||
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.root.pk}).status_code, 403)
|
||||
|
||||
def test_admin_root_reset_keeps_root_and_private_files(self):
|
||||
item, private_path = self.private_file()
|
||||
root_folder = DriveItem.objects.create(owner=self.admin, space=self.root,
|
||||
name='public-folder', is_folder=True)
|
||||
public_path = self.storage / 'public-folder'
|
||||
public_path.mkdir()
|
||||
shared_path = self.storage / self.assigned.relative_path
|
||||
self.assertTrue(shared_path.is_dir())
|
||||
with self.captureOnCommitCallbacks(execute=True):
|
||||
res = self.client.delete(f'/api/drive/spaces/{self.root.pk}/',
|
||||
{'confirm_name':self.root.display_name}, format='json')
|
||||
self.assertEqual(res.status_code, 200, res.data)
|
||||
self.assertTrue(res.data['root_reset'])
|
||||
self.assertTrue(DriveSpace.objects.filter(pk=self.root.pk).exists())
|
||||
self.assertFalse(DriveSpace.objects.filter(pk=self.assigned.pk).exists())
|
||||
self.assertFalse(DriveItem.objects.filter(pk=root_folder.pk).exists())
|
||||
self.assertFalse(shared_path.exists())
|
||||
self.assertFalse(public_path.exists())
|
||||
self.assertTrue(private_path.exists())
|
||||
self.assertTrue(DriveItem.objects.filter(pk=item.pk).exists())
|
||||
self.assertTrue((self.storage/'privates').is_dir())
|
||||
|
||||
def test_user_deletion_removes_private_directory_and_keeps_shared_files(self):
|
||||
item, path = self.private_file()
|
||||
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.admin.pk}/',
|
||||
{'confirm_username':self.admin.username}, format='json').status_code, 403)
|
||||
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
||||
{'confirm_username':'wrong'}, format='json').status_code, 400)
|
||||
shared = DriveItem.objects.create(owner=self.alice, space=self.root,
|
||||
name='communal.txt', storage_key=str(uuid.uuid4()),
|
||||
storage_location='communal.txt', size=4)
|
||||
shared_path = self.storage / 'communal.txt'
|
||||
shared_path.write_bytes(b'team')
|
||||
with self.captureOnCommitCallbacks(execute=True):
|
||||
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
||||
{'confirm_username':self.alice.username}, format='json').status_code, 204)
|
||||
self.alice.refresh_from_db()
|
||||
self.assertFalse(self.alice.is_active)
|
||||
self.assertFalse(self.alice.has_usable_password())
|
||||
self.assertTrue(ArchivedDriveAccount.objects.filter(user=self.alice).exists())
|
||||
self.assertNotIn(self.alice.pk, [u['id'] for u in self.client.get('/api/drive/admin/users/').data['users']])
|
||||
self.assertFalse(path.exists())
|
||||
self.assertFalse((self.storage / self.private.relative_path).exists())
|
||||
self.assertFalse(DriveSpace.objects.filter(pk=self.private.pk).exists())
|
||||
self.assertFalse(DriveItem.objects.filter(pk=item.pk).exists())
|
||||
self.assertTrue(shared_path.exists())
|
||||
self.assertTrue(DriveItem.objects.filter(pk=shared.pk).exists())
|
||||
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
||||
{'confirm_username':self.alice.username}, format='json').status_code, 404)
|
||||
|
||||
def test_non_admin_cannot_delete_other_users(self):
|
||||
self.client.force_authenticate(user=self.alice)
|
||||
self.assertEqual(self.client.delete(f'/api/drive/admin/users/{self.admin.pk}/',
|
||||
{'confirm_username':self.admin.username}, format='json').status_code, 403)
|
||||
|
||||
def test_admin_can_revoke_space_grants_without_removing_private_storage(self):
|
||||
from .models import SpaceGrant
|
||||
# Access granted explicitly by invitation/existing-account grant.
|
||||
grant = SpaceGrant.objects.create(user=self.alice, space=self.root,
|
||||
can_write=True, can_invite=True)
|
||||
rsp = self.client.get(f'/api/drive/admin/users/{self.alice.pk}/access/')
|
||||
self.assertEqual(rsp.status_code, 200, rsp.data)
|
||||
self.assertTrue(any(a['kind'] == 'grant' and a['space_id'] == self.root.pk
|
||||
for a in rsp.data['accesses']))
|
||||
self.assertNotIn(self.private.pk, [a['space_id'] for a in rsp.data['accesses']])
|
||||
rsp = self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/access/grant/{self.root.pk}/')
|
||||
self.assertEqual(rsp.status_code, 204)
|
||||
self.assertFalse(SpaceGrant.objects.filter(pk=grant.pk).exists())
|
||||
self.assertTrue(DriveSpace.objects.filter(pk=self.root.pk).exists())
|
||||
self.assertTrue(DriveSpace.objects.filter(pk=self.private.pk).exists())
|
||||
|
||||
# Older accounts are linked directly to their assigned shared space.
|
||||
rsp = self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/access/owner/{self.assigned.pk}/')
|
||||
self.assertEqual(rsp.status_code, 204)
|
||||
self.assigned.refresh_from_db()
|
||||
self.assertIsNone(self.assigned.user_id)
|
||||
self.assertIsNone(ensure_space(self.alice))
|
||||
self.client.force_authenticate(user=self.alice)
|
||||
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.root.pk}).status_code,403)
|
||||
self.assertEqual(self.client.get('/api/drive/items/', {'space':self.private.pk}).status_code,200)
|
||||
self.assertEqual(self.client.get(f'/api/drive/admin/users/{self.alice.pk}/access/').status_code,403)
|
||||
|
||||
def test_account_delete_rejects_private_folder_outside_privates(self):
|
||||
self.private.relative_path = f'{self.assigned.relative_path}/legacy'
|
||||
self.private.save(update_fields=['relative_path'])
|
||||
original = self.storage / self.private.relative_path
|
||||
original.mkdir(parents=True, exist_ok=True)
|
||||
rsp=self.client.delete(f'/api/drive/admin/users/{self.alice.pk}/',
|
||||
{'confirm_username':self.alice.username},format='json')
|
||||
self.assertEqual(rsp.status_code,409)
|
||||
self.assertTrue(DriveSpace.objects.filter(pk=self.private.pk).exists())
|
||||
self.assertTrue(original.exists())
|
||||
@ -1,120 +0,0 @@
|
||||
"""Regressions: an invitee cannot enlarge their assigned storage allocation.
|
||||
|
||||
Run with: USE_SQLITE=true python manage.py test drive.test_quota_allocation_protection
|
||||
"""
|
||||
import tempfile
|
||||
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 SpaceGrant
|
||||
from .spaces import _make_numbered_space, ensure_space, space_info
|
||||
|
||||
|
||||
class AllocatedQuotaProtectionTests(TestCase):
|
||||
def setUp(self):
|
||||
directory = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(directory.cleanup)
|
||||
settings = override_settings(DRIVE_STORAGE_ROOT=Path(directory.name),
|
||||
FILE_UPLOAD_TEMP_DIR=str(Path(directory.name) / '.uploads'))
|
||||
settings.enable()
|
||||
self.addCleanup(settings.disable)
|
||||
|
||||
User = get_user_model()
|
||||
self.admin = User.objects.create_superuser('quota-admin', 'admin@example.com', 'TestPassword123!')
|
||||
self.member = User.objects.create_user('quota-member', password='TestPassword123!')
|
||||
self.client = APIClient()
|
||||
self.root = ensure_space(self.admin)
|
||||
self.root.quota_bytes = 100 * 1024**3
|
||||
self.root.save(update_fields=['quota_bytes'])
|
||||
self.allocated = _make_numbered_space(
|
||||
self.root, managed_by=self.admin, display_name='Espace attribué', quota_bytes=50 * 1024**3)
|
||||
SpaceGrant.objects.create(
|
||||
user=self.member, space=self.allocated,
|
||||
permissions=['view', 'download', 'manage_space', 'manage_access', 'create_spaces'])
|
||||
self.child_by_admin = _make_numbered_space(
|
||||
self.allocated, managed_by=self.admin, display_name='Sous-espace administrateur',
|
||||
quota_bytes=20 * 1024**3)
|
||||
self.child_by_member = _make_numbered_space(
|
||||
self.allocated, managed_by=self.member, display_name='Sous-espace personnel',
|
||||
quota_bytes=20 * 1024**3)
|
||||
self.client.force_authenticate(self.member)
|
||||
|
||||
def test_member_cannot_change_allocation_through_quota_endpoint(self):
|
||||
response = self.client.patch(
|
||||
f'/api/drive/spaces/{self.allocated.pk}/quota/',
|
||||
{'quota_bytes': 80 * 1024**3}, format='json')
|
||||
self.assertEqual(response.status_code, 403, response.data)
|
||||
self.allocated.refresh_from_db()
|
||||
self.assertEqual(self.allocated.quota_bytes, 50 * 1024**3)
|
||||
|
||||
def test_member_can_rename_allocated_space_but_not_change_quota(self):
|
||||
response = self.client.patch(
|
||||
f'/api/drive/spaces/{self.allocated.pk}/',
|
||||
{'display_name': 'Nom autorisé'}, format='json')
|
||||
self.assertEqual(response.status_code, 200, response.data)
|
||||
forbidden = self.client.patch(
|
||||
f'/api/drive/spaces/{self.allocated.pk}/',
|
||||
{'display_name': 'Nom non enregistré', 'quota_bytes': None}, format='json')
|
||||
self.assertEqual(forbidden.status_code, 403, forbidden.data)
|
||||
self.allocated.refresh_from_db()
|
||||
self.assertEqual(self.allocated.display_name, 'Nom autorisé')
|
||||
self.assertEqual(self.allocated.quota_bytes, 50 * 1024**3)
|
||||
self.assertFalse(space_info(self.allocated, self.member)['can_edit_quota'])
|
||||
|
||||
def test_member_cannot_change_other_users_allocations_using_manage_access(self):
|
||||
response = self.client.patch(
|
||||
f'/api/drive/spaces/members/{self.child_by_admin.pk}/',
|
||||
{'quota_bytes': 40 * 1024**3}, format='json')
|
||||
self.assertEqual(response.status_code, 403, response.data)
|
||||
self.child_by_admin.refresh_from_db()
|
||||
self.assertEqual(self.child_by_admin.quota_bytes, 20 * 1024**3)
|
||||
|
||||
def test_member_can_adjust_the_subspace_they_created(self):
|
||||
self.assertTrue(space_info(self.child_by_member, self.member)['can_edit_quota'])
|
||||
response = self.client.patch(
|
||||
f'/api/drive/spaces/{self.child_by_member.pk}/',
|
||||
{'quota_bytes': 30 * 1024**3}, format='json')
|
||||
self.assertEqual(response.status_code, 200, response.data)
|
||||
self.child_by_member.refresh_from_db()
|
||||
self.assertEqual(self.child_by_member.quota_bytes, 30 * 1024**3)
|
||||
|
||||
def test_admin_still_controls_allocated_quota(self):
|
||||
self.client.force_authenticate(self.admin)
|
||||
response = self.client.patch(
|
||||
f'/api/drive/spaces/{self.allocated.pk}/quota/',
|
||||
{'quota_bytes': 75 * 1024**3}, format='json')
|
||||
self.assertEqual(response.status_code, 200, response.data)
|
||||
self.allocated.refresh_from_db()
|
||||
self.assertEqual(self.allocated.quota_bytes, 75 * 1024**3)
|
||||
self.assertTrue(space_info(self.allocated, self.admin)['can_edit_quota'])
|
||||
|
||||
def test_legacy_assigned_owner_must_not_enlarge_allocation(self):
|
||||
User = get_user_model()
|
||||
legacy = User.objects.create_user('quota-legacy', password='TestPassword123!')
|
||||
allocated = _make_numbered_space(
|
||||
self.root, user=legacy, display_name='Ancien espace attribué',
|
||||
quota_bytes=40 * 1024**3)
|
||||
self.client.force_authenticate(legacy)
|
||||
response = self.client.patch(
|
||||
f'/api/drive/spaces/{allocated.pk}/',
|
||||
{'quota_bytes': 70 * 1024**3}, format='json')
|
||||
self.assertEqual(response.status_code, 403, response.data)
|
||||
allocated.refresh_from_db()
|
||||
self.assertEqual(allocated.quota_bytes, 40 * 1024**3)
|
||||
|
||||
def test_grant_makes_a_created_space_an_assigned_space_not_resizable(self):
|
||||
# Even a creator loses quota control if this space has become their
|
||||
# assigned access point through an explicit share grant.
|
||||
SpaceGrant.objects.create(
|
||||
user=self.member, space=self.child_by_member,
|
||||
permissions=['view', 'download', 'manage_space'])
|
||||
self.assertFalse(space_info(self.child_by_member, self.member)['can_edit_quota'])
|
||||
response = self.client.patch(
|
||||
f'/api/drive/spaces/{self.child_by_member.pk}/',
|
||||
{'quota_bytes': 45 * 1024**3}, format='json')
|
||||
self.assertEqual(response.status_code, 403, response.data)
|
||||
self.child_by_member.refresh_from_db()
|
||||
self.assertEqual(self.child_by_member.quota_bytes, 20 * 1024**3)
|
||||
@ -1,133 +0,0 @@
|
||||
"""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())
|
||||
@ -1,154 +0,0 @@
|
||||
"""Run: USE_SQLITE=true python manage.py test drive.test_space_reparent_and_access"""
|
||||
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 .permissions import space_rights
|
||||
from .spaces import _make_numbered_space, ensure_space, ensure_private_space
|
||||
from .quotas import occupancy
|
||||
|
||||
|
||||
class SpaceReparentingTests(TestCase):
|
||||
def setUp(self):
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.storage = Path(tmp.name)
|
||||
override = override_settings(DRIVE_STORAGE_ROOT=self.storage,
|
||||
FILE_UPLOAD_TEMP_DIR=str(self.storage / '.uploads'))
|
||||
override.enable()
|
||||
self.addCleanup(override.disable)
|
||||
User = get_user_model()
|
||||
self.admin = User.objects.create_superuser('space-admin', 'admin@example.com', 'password-for-tests')
|
||||
self.member = User.objects.create_user('space-member', password='password-for-tests')
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(self.admin)
|
||||
self.root = ensure_space(self.admin)
|
||||
gb = 1024 ** 3
|
||||
self.root.quota_bytes = 100 * gb
|
||||
self.root.save(update_fields=['quota_bytes'])
|
||||
self.a = _make_numbered_space(self.root, managed_by=self.admin,
|
||||
display_name='A', quota_bytes=60 * gb)
|
||||
self.b = _make_numbered_space(self.root, managed_by=self.admin,
|
||||
display_name='B', quota_bytes=50 * gb)
|
||||
self.child = _make_numbered_space(self.a, managed_by=self.admin,
|
||||
display_name='Child', quota_bytes=30 * gb)
|
||||
self.nested = _make_numbered_space(self.child, managed_by=self.admin,
|
||||
display_name='Nested', quota_bytes=10 * gb)
|
||||
self.old_path = self.child.relative_path
|
||||
self.file = DriveItem.objects.create(owner=self.admin, space=self.nested,
|
||||
name='example.pdf', size=15,
|
||||
storage_key=str(uuid.uuid4()),
|
||||
storage_location=f'{self.nested.relative_path}/example.pdf')
|
||||
filepath = self.storage / self.file.storage_location
|
||||
filepath.parent.mkdir(parents=True, exist_ok=True)
|
||||
filepath.write_bytes(b'important bytes')
|
||||
|
||||
def test_move_subtree_preserves_files_and_nested_spaces(self):
|
||||
res = self.client.patch(f'/api/drive/spaces/{self.child.pk}/',
|
||||
{'parent_space': self.b.pk}, format='json')
|
||||
self.assertEqual(res.status_code, 200, res.data)
|
||||
self.child.refresh_from_db()
|
||||
self.nested.refresh_from_db()
|
||||
self.file.refresh_from_db()
|
||||
expected = f'{self.b.relative_path}/{self.child.directory_name}'
|
||||
self.assertEqual(self.child.parent_id, self.b.pk)
|
||||
self.assertEqual(self.child.relative_path, expected)
|
||||
self.assertEqual(self.nested.relative_path, f'{expected}/{self.nested.directory_name}')
|
||||
self.assertEqual(self.file.storage_location,
|
||||
f'{self.nested.relative_path}/example.pdf')
|
||||
self.assertEqual((self.storage / self.file.storage_location).read_bytes(), b'important bytes')
|
||||
self.assertFalse((self.storage / self.old_path).exists())
|
||||
self.assertEqual(occupancy(self.b)[0], 15)
|
||||
self.assertEqual(occupancy(self.a)[0], 0)
|
||||
|
||||
def test_move_refused_if_it_would_create_cycle(self):
|
||||
res = self.client.patch(f'/api/drive/spaces/{self.a.pk}/',
|
||||
{'parent_space': self.nested.pk}, format='json')
|
||||
self.assertEqual(res.status_code, 400, res.data)
|
||||
self.a.refresh_from_db()
|
||||
self.assertEqual(self.a.parent_id, self.root.pk)
|
||||
self.assertTrue((self.storage / self.old_path).exists())
|
||||
|
||||
def test_move_rolls_back_when_new_quota_is_insufficient(self):
|
||||
res = self.client.patch(f'/api/drive/spaces/{self.child.pk}/',
|
||||
{'parent_space': self.b.pk, 'quota_bytes': 0}, format='json')
|
||||
self.assertEqual(res.status_code, 400, res.data)
|
||||
self.child.refresh_from_db()
|
||||
self.file.refresh_from_db()
|
||||
self.assertEqual(self.child.parent_id, self.a.pk)
|
||||
self.assertEqual(self.child.relative_path, self.old_path)
|
||||
self.assertTrue((self.storage / self.file.storage_location).is_file())
|
||||
|
||||
def test_manage_space_without_create_right_on_target_cannot_move(self):
|
||||
SpaceGrant.objects.create(user=self.member, space=self.child,
|
||||
permissions=['view','manage_space'])
|
||||
SpaceGrant.objects.create(user=self.member, space=self.b,
|
||||
permissions=['view'])
|
||||
self.client.force_authenticate(self.member)
|
||||
res = self.client.patch(f'/api/drive/spaces/{self.child.pk}/',
|
||||
{'parent_space': self.b.pk}, format='json')
|
||||
self.assertEqual(res.status_code, 403, res.data)
|
||||
self.child.refresh_from_db()
|
||||
self.assertEqual(self.child.parent_id, self.a.pk)
|
||||
|
||||
def test_private_space_keeps_its_separate_physical_storage(self):
|
||||
SpaceGrant.objects.create(user=self.member, space=self.child,
|
||||
permissions=['view','download'])
|
||||
private = ensure_private_space(self.member)
|
||||
self.assertTrue(private.relative_path.startswith('privates/'))
|
||||
res = self.client.patch(f'/api/drive/spaces/{self.child.pk}/',
|
||||
{'parent_space': self.b.pk}, format='json')
|
||||
self.assertEqual(res.status_code, 200, res.data)
|
||||
private.refresh_from_db()
|
||||
self.assertEqual(private.parent_id, self.child.pk)
|
||||
self.assertTrue((self.storage/private.relative_path).is_dir())
|
||||
|
||||
def test_assigned_space_move_does_not_grant_parent_permissions(self):
|
||||
assigned = _make_numbered_space(self.a, user=self.member, display_name='Assigned',
|
||||
quota_bytes=10 * 1024**3, can_view_parent=True,
|
||||
can_write_parent=True, can_invite=True,
|
||||
parent_permissions=['view','download','add','manage_access'])
|
||||
self.assertIn('manage_access', space_rights(self.member,self.a))
|
||||
res = self.client.patch(f'/api/drive/spaces/{assigned.pk}/',
|
||||
{'parent_space':self.b.pk}, format='json')
|
||||
self.assertEqual(res.status_code,200,res.data)
|
||||
assigned.refresh_from_db()
|
||||
self.assertFalse(assigned.can_view_parent)
|
||||
self.assertFalse(assigned.can_invite)
|
||||
self.assertFalse(space_rights(self.member,self.b))
|
||||
|
||||
def test_move_is_blocked_by_occupied_destination_quota(self):
|
||||
huge = 50 * 1024**3
|
||||
DriveItem.objects.create(owner=self.admin,space=self.b,name='full.bin',size=huge,
|
||||
storage_key=str(uuid.uuid4()),storage_location='legacy-file')
|
||||
res = self.client.patch(f'/api/drive/spaces/{self.child.pk}/',
|
||||
{'parent_space':self.b.pk},format='json')
|
||||
self.assertEqual(res.status_code,400,res.data)
|
||||
self.child.refresh_from_db()
|
||||
self.assertEqual(self.child.parent_id,self.a.pk)
|
||||
self.assertTrue((self.storage/self.old_path).is_dir())
|
||||
|
||||
def test_admin_edits_direct_and_assigned_access_without_creating_duplicates(self):
|
||||
assigned = _make_numbered_space(self.root, user=self.member,
|
||||
display_name='Assigned', quota_bytes=10 * 1024**3)
|
||||
listing = self.client.get(f'/api/drive/admin/users/{self.member.pk}/access/')
|
||||
self.assertEqual(listing.status_code, 200, listing.data)
|
||||
self.assertEqual(sum(a['space_id'] == assigned.pk for a in listing.data['accesses']), 1)
|
||||
url = f'/api/drive/admin/users/{self.member.pk}/access/owner/{assigned.pk}/'
|
||||
saved = self.client.patch(url, {'permissions':['view','download']}, format='json')
|
||||
self.assertEqual(saved.status_code, 200, saved.data)
|
||||
self.assertEqual(space_rights(self.member, assigned), {'view','download'})
|
||||
denied = self.client.patch(url, {'permissions':['modify']}, format='json')
|
||||
self.assertEqual(denied.status_code, 400, denied.data)
|
||||
listing = self.client.get(f'/api/drive/admin/users/{self.member.pk}/access/')
|
||||
self.assertEqual(sum(a['space_id'] == assigned.pk for a in listing.data['accesses']), 1)
|
||||
self.assertEqual(next(access['permissions'] for access in listing.data['accesses']
|
||||
if access['space_id'] == assigned.pk), ['view','download'])
|
||||
revoked = self.client.delete(url)
|
||||
self.assertEqual(revoked.status_code, 204)
|
||||
self.assertFalse(SpaceGrant.objects.filter(user=self.member,space=assigned).exists())
|
||||
@ -1,624 +0,0 @@
|
||||
"""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!'))
|
||||
@ -29,6 +29,12 @@ export default function DriveExplorer({ controller, user }) {
|
||||
}, [display, context, closeContext])
|
||||
useEffect(() => { closeContext() }, [view, activeSpace?.id, closeContext])
|
||||
|
||||
// In "Partagés avec moi", space statistics belong to the last selected
|
||||
// space. Count the displayed shared files and folders instead.
|
||||
const visibleCounts = loading ? null : view === 'shared'
|
||||
? { folders: display.filter(item => item.is_folder).length, files: display.filter(item => !item.is_folder).length }
|
||||
: view === 'files' ? stats : null
|
||||
|
||||
const title = search
|
||||
? `« ${search} »`
|
||||
: trail.length && !search ? trail.at(-1)?.name : view === 'files' ? (activeSpace?.display_name || headerNames[view]) : (view === 'shared' ? 'Partagés avec moi' : headerNames[view])
|
||||
@ -119,7 +125,7 @@ export default function DriveExplorer({ controller, user }) {
|
||||
onChoose={type => performItemAction(context.item,type)} />
|
||||
<footer className="flex justify-between gap-2 px-1 pt-3 text-[11px] text-muted-foreground">
|
||||
<span className="hidden md:inline">Clic droit sur un élément pour les actions.</span>
|
||||
<span>{!isTrash && stats ? `${stats.folders} dossiers · ${stats.files} fichiers` : ''}</span>
|
||||
<span>{visibleCounts ? `${visibleCounts.folders} dossier${visibleCounts.folders === 1 ? '' : 's'} · ${visibleCounts.files} fichier${visibleCounts.files === 1 ? '' : 's'}` : ''}</span>
|
||||
</footer>
|
||||
</section>
|
||||
)
|
||||
|
||||
@ -112,7 +112,7 @@ export default function DriveSidebar({ controller, user, onLogout }) {
|
||||
</div>
|
||||
|
||||
<div className="mt-auto space-y-3 p-4">
|
||||
<div className="rounded-2xl border bg-muted/45 p-4">
|
||||
{view === 'files' && activeSpace && <div className="rounded-2xl border bg-muted/45 p-4">
|
||||
<div className="mb-3 flex items-center justify-between">
|
||||
<span className="flex items-center gap-2 text-xs font-semibold">
|
||||
<Database className="size-4 text-primary" /> Stockage de l’espace
|
||||
@ -131,7 +131,7 @@ export default function DriveSidebar({ controller, user, onLogout }) {
|
||||
{trashBytes > 0 && <p>Corbeille : {fmtSize(trashBytes)} (vider pour libérer)</p>}
|
||||
{reservedBytes > 0 && <p>Transferts réservés : {fmtSize(reservedBytes)} (annuler pour libérer)</p>}
|
||||
</div>
|
||||
</div>
|
||||
</div>}
|
||||
|
||||
{orphaned.length > 0 && <div className="rounded-xl border p-3 text-xs">
|
||||
<p className="mb-2 font-semibold">Imports non terminés ({orphaned.length})</p>
|
||||
|
||||
@ -21,7 +21,7 @@ export function useDriveController(user) {
|
||||
const [loading, setLoading] = useState(true)
|
||||
const [error, setError] = useState('')
|
||||
const [note, setNote] = useState('')
|
||||
const [layout, setLayout] = useState('list')
|
||||
const [layout, setLayout] = useState('grid')
|
||||
const [sort, setSort] = useState('name')
|
||||
const [dark, setDark] = useState(localStorage.getItem('drive_theme') === 'dark')
|
||||
const [sidebar, setSidebar] = useState(false)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user