diff --git a/requirements-optional.txt b/requirements-optional.txt index b7d210a..3cf06c2 100644 --- a/requirements-optional.txt +++ b/requirements-optional.txt @@ -1,5 +1,5 @@ -requests -Werkzeug -requests-aws4auth >= 0.9 -requests-aliyun >= 0.2.5 +httpx paramiko +git+https://github.com/tedder/requests-aws4auth.git@master +requests-aliyun >= 0.2.5 +Werkzeug \ No newline at end of file diff --git a/sqlalchemy_media/stores/s3.py b/sqlalchemy_media/stores/s3.py index 2e95c34..445c5ca 100644 --- a/sqlalchemy_media/stores/s3.py +++ b/sqlalchemy_media/stores/s3.py @@ -2,9 +2,9 @@ # Importing optional stuff required by http based store try: - import requests + import httpx except ImportError: # pragma: no cover - requests = None + httpx = None # Importing optional stuff required by S3 store @@ -38,7 +38,7 @@ class S3Store(Store): def __init__(self, bucket: str, access_key: str, secret_key: str, region: str, max_age: int = DEFAULT_MAX_AGE, prefix: str = None, base_url: str = None, - cdn_url: str = None, cdn_prefix_ignore: bool = False, + cdn_url: str = None, cdn_prefix_ignore: bool = False, acl: str = 'private'): self.bucket = bucket self.access_key = access_key @@ -68,7 +68,7 @@ def __init__(self, bucket: str, access_key: str, secret_key: str, def _get_s3_url(self, filename: str): return '{0}/{1}'.format(self.base_url, filename) - def _upload_file(self, url: str, data: str, content_type: str, + def _upload_file(self, url: str, content: str, content_type: str, rrs: bool = False): ensure_aws4auth() @@ -84,23 +84,23 @@ def _upload_file(self, url: str, data: str, content_type: str, } if content_type: headers['Content-Type'] = content_type - res = requests.put(url, auth=auth, data=data, headers=headers) + res = httpx.put(url, auth=auth, content=content, headers=headers) if not 200 <= res.status_code < 300: raise S3Error(res.text) def put(self, filename: str, stream: FileLike): url = self._get_s3_url(filename) - data = stream.read() + content = stream.read() content_type = getattr(stream, 'content_type', None) rrs = getattr(stream, 'reproducible', False) - self._upload_file(url, data, content_type, rrs=rrs) - return len(data) + self._upload_file(url, content, content_type, rrs=rrs) + return len(content) def delete(self, filename: str): ensure_aws4auth() url = self._get_s3_url(filename) auth = AWS4Auth(self.access_key, self.secret_key, self.region, 's3') - res = requests.delete(url, auth=auth) + res = httpx.delete(url, auth=auth) if not 200 <= res.status_code < 300: raise S3Error(res.text) @@ -108,7 +108,7 @@ def open(self, filename: str, mode: str = 'rb') -> FileLike: ensure_aws4auth() url = self._get_s3_url(filename) auth = AWS4Auth(self.access_key, self.secret_key, self.region, 's3') - res = requests.get(url, auth=auth) + res = httpx.get(url, auth=auth) if not 200 <= res.status_code < 300: raise S3Error(res.text) return BytesIO(res.content) diff --git a/sqlalchemy_media/tests/helpers/__init__.py b/sqlalchemy_media/tests/helpers/__init__.py index 244d202..68d5978 100644 --- a/sqlalchemy_media/tests/helpers/__init__.py +++ b/sqlalchemy_media/tests/helpers/__init__.py @@ -1,7 +1,7 @@ - +from .config import TEST_ACCESS_KEY, TEST_BUCKET, TEST_REGION, TEST_SECRET_KEY from .http import simple_http_server, encode_multipart_data from .os2 import mockup_os2_server -from .s3 import mockup_s3_server +from .s3 import mockup_s3_server, create_s3_store, S3TestCase from .ssh import MockupSSHServer, MockupSSHTestCase from .ftp import MockFTP from .static import mockup_http_static_server diff --git a/sqlalchemy_media/tests/helpers/config.py b/sqlalchemy_media/tests/helpers/config.py new file mode 100644 index 0000000..a88d5c7 --- /dev/null +++ b/sqlalchemy_media/tests/helpers/config.py @@ -0,0 +1,7 @@ + +# Settings for S3 storage + +TEST_BUCKET = 'test-bucket' +TEST_ACCESS_KEY = 'test_access_key' +TEST_SECRET_KEY = 'test_secret_key' +TEST_REGION = 'ap-northeast-2' diff --git a/sqlalchemy_media/tests/helpers/s3.py b/sqlalchemy_media/tests/helpers/s3.py index 4f6853e..f04022a 100644 --- a/sqlalchemy_media/tests/helpers/s3.py +++ b/sqlalchemy_media/tests/helpers/s3.py @@ -1,8 +1,12 @@ +import unittest import contextlib from wsgiref.simple_server import WSGIRequestHandler, WSGIServer -import requests +import httpx +from sqlalchemy_media.stores import S3Store + +from .config import TEST_ACCESS_KEY, TEST_BUCKET, TEST_REGION, TEST_SECRET_KEY from .http import simple_http_server @@ -13,13 +17,37 @@ def mockup_s3_server(bucket, **kwargs): mock_app.debug = False with simple_http_server( WSGIRequestHandler, - server_class=WSGIServer, - app=mock_app, + server_class=WSGIServer, + app=mock_app, **kwargs ) as server: url = 'http://localhost:%s' % server.server_address[1] # Create the bucket bucket_uri = '%s/%s' % (url, bucket) - res = requests.put(bucket_uri) + res = httpx.put(bucket_uri) assert res.status_code == 200 yield server, bucket_uri + + +def create_s3_store(bucket=TEST_BUCKET, **kwargs): + return S3Store( + bucket, + TEST_ACCESS_KEY, + TEST_SECRET_KEY, + TEST_REGION, + **kwargs + ) + + +class S3TestCase(unittest.TestCase): + """Mixin for runnning the S3 server""" + + def run(self, result): + with mockup_s3_server(bucket=TEST_BUCKET) as (server, bucket_uri): + self.storage = create_s3_store( + bucket=TEST_BUCKET, base_url=bucket_uri + ) + self.bucket_name = TEST_BUCKET + self.server = server + self.base_url = bucket_uri + super(S3TestCase, self).run(result) diff --git a/sqlalchemy_media/tests/helpers/testcases.py b/sqlalchemy_media/tests/helpers/testcases.py index b212372..ed5a9eb 100644 --- a/sqlalchemy_media/tests/helpers/testcases.py +++ b/sqlalchemy_media/tests/helpers/testcases.py @@ -10,6 +10,9 @@ from sqlalchemy_media import StoreManager, FileSystemStore +from .config import TEST_BUCKET +from .s3 import mockup_s3_server, create_s3_store + class SqlAlchemyTestCase(unittest.TestCase): @classmethod @@ -20,13 +23,17 @@ def setUp(self): self.Base = declarative_base() self.engine = create_engine(self.db_uri, echo=False) - def create_all_and_get_session(self): + def create_all_and_get_session(self, expire_on_commit: bool = False): + """ + A factory method for making a SQLAlchemy session factory object. + :param expire_on_commit: @see: https://docs.sqlalchemy.org/en/14/orm/session_api.html?highlight=expire_on_commit#sqlalchemy.orm.Session.params.expire_on_commit + """ self.Base.metadata.create_all(self.engine, checkfirst=True) self.session_factory = sessionmaker( bind=self.engine, autoflush=False, autocommit=False, - expire_on_commit=True, + expire_on_commit=expire_on_commit, twophase=False ) return self.session_factory() @@ -54,7 +61,7 @@ def setUp(self): self.__class__.__name__, self._testMethodName ) - self.base_url = 'http://static1.example.orm' + self.base_url = 'http://localhost:9000' # Remove previous files, if any! to make a clean temp directory: if exists(self.temp_path): # pragma: no cover diff --git a/sqlalchemy_media/tests/test_image.py b/sqlalchemy_media/tests/test_image.py index e67cb08..8281a9e 100644 --- a/sqlalchemy_media/tests/test_image.py +++ b/sqlalchemy_media/tests/test_image.py @@ -243,11 +243,11 @@ class Person(self.Base): person1 = session.query(Person).filter(Person.id == person1.id).one() with StoreManager(session): self.assertTrue(person1.image.locate().startswith( - 'http://static1.example.orm/images/image-' + 'http://localhost:9000/images/image-' )) thumbnail = person1.image.get_thumbnail(width=100) self.assertTrue(thumbnail.locate().startswith( - 'http://static1.example.orm/thumbnails/thumbnail-' + 'http://localhost:9000/thumbnails/thumbnail-' )) def test_image_list(self): diff --git a/sqlalchemy_media/tests/test_s3_store.py b/sqlalchemy_media/tests/test_s3_store.py index 2622f04..96e685b 100644 --- a/sqlalchemy_media/tests/test_s3_store.py +++ b/sqlalchemy_media/tests/test_s3_store.py @@ -7,30 +7,18 @@ from sqlalchemy_media.attachments import File, Image as BaseImage, \ Thumbnail as BaseThumbnail +from sqlalchemy_media.stores import StoreManager, S3Store from sqlalchemy_media.exceptions import S3Error -from sqlalchemy_media.stores import S3Store -from sqlalchemy_media.stores import StoreManager -from sqlalchemy_media.tests.helpers import Json, SqlAlchemyTestCase, \ - mockup_s3_server +from sqlalchemy_media.tests.helpers import ( + TEST_BUCKET, + Json, + S3TestCase, + SqlAlchemyTestCase, + create_s3_store, +) -TEST_BUCKET = 'test-bucket' -TEST_ACCESS_KEY = 'test_access_key' -TEST_SECRET_KEY = 'test_secret_key' -TEST_REGION = 'ap-northeast-2' - - -def create_s3_store(bucket=TEST_BUCKET, **kwargs): - return S3Store( - bucket, - TEST_ACCESS_KEY, - TEST_SECRET_KEY, - TEST_REGION, - **kwargs - ) - - -class S3StoreTestCase(SqlAlchemyTestCase): +class S3StoreTestCase(SqlAlchemyTestCase, S3TestCase): @classmethod def setUpClass(cls): @@ -38,358 +26,336 @@ def setUpClass(cls): cls.this_dir = abspath(dirname(__file__)) cls.stuff_path = join(cls.this_dir, 'stuff') cls.dog_jpeg = join(cls.stuff_path, 'dog.jpg') - cls.base_url = 'http://static1.example.orm' cls.sample_text_file1 = join(cls.stuff_path, 'sample_text_file1.txt') def test_put_from_stream(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - store = create_s3_store(base_url=uri) - target_filename = 'test_put_from_stream/file_from_stream1.txt' - content = b'Lorem ipsum dolor sit amet' - stream = io.BytesIO(content) - length = store.put(target_filename, stream) - self.assertEqual(length, len(content)) - self.assertIsInstance(store.open(target_filename), io.BytesIO) + target_filename = 'test_put_from_stream/file_from_stream1.txt' + content = b'Lorem ipsum dolor sit amet' + stream = io.BytesIO(content) + length = self.storage.put(target_filename, stream) + self.assertEqual(length, len(content)) + self.assertIsInstance(self.storage.open(target_filename), io.BytesIO) def test_put_error(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - store = create_s3_store(base_url=uri[:-2]) - target_filename = 'test_put_from_stream/file_from_stream1.txt' - content = b'Lorem ipsum dolor sit amet' - stream = io.BytesIO(content) + store = create_s3_store(base_url=self.base_url[:-2]) + target_filename = 'test_put_from_stream/file_from_stream1.txt' + content = b'Lorem ipsum dolor sit amet' + stream = io.BytesIO(content) - with self.assertRaises(S3Error): - store.put(target_filename, stream) + with self.assertRaises(S3Error): + store.put(target_filename, stream) def test_rrs_put(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - StoreManager.register( - 's3', - functools.partial(create_s3_store, base_url=uri), - default=True - ) + StoreManager.register( + 's3', + functools.partial(create_s3_store, base_url=self.base_url), + default=True + ) - class Thumbnail(BaseThumbnail): - __reproducible__ = True + class Thumbnail(BaseThumbnail): + __reproducible__ = True - class Image(BaseImage): - __thumbnail_type__ = Thumbnail + class Image(BaseImage): + __thumbnail_type__ = Thumbnail - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - image = Column(Image.as_mutable(Json)) + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + image = Column(Image.as_mutable(Json)) - session = self.create_all_and_get_session() + session = self.create_all_and_get_session() - person1 = Person() - self.assertIsNone(person1.image) + person1 = Person() + self.assertIsNone(person1.image) - with StoreManager(session): - person1 = Person() - person1.image = Image.create_from(self.dog_jpeg) - self.assertIsInstance(person1.image, Image) + with StoreManager(session): + person1 = Person() + person1.image = Image.create_from(self.dog_jpeg) + self.assertIsInstance(person1.image, Image) - thumbnail = person1.image.get_thumbnail( - width=100, - auto_generate=True - ) - self.assertIsInstance(thumbnail, Thumbnail) - self.assertTrue(thumbnail.reproducible, True) + thumbnail = person1.image.get_thumbnail( + width=100, + auto_generate=True + ) + self.assertIsInstance(thumbnail, Thumbnail) + self.assertTrue(thumbnail.reproducible, True) def test_delete(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - store = create_s3_store(base_url=uri) - target_filename = 'test_delete/sample_text_file1.txt' - with open(self.sample_text_file1, 'rb') as f: - length = store.put(target_filename, f) - self.assertEqual(length, getsize(self.sample_text_file1)) - self.assertIsInstance(store.open(target_filename), io.BytesIO) + target_filename = 'test_delete/sample_text_file1.txt' + with open(self.sample_text_file1, 'rb') as f: + length = self.storage.put(target_filename, f) + self.assertEqual(length, getsize(self.sample_text_file1)) + self.assertIsInstance(self.storage.open(target_filename), io.BytesIO) - store.delete(target_filename) + self.storage.delete(target_filename) - with self.assertRaises(S3Error): - store.open(target_filename) + with self.assertRaises(S3Error): + self.storage.open(target_filename) def test_delete_error(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - store = create_s3_store(base_url=uri) - wrong_store = create_s3_store(base_url=uri[:-2]) - target_filename = 'test_delete/sample_text_file1.txt' - with open(self.sample_text_file1, 'rb') as f: - length = store.put(target_filename, f) - self.assertEqual(length, getsize(self.sample_text_file1)) - self.assertIsInstance(store.open(target_filename), io.BytesIO) - - with self.assertRaises(S3Error): - wrong_store.delete(target_filename) + wrong_store = create_s3_store(base_url=self.base_url[:-2]) + target_filename = 'test_delete/sample_text_file1.txt' + with open(self.sample_text_file1, 'rb') as f: + length = self.storage.put(target_filename, f) + self.assertEqual(length, getsize(self.sample_text_file1)) + self.assertIsInstance(self.storage.open(target_filename), io.BytesIO) + + with self.assertRaises(S3Error): + wrong_store.delete(target_filename) def test_open(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - store = create_s3_store(base_url=uri) - target_filename = 'test_delete/sample_text_file1.txt' - with open(self.sample_text_file1, 'rb') as f: - length = store.put(target_filename, f) - self.assertEqual(length, getsize(self.sample_text_file1)) - self.assertIsInstance(store.open(target_filename), io.BytesIO) - - # Reading - with store.open(target_filename, mode='rb') as stored_file, \ - open(self.sample_text_file1, mode='rb') as original_file: - self.assertEqual(stored_file.read(), original_file.read()) + target_filename = 'test_delete/sample_text_file1.txt' + with open(self.sample_text_file1, 'rb') as f: + length = self.storage.put(target_filename, f) + self.assertEqual(length, getsize(self.sample_text_file1)) + self.assertIsInstance(self.storage.open(target_filename), io.BytesIO) + + # Reading + with self.storage.open(target_filename, mode='rb') as stored_file, \ + open(self.sample_text_file1, mode='rb') as original_file: + self.assertEqual(stored_file.read(), original_file.read()) def test_locate(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - StoreManager.register( - 's3', - functools.partial(create_s3_store, base_url=uri), - default=True - ) + StoreManager.register( + 's3', + functools.partial(create_s3_store, base_url=self.base_url), + default=True + ) + + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + file = Column(File.as_mutable(Json)) - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - file = Column(File.as_mutable(Json)) + session = self.create_all_and_get_session() - session = self.create_all_and_get_session() + person1 = Person() + self.assertIsNone(person1.file) + sample_content = b'Simple text.' + with StoreManager(session): person1 = Person() - self.assertIsNone(person1.file) - sample_content = b'Simple text.' - - with StoreManager(session): - person1 = Person() - person1.file = File.create_from(io.BytesIO(sample_content), - content_type='text/plain', - extension='.txt') - self.assertIsInstance(person1.file, File) - self.assertEqual( - person1.file.locate(), - '%s/%s?_ts=%s' % ( - uri, person1.file.path, person1.file.timestamp - ) + person1.file = File.create_from(io.BytesIO(sample_content), + content_type='text/plain', + extension='.txt') + self.assertIsInstance(person1.file, File) + self.assertEqual( + person1.file.locate(), + '%s/%s?_ts=%s' % ( + self.base_url, person1.file.path, person1.file.timestamp ) + ) def test_prefix(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - prefix = 'test' - StoreManager.register( - 's3', - functools.partial(create_s3_store, base_url=uri, prefix=prefix), - default=True - ) - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - file = Column(File.as_mutable(Json)) + prefix = 'test' + StoreManager.register( + 's3', + functools.partial( + create_s3_store, base_url=self.base_url, prefix=prefix), + default=True + ) - session = self.create_all_and_get_session() + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + file = Column(File.as_mutable(Json)) + session = self.create_all_and_get_session() + + person1 = Person() + self.assertIsNone(person1.file) + sample_content = b'Simple text.' + + with StoreManager(session): person1 = Person() - self.assertIsNone(person1.file) - sample_content = b'Simple text.' - - with StoreManager(session): - person1 = Person() - person1.file = File.create_from(io.BytesIO(sample_content), - content_type='text/plain', - extension='.txt') - self.assertIsInstance(person1.file, File) - self.assertEqual( - person1.file.locate(), - '%s/%s/%s?_ts=%s' % ( - uri, prefix, person1.file.path, person1.file.timestamp - ) + person1.file = File.create_from(io.BytesIO(sample_content), + content_type='text/plain', + extension='.txt') + self.assertIsInstance(person1.file, File) + self.assertEqual( + person1.file.locate(), + '%s/%s/%s?_ts=%s' % ( + self.base_url, prefix, person1.file.path, person1.file.timestamp ) + ) def test_default_base_url(self): - store = S3Store( - TEST_BUCKET, - TEST_ACCESS_KEY, - TEST_SECRET_KEY, - TEST_REGION - ) + store = create_s3_store() assert store.base_url == 'https://%s.s3.amazonaws.com' % TEST_BUCKET def test_public_base_url_strip(self): - with mockup_s3_server(TEST_BUCKET) as (server, uri): - base_url = '%s/' % uri - StoreManager.register( - 's3', - functools.partial(create_s3_store, base_url=base_url), - default=True - ) + StoreManager.register( + 's3', + functools.partial(create_s3_store, base_url=self.base_url), + default=True + ) + + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + file = Column(File.as_mutable(Json)) - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - file = Column(File.as_mutable(Json)) + session = self.create_all_and_get_session() - session = self.create_all_and_get_session() + person1 = Person() + self.assertIsNone(person1.file) + sample_content = b'Simple text.' + with StoreManager(session): person1 = Person() - self.assertIsNone(person1.file) - sample_content = b'Simple text.' - - with StoreManager(session): - person1 = Person() - person1.file = File.create_from(io.BytesIO(sample_content), - content_type='text/plain', - extension='.txt') - self.assertIsInstance(person1.file, File) - self.assertEqual(person1.file.locate(), '%s%s?_ts=%s' % ( - base_url, person1.file.path, person1.file.timestamp)) + person1.file = File.create_from(io.BytesIO(sample_content), + content_type='text/plain', + extension='.txt') + self.assertIsInstance(person1.file, File) + self.assertEqual(person1.file.locate(), '%s/%s?_ts=%s' % ( + self.base_url, person1.file.path, person1.file.timestamp)) def test_cdn_url(self): cdn_url = 'http//test.sqlalchemy-media.com' - with mockup_s3_server(TEST_BUCKET) as (server, uri): - StoreManager.register( - 's3', - functools.partial( - create_s3_store, - base_url=uri, - cdn_url=cdn_url - ), - default=True - ) + StoreManager.register( + 's3', + functools.partial( + create_s3_store, + base_url=self.base_url, + cdn_url=cdn_url + ), + default=True + ) + + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + file = Column(File.as_mutable(Json)) - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - file = Column(File.as_mutable(Json)) + session = self.create_all_and_get_session() - session = self.create_all_and_get_session() + person1 = Person() + self.assertIsNone(person1.file) + sample_content = b'Simple text.' + with StoreManager(session): person1 = Person() - self.assertIsNone(person1.file) - sample_content = b'Simple text.' - - with StoreManager(session): - person1 = Person() - person1.file = File.create_from( - io.BytesIO(sample_content), - content_type='text/plain', - extension='.txt' - ) - self.assertIsInstance(person1.file, File) - self.assertEqual(person1.file.locate(), '%s/%s?_ts=%s' % ( - cdn_url, person1.file.path, person1.file.timestamp - )) + person1.file = File.create_from( + io.BytesIO(sample_content), + content_type='text/plain', + extension='.txt' + ) + self.assertIsInstance(person1.file, File) + self.assertEqual(person1.file.locate(), '%s/%s?_ts=%s' % ( + cdn_url, person1.file.path, person1.file.timestamp + )) def test_cdn_url_strip(self): cdn_url = 'http//test.sqlalchemy-media.com/' - with mockup_s3_server(TEST_BUCKET) as (server, uri): - StoreManager.register( - 's3', - functools.partial( - create_s3_store, - base_url=uri, - cdn_url=cdn_url - ), - default=True - ) + StoreManager.register( + 's3', + functools.partial( + create_s3_store, + base_url=self.base_url, + cdn_url=cdn_url + ), + default=True + ) - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - file = Column(File.as_mutable(Json)) + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + file = Column(File.as_mutable(Json)) - session = self.create_all_and_get_session() + session = self.create_all_and_get_session() + person1 = Person() + self.assertIsNone(person1.file) + sample_content = b'Simple text.' + + with StoreManager(session): person1 = Person() - self.assertIsNone(person1.file) - sample_content = b'Simple text.' - - with StoreManager(session): - person1 = Person() - person1.file = File.create_from( - io.BytesIO(sample_content), - content_type='text/plain', - extension='.txt' - ) - self.assertIsInstance(person1.file, File) - self.assertEqual(person1.file.locate(), '%s%s?_ts=%s' % ( - cdn_url, person1.file.path, person1.file.timestamp)) + person1.file = File.create_from( + io.BytesIO(sample_content), + content_type='text/plain', + extension='.txt' + ) + self.assertIsInstance(person1.file, File) + self.assertEqual(person1.file.locate(), '%s%s?_ts=%s' % ( + cdn_url, person1.file.path, person1.file.timestamp)) def test_cdn_url_with_prefix(self): prefix = 'media' cdn_url = 'http//test.sqlalchemy-media.com' - with mockup_s3_server(TEST_BUCKET) as (server, uri): - StoreManager.register( - 's3', - functools.partial( - create_s3_store, - prefix=prefix, - base_url=uri, - cdn_url=cdn_url - ), - default=True - ) + StoreManager.register( + 's3', + functools.partial( + create_s3_store, + prefix=prefix, + base_url=self.base_url, + cdn_url=cdn_url + ), + default=True + ) + + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + file = Column(File.as_mutable(Json)) - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - file = Column(File.as_mutable(Json)) + session = self.create_all_and_get_session() - session = self.create_all_and_get_session() + person1 = Person() + self.assertIsNone(person1.file) + sample_content = b'Simple text.' + with StoreManager(session): person1 = Person() - self.assertIsNone(person1.file) - sample_content = b'Simple text.' - - with StoreManager(session): - person1 = Person() - person1.file = File.create_from(io.BytesIO(sample_content), - content_type='text/plain', - extension='.txt') - self.assertIsInstance(person1.file, File) - self.assertEqual(person1.file.locate(), '%s/%s/%s?_ts=%s' % ( - cdn_url, prefix, person1.file.path, person1.file.timestamp) - ) + person1.file = File.create_from(io.BytesIO(sample_content), + content_type='text/plain', + extension='.txt') + self.assertIsInstance(person1.file, File) + self.assertEqual(person1.file.locate(), '%s/%s/%s?_ts=%s' % ( + cdn_url, prefix, person1.file.path, person1.file.timestamp) + ) def test_cdn_url_with_ignore_prefix(self): prefix = 'media' cdn_url = 'http//test.sqlalchemy-media.com' - with mockup_s3_server(TEST_BUCKET) as (server, uri): - StoreManager.register( - 's3', - functools.partial( - create_s3_store, - prefix=prefix, - base_url=uri, - cdn_url=cdn_url, - cdn_prefix_ignore=True - ), - default=True - ) + StoreManager.register( + 's3', + functools.partial( + create_s3_store, + prefix=prefix, + base_url=self.base_url, + cdn_url=cdn_url, + cdn_prefix_ignore=True + ), + default=True + ) + + class Person(self.Base): + __tablename__ = 'person' + id = Column(Integer, primary_key=True) + file = Column(File.as_mutable(Json)) - class Person(self.Base): - __tablename__ = 'person' - id = Column(Integer, primary_key=True) - file = Column(File.as_mutable(Json)) + session = self.create_all_and_get_session() - session = self.create_all_and_get_session() + person1 = Person() + self.assertIsNone(person1.file) + sample_content = b'Simple text.' + with StoreManager(session): person1 = Person() - self.assertIsNone(person1.file) - sample_content = b'Simple text.' - - with StoreManager(session): - person1 = Person() - person1.file = File.create_from( - io.BytesIO(sample_content), - content_type='text/plain', - extension='.txt' - ) - self.assertIsInstance(person1.file, File) - self.assertEqual( - person1.file.locate(), '%s/%s?_ts=%s' % ( - cdn_url, - person1.file.path, - person1.file.timestamp - ) + person1.file = File.create_from( + io.BytesIO(sample_content), + content_type='text/plain', + extension='.txt' + ) + self.assertIsInstance(person1.file, File) + self.assertEqual( + person1.file.locate(), '%s/%s?_ts=%s' % ( + cdn_url, + person1.file.path, + person1.file.timestamp ) + ) if __name__ == '__main__': # pragma: no cover diff --git a/sqlalchemy_media/tests/test_ssh_store.py b/sqlalchemy_media/tests/test_ssh_store.py index cf4dc1d..191e72a 100644 --- a/sqlalchemy_media/tests/test_ssh_store.py +++ b/sqlalchemy_media/tests/test_ssh_store.py @@ -12,7 +12,7 @@ class SSHStoreTestCase(MockupSSHTestCase): def setUp(self): super().setUp() - self.base_url = 'http://static1.example.orm' + self.base_url = 'http://localhost:9000' self.stuff_path = join(self.here, 'stuff') self.sample_text_file1 = join(self.stuff_path, 'sample_text_file1.txt')