Skip to content

Commit

Permalink
flush/noflush tests for sqlalchemy
Browse files Browse the repository at this point in the history
  • Loading branch information
@amol-
amol- committed Mar 15, 2017
1 parent c144c1b commit 0963497
Showing 1 changed file with 62 additions and 110 deletions.
172 changes: 62 additions & 110 deletions tests/test_fields_sqlalchemy.py
View file
Expand Up @@ -53,7 +53,22 @@ class Confidential(Document):
__mapper_args__ = {'polymorphic_identity': 'confidential'}


class TestSQLAAttachments(object):
class SQLATestCase(object):
FLUSH_SESSION = True

def _session_flush(self):
if self.FLUSH_SESSION:
DBSession.flush()

def file_exists(self, file_id):
try:
get_file(file_id)
return True
except IOError:
return False


class TestSQLAAttachments(SQLATestCase):
def __init__(self):
self.file_content = b'this is the file content'
self.fake_file = tempfile.NamedTemporaryFile()
Expand All @@ -67,8 +82,9 @@ def test_create_fromfile(self):
doc = Document(name=u_('Foo'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()
DBSession.remove()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
assert d.content.file.read() == self.file_content
Expand All @@ -78,8 +94,9 @@ def test_create_polymorphic_from_file(self):
doc = Confidential(name=u_('Secret'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()
DBSession.remove()

d = DBSession.query(Confidential).filter_by(name=u_('Secret')).first()
assert d.content.file.read() == self.file_content
Expand All @@ -89,33 +106,7 @@ def test_edit_existing(self):
doc = Document(name=u_('Foo2'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
DBSession.commit()
DBSession.remove()

d = DBSession.query(Document).filter_by(name=u_('Foo2')).first()
old_file = d.content.path

d.content = b'HELLO'
new_file = d.content.path

DBSession.flush()
DBSession.commit()
DBSession.remove()

assert get_file(new_file).read() == b'HELLO'

try:
fold = get_file(old_file)
assert False, 'Should have raised IOError here'
except IOError:
pass

def test_edit_existing_noflush(self):
doc = Document(name=u_('Foo2'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()
DBSession.remove()

Expand All @@ -125,6 +116,7 @@ def test_edit_existing_noflush(self):
d.content = b'HELLO'
new_file = d.content.path

self._session_flush()
DBSession.commit()
DBSession.remove()

Expand All @@ -140,33 +132,7 @@ def test_edit_existing_rollback(self):
doc = Document(name=u_('Foo3'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
DBSession.commit()
DBSession.remove()

d = DBSession.query(Document).filter_by(name=u_('Foo3')).first()
old_file = d.content.path

d.content = b'HELLO'
new_file = d.content.path

DBSession.flush()
DBSession.rollback()
DBSession.remove()

assert get_file(old_file).read() == self.file_content

try:
fold = get_file(new_file)
assert False, 'Should have raised IOError here'
except IOError:
pass

def test_edit_existing_rollback_noflush(self):
doc = Document(name=u_('Foo3'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()
DBSession.remove()

Expand All @@ -176,23 +142,19 @@ def test_edit_existing_rollback_noflush(self):
d.content = b'HELLO'
new_file = d.content.path

self._session_flush()
DBSession.rollback()
DBSession.remove()

assert get_file(old_file).read() == self.file_content

try:
fold = get_file(new_file)
assert False, 'Should have raised IOError here'
except IOError:
pass
assert not self.file_exists(new_file)

def test_create_fromfield(self):
field = create_cgifs('image/jpeg', self.fake_file, 'test.jpg')

doc = Document(name=u_('Foo'), content=field)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -204,7 +166,7 @@ def test_create_fromfield(self):
def test_create_empty(self):
doc = Document(name=u_('Foo'), content=None)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -214,37 +176,33 @@ def test_delete_existing(self):
doc = Document(name=u_('Foo2'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()
DBSession.remove()

d = DBSession.query(Document).filter_by(name=u_('Foo2')).first()
old_file = d.content.path
DBSession.delete(d)

DBSession.flush()
self._session_flush()
DBSession.commit()
DBSession.remove()

try:
fold = get_file(old_file)
assert False, 'Should have raised IOError here'
except IOError:
pass
assert not self.file_exists(old_file)

def test_delete_existing_rollback(self):
doc = Document(name=u_('Foo3'))
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()
DBSession.remove()

d = DBSession.query(Document).filter_by(name=u_('Foo3')).first()
old_file = d.content.path
DBSession.delete(d)

DBSession.flush()
self._session_flush()
DBSession.rollback()
DBSession.remove()

Expand All @@ -254,7 +212,7 @@ def test_create_with_alias(self):
doc = Document(name=u_('Foo'))
doc.targeted_content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -263,7 +221,11 @@ def test_create_with_alias(self):
assert d.targeted_content.depot_name == 'another'


class TestSQLAImageAttachments(object):
class TestSQLAAttachmentsNoFlush(TestSQLAAttachments):
FLUSH_SESSION = False


class TestSQLAImageAttachments(SQLATestCase):
def __init__(self):
self.file_content = b'''R0lGODlhEQAUAPcAAC4uLjAwMDIyMjMzMjQ0NDU1NDY2Njk2Mzg4ODo6Oj49Ozw8PD4+PkE+OEA/PkhAN0tCNk5JPFFGNV1KMFhNNVFHOFJJPVVLPVhOPXZfKXVcK2ZQNGZXNGtZMnNcNHZeNnldMHJfOn1hKXVjOH9oO0BAQEJCQkREREVFREZGRklGQ05KQ0hISEpKSkxMTE5OTlZRSlFQT19XSFBQUFJSUlRUVGFUQmFVQ2ZZQGtdQnNiQqJ/HI1uIYBnLIllKoZrK4FqLoVqL4luLIpsLpt7J515JJ50KZhzLYFnMIFlM4ZlMIFkNI1uNoJoOoVrPIlvO49yMolwPpB2O5p4Op98PaB3IKN4JqN8J6h7I6J5LaZ+LLF+ILGGG7+JG72OGLKEI7aHIrOEJL2JI7mMN76YNcGJG8SOG8WLHMONH86eEs+aFsGSG8eQHMySG9uVFduXFdeeE9eaFdScF96YE9yaFOKcEuOdEtWgFNiiEduhE96pEuqlD+qmD+KpD+yoDu6rDuysDvCuDfCvDeuwD/SzC/a2CvGwDfKxDPi5Cfi5Cvq8CeCjEuehEOagEeijEOKoEOStFMOOK8+TLM6YNNChItGgLtylKt6gMNqgON6jPOChLfi/JOSrNeGvN9KhRtykRNWkSOCnQOCpSOawQue1T+a6Su67SOGsUO/AVAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACH5BAAAAAAALAAAAAARABQAAAj+AAEIHEiwoMAACBMqXIhQgMOHDwdAfEigYsUCT6KQScNjxwGLBAyINPBhCilUmxIV6uMlw0gEMJeMGWWqFCU8hA4JEgETQYIEDcRw6qRlwgMIGvJwkfAzwYIFXQBB8qHg6VMKFawuYNBBjaIqDhhI+cGgrNmyJXooGrShBJBKcIpwiFCibl0TehDdsWBCiBxLZuIwolMGiwcTJ4gUenThBAokSVRgGFJnzhYQJ1KsyRkkhWfPK87McWPEM4sRhgItCsGitQ5PmtxYUdK6BY4rf/zwYRNmUihRpyQdaUHchQsoX/Y4amTnUqZPoG7YMO7ihfUcYNC0eRMJExMqMKweW59BfkYMEk2ykHAio3x5GvDjy58Pv4b9+/jz2w8IADs='''
self.file_content = base64.b64decode(self.file_content)
Expand All @@ -284,7 +246,7 @@ def test_create_fromfile(self):
doc = Document(name=u_('Foo'))
doc.photo = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -311,7 +273,7 @@ def test_create_fromfield(self):

doc = Document(name=u_('Foo'), photo=field)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -326,7 +288,7 @@ def test_thumbnail(self):

doc = Document(name=u_('Foo'), photo=field)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -341,7 +303,7 @@ def test_maximum_size(self):

doc = Document(name=u_('Foo'), photo=field)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -358,7 +320,7 @@ def test_public_url(self):
doc.photo = open(self.fake_file.name, 'rb')
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -376,28 +338,23 @@ def test_rollback(self):
doc = Document(name=u_('Foo3'))
doc.photo = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()

uploaded_file = doc.photo.path
uploaded_thumb = doc.photo.thumb_path

DBSession.rollback()
DBSession.remove()

try:
fold = get_file(uploaded_file)
assert False, 'Should have raised IOError here'
except IOError:
pass
assert not self.file_exists(uploaded_file)
assert not self.file_exists(uploaded_thumb)

try:
fold = get_file(uploaded_thumb)
assert False, 'Should have raised IOError here'
except IOError:
pass

class TestSQLAImageAttachmentsNoFlush(TestSQLAImageAttachments):
FLUSH_SESSION = False

class TestSQLAThumbnailFilter(object):

class TestSQLAThumbnailFilter(SQLATestCase):
def __init__(self):
self.file_content = b'''R0lGODlhEQAUAPcAAC4uLjAwMDIyMjMzMjQ0NDU1NDY2Njk2Mzg4ODo6Oj49Ozw8PD4+PkE+OEA/PkhAN0tCNk5JPFFGNV1KMFhNNVFHOFJJPVVLPVhOPXZfKXVcK2ZQNGZXNGtZMnNcNHZeNnldMHJfOn1hKXVjOH9oO0BAQEJCQkREREVFREZGRklGQ05KQ0hISEpKSkxMTE5OTlZRSlFQT19XSFBQUFJSUlRUVGFUQmFVQ2ZZQGtdQnNiQqJ/HI1uIYBnLIllKoZrK4FqLoVqL4luLIpsLpt7J515JJ50KZhzLYFnMIFlM4ZlMIFkNI1uNoJoOoVrPIlvO49yMolwPpB2O5p4Op98PaB3IKN4JqN8J6h7I6J5LaZ+LLF+ILGGG7+JG72OGLKEI7aHIrOEJL2JI7mMN76YNcGJG8SOG8WLHMONH86eEs+aFsGSG8eQHMySG9uVFduXFdeeE9eaFdScF96YE9yaFOKcEuOdEtWgFNiiEduhE96pEuqlD+qmD+KpD+yoDu6rDuysDvCuDfCvDeuwD/SzC/a2CvGwDfKxDPi5Cfi5Cvq8CeCjEuehEOagEeijEOKoEOStFMOOK8+TLM6YNNChItGgLtylKt6gMNqgON6jPOChLfi/JOSrNeGvN9KhRtykRNWkSOCnQOCpSOawQue1T+a6Su67SOGsUO/AVAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACH5BAAAAAAALAAAAAARABQAAAj+AAEIHEiwoMAACBMqXIhQgMOHDwdAfEigYsUCT6KQScNjxwGLBAyINPBhCilUmxIV6uMlw0gEMJeMGWWqFCU8hA4JEgETQYIEDcRw6qRlwgMIGvJwkfAzwYIFXQBB8qHg6VMKFawuYNBBjaIqDhhI+cGgrNmyJXooGrShBJBKcIpwiFCibl0TehDdsWBCiBxLZuIwolMGiwcTJ4gUenThBAokSVRgGFJnzhYQJ1KsyRkkhWfPK87McWPEM4sRhgItCsGitQ5PmtxYUdK6BY4rf/zwYRNmUihRpyQdaUHchQsoX/Y4amTnUqZPoG7YMO7ihfUcYNC0eRMJExMqMKweW59BfkYMEk2ykHAio3x5GvDjy58Pv4b9+/jz2w8IADs='''
self.file_content = base64.b64decode(self.file_content)
Expand All @@ -413,7 +370,7 @@ def test_create_fromfile(self):
doc = Document(name=u_('Foo'))
doc.second_photo = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -424,7 +381,7 @@ def test_create_from_bytes(self):
doc = Document(name=u_('Foo'))
doc.second_photo = self.file_content
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -438,7 +395,7 @@ def test_create_fromfield(self):

doc = Document(name=u_('Foo'), second_photo=field)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -453,7 +410,7 @@ def test_create_fileintent(self):

doc = Document(name=u_('Foo'), second_photo=field)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -469,7 +426,7 @@ def test_thumbnail(self):

doc = Document(name=u_('Foo'), second_photo=field)
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -487,7 +444,7 @@ def test_public_url(self):
doc.second_photo = open(self.fake_file.name, 'rb')
doc.content = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()
DBSession.commit()

d = DBSession.query(Document).filter_by(name=u_('Foo')).first()
Expand All @@ -504,22 +461,17 @@ def test_rollback(self):
doc = Document(name=u_('Foo3'))
doc.second_photo = open(self.fake_file.name, 'rb')
DBSession.add(doc)
DBSession.flush()
self._session_flush()

uploaded_file = doc.second_photo.path
uploaded_thumb = doc.second_photo.thumb_12x12_path

DBSession.rollback()
DBSession.remove()

try:
fold = get_file(uploaded_file)
assert False, 'Should have raised IOError here'
except IOError:
pass
assert not self.file_exists(uploaded_file)
assert not self.file_exists(uploaded_thumb)

try:
fold = get_file(uploaded_thumb)
assert False, 'Should have raised IOError here'
except IOError:
pass

class TestSQLAThumbnailFilterNoFlush(TestSQLAThumbnailFilter):
FLUSH_SESSION = False

0 comments on commit 0963497

Please sign in to comment.