82 lines
4.8 KiB
Python
82 lines
4.8 KiB
Python
import io
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
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
|
|
|
|
class DriveTests(TestCase):
|
|
def setUp(self):
|
|
User=get_user_model()
|
|
self.a=User.objects.create_user('alice',password='Some-password12')
|
|
self.b=User.objects.create_user('bob',password='Some-password12')
|
|
self.tmp=tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.override=override_settings(DRIVE_STORAGE_ROOT=Path(self.tmp.name))
|
|
self.override.enable();self.addCleanup(self.override.disable)
|
|
self.client=APIClient();self.client.force_authenticate(user=self.a)
|
|
def folder(self,name,parent=None):
|
|
return self.client.post('/api/drive/folders/',{'name':name,'parent':parent},format='json')
|
|
def test_owner_and_shared_access(self):
|
|
r=self.folder('Photos');self.assertEqual(r.status_code,201)
|
|
folder=r.data['id']
|
|
self.client.force_authenticate(user=self.b)
|
|
self.assertEqual(self.client.get(f'/api/drive/items/?parent={folder}').status_code,403)
|
|
self.client.force_authenticate(user=self.a)
|
|
self.client.patch(f'/api/drive/items/{folder}/',{'is_shared':True},format='json')
|
|
self.client.force_authenticate(user=self.b)
|
|
self.assertEqual(self.client.get(f'/api/drive/items/?parent={folder}').status_code,200)
|
|
self.assertEqual(self.folder('Nope',folder).status_code,403)
|
|
def test_upload_download_and_trash(self):
|
|
res=self.client.post('/api/drive/uploads/',{'files':[SimpleUploadedFile('hello.txt',b'hello world')]},format='multipart')
|
|
self.assertEqual(res.status_code,201,res.data)
|
|
pk=res.data['items'][0]['id']
|
|
link=self.client.post(f'/api/drive/items/{pk}/link/',{},format='json')
|
|
self.assertEqual(link.status_code,200)
|
|
resp=self.client.get(link.data['url'])
|
|
self.assertEqual(b''.join(resp.streaming_content),b'hello world')
|
|
self.client.force_authenticate(user=self.b)
|
|
self.assertEqual(self.client.post(f'/api/drive/items/{pk}/link/').status_code,403)
|
|
self.client.force_authenticate(user=self.a)
|
|
self.assertEqual(self.client.post(f'/api/drive/items/{pk}/trash/').status_code,200)
|
|
self.assertEqual(self.client.get(link.data['url']).status_code,404)
|
|
self.assertEqual(self.client.post(f'/api/drive/items/{pk}/restore/').status_code,200)
|
|
self.assertEqual(self.client.get(link.data['url']).status_code,200)
|
|
def test_cycle_and_invalid_path(self):
|
|
parent=self.folder('Parent').data['id']
|
|
child=self.folder('Child',parent).data['id']
|
|
self.assertEqual(self.client.patch(f'/api/drive/items/{parent}/',{'parent':child},format='json').status_code,400)
|
|
self.assertEqual(self.folder('../escape').status_code,400)
|
|
def test_shared_parent_descendants(self):
|
|
top=self.folder('Shared').data['id']
|
|
inner=self.folder('Nested',top).data['id']
|
|
self.client.patch(f'/api/drive/items/{top}/',{'is_shared':True},format='json')
|
|
self.client.force_authenticate(user=self.b)
|
|
self.assertEqual(self.client.get(f'/api/drive/items/?parent={inner}').status_code,200)
|
|
def test_resume_upload(self):
|
|
start=self.client.post('/api/drive/uploads/start/',{'name':'big.bin','size':6},format='json')
|
|
self.assertEqual(start.status_code,201,start.data)
|
|
pk=start.data['id']
|
|
chunk=self.client.post(f'/api/drive/uploads/{pk}/chunk/',{
|
|
'offset':'0','part':SimpleUploadedFile('part.bin',b'abc')},format='multipart')
|
|
self.assertEqual(chunk.status_code,200,chunk.data)
|
|
self.assertEqual(self.client.get(f'/api/drive/uploads/{pk}/').data['offset'],3)
|
|
repeated=self.client.post(f'/api/drive/uploads/{pk}/chunk/',{
|
|
'offset':'0','part':SimpleUploadedFile('part.bin',b'abc')},format='multipart')
|
|
self.assertEqual(repeated.status_code,409)
|
|
self.client.post(f'/api/drive/uploads/{pk}/chunk/',{
|
|
'offset':'3','part':SimpleUploadedFile('part.bin',b'def')},format='multipart')
|
|
done=self.client.post(f'/api/drive/uploads/{pk}/finish/')
|
|
self.assertEqual(done.status_code,201,done.data)
|
|
self.assertEqual(done.data['size'],6)
|
|
def test_resume_forbidden_other_user(self):
|
|
pk=self.client.post('/api/drive/uploads/start/',{'name':'private.bin','size':5},format='json').data['id']
|
|
self.client.force_authenticate(user=self.b)
|
|
self.assertEqual(self.client.get(f'/api/drive/uploads/{pk}/').status_code,404)
|
|
def test_anonymous_rejected(self):
|
|
self.client.force_authenticate(user=None)
|
|
self.assertIn(self.client.get('/api/drive/items/').status_code,(401,403))
|