111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
from django.test import TestCase
|
|
from django.core.files.storage import FileSystemStorage
|
|
from django.core.exceptions import ValidationError
|
|
|
|
from byostorage.models import UserStorage
|
|
from byostorage.user import BYOStorage
|
|
|
|
from example.models import LinkedStorageTestModel
|
|
|
|
import tempfile
|
|
import os.path
|
|
|
|
#import logging
|
|
#logging.basicConfig(level=logging.DEBUG)
|
|
|
|
class UserStorageTestCase(TestCase):
|
|
|
|
def test_construction(self):
|
|
us = UserStorage.objects.create(name='one', storage='django.core.files.storage.FileSystemStorage',
|
|
settings_data='{"location": "storage/one"}')
|
|
|
|
self.assertEqual(str(us), 'one')
|
|
|
|
storage = us.instance()
|
|
self.assertIsInstance(storage, FileSystemStorage)
|
|
self.assertEqual(storage.base_location, 'storage/one')
|
|
|
|
def test_bad_json(self):
|
|
with self.assertRaisesMessage(ValidationError, "Unterminated string"):
|
|
UserStorage.objects.create(name='one', storage='django.core.files.storage.FileSystemStorage',
|
|
settings_data='{"location": "foo}')
|
|
|
|
def test_connection(self):
|
|
instance = UserStorage.objects.create(name='one', storage='storages.backends.s3boto3.S3Boto3Storage',
|
|
settings_data='''
|
|
{
|
|
"access_key": "minioadmin",
|
|
"secret_key": "minioadmin",
|
|
"endpoint_url": "http://localhost:9000",
|
|
"bucket_name": "byostorage_test"
|
|
}
|
|
''')
|
|
|
|
|
|
def test_bad_connection(self):
|
|
try:
|
|
UserStorage.objects.create(name='one', storage='storages.backends.s3boto3.S3Boto3Storage',
|
|
settings_data='''
|
|
{
|
|
"access_key": "minioadmin",
|
|
"secret_key": "the_wrong_secret",
|
|
"endpoint_url": "http://localhost:9000",
|
|
"bucket_name": "byostorage_test"
|
|
}
|
|
''')
|
|
self.fail("Should have raised an exception")
|
|
except Exception as e:
|
|
pass
|
|
|
|
|
|
class BYOStorageTestCase(TestCase):
|
|
|
|
def setUp(self):
|
|
UserStorage.objects.create(name='one', storage='django.core.files.storage.FileSystemStorage',
|
|
settings_data='{"location": "storage/one"}')
|
|
|
|
def test_pickup_changes(self):
|
|
storage = BYOStorage()
|
|
one = storage.get_storage('one')
|
|
self.assertEqual(one.base_location, 'storage/one')
|
|
|
|
# if not already in cache then no issues
|
|
UserStorage.objects.create(name='two', storage='django.core.files.storage.FileSystemStorage',
|
|
settings_data='{"location": "storage/two"}')
|
|
two = storage.get_storage('two')
|
|
self.assertEqual(two.base_location, 'storage/two')
|
|
|
|
# modify the settings
|
|
o = UserStorage.objects.get(name='one')
|
|
o.settings_data = '{"location": "other/one"}'
|
|
o.save()
|
|
|
|
# signal should have fired and invalidated 'one'
|
|
one = storage.get_storage('one')
|
|
self.assertEqual(one.base_location, 'other/one')
|
|
|
|
class BYOStorageFieldTestCase(TestCase):
|
|
|
|
def setUp(self):
|
|
self.test_dir = tempfile.TemporaryDirectory()
|
|
self.one = UserStorage.objects.create(name='one', storage='django.core.files.storage.FileSystemStorage',
|
|
settings_data = '{"location": "%s/one"}' % self.test_dir.name)
|
|
|
|
def tearDown(self):
|
|
self.test_dir.cleanup()
|
|
|
|
def test_linked_storage(self):
|
|
|
|
from django.core.files.uploadedfile import UploadedFile
|
|
|
|
with open(__file__, 'rb') as f:
|
|
result = LinkedStorageTestModel.objects.create(name="first", my_storage_id=self.one.pk, my_file=UploadedFile(f))
|
|
|
|
expected = os.path.basename(__file__)
|
|
|
|
self.assertEqual(result.my_file.name, expected)
|
|
self.assertGreater(result.my_file.size, 0)
|
|
|
|
self.assertEqual(os.listdir(os.path.join(self.test_dir.name, 'one')), [expected])
|
|
|