2021-03-22 15:39:50 +11:00

120 lines
4.1 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
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_bad_settings(self):
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"
}
''')
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)