140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
from django.test import TestCase
|
|
from django.core.files.storage import FileSystemStorage
|
|
from django.core.exceptions import ValidationError
|
|
|
|
from .multi import MultiStorage
|
|
from .models import UserStorage
|
|
from .user import BYOStorage
|
|
from .http import HTTPStorage
|
|
|
|
#import logging
|
|
#logging.basicConfig(level=logging.DEBUG)
|
|
|
|
class MultiStorageTestCase(TestCase):
|
|
|
|
def test_storage_selection(self):
|
|
ms = MultiStorage({'one': {'location': 'storage/one'}, 'two': {}})
|
|
self.assertEqual(str(ms), "<MultiStorage: 2 storages>")
|
|
|
|
# use the default if nothing set
|
|
default = ms.get_storage('foo')
|
|
self.assertIsInstance(default, FileSystemStorage)
|
|
self.assertEqual(default.base_location, 'media')
|
|
|
|
# get first instance
|
|
one = ms.get_storage('one')
|
|
self.assertEqual(one.base_location, 'storage/one')
|
|
|
|
def test_storage_proxies(self):
|
|
ms = MultiStorage({
|
|
'one': {'location': 'storage/one', 'base_url': 'http://one'},
|
|
'two': {'location': 'storage/two', 'base_url': 'http://two'}
|
|
})
|
|
|
|
|
|
self.assertEqual(ms.url("one:foo.txt"), 'http://one/foo.txt')
|
|
self.assertEqual(ms.url("two:foo.txt"), 'http://two/foo.txt')
|
|
|
|
def test_with_http(self):
|
|
ms = MultiStorage({
|
|
'https': {'storage': 'byostorage.http.HTTPStorage', 'protocol': 'https'}
|
|
})
|
|
|
|
self.assertEqual(ms.url('https://google.com'), 'https://google.com')
|
|
|
|
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": "polyphonic_test_key",
|
|
"secret_key": "polyphonic_secret",
|
|
"endpoint_url": "http://localhost:9000",
|
|
"bucket_name": "personal"
|
|
}
|
|
''')
|
|
|
|
|
|
def test_bad_connection(self):
|
|
try:
|
|
UserStorage.objects.create(name='one', storage='storages.backends.s3boto3.S3Boto3Storage',
|
|
settings_data='''
|
|
{
|
|
"access_key": "polyphonic_test_key",
|
|
"secret_key": "the_wrong_secret",
|
|
"endpoint_url": "http://localhost:9000",
|
|
"bucket_name": "missing"
|
|
}
|
|
''')
|
|
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 HTTPStorageTestCase(TestCase):
|
|
|
|
def test_url(self):
|
|
s = HTTPStorage()
|
|
|
|
self.assertEqual(s.url('//google.com'), 'https://google.com')
|
|
|
|
def test_exists(self):
|
|
s = HTTPStorage()
|
|
|
|
self.assertTrue(s.exists('//gitea.tfconsulting.com.au/tris'))
|
|
self.assertFalse(s.exists('//gitea.tfconsulting.com.au/foo.txt'))
|
|
|
|
def test_save(self):
|
|
s = HTTPStorage()
|
|
with self.assertRaisesMessage(NotImplementedError, "Unable to save to web locations"):
|
|
s.save('//gitea.tfconsulting.com.au/foo', 'Some content')
|
|
|
|
def test_open(self):
|
|
s = HTTPStorage()
|
|
with s.open('//gitea.tfconsulting.com.au', 'rb') as f:
|
|
data = f.read()
|
|
|
|
self.assertTrue(len(data) > 4000) |