diff options
Diffstat (limited to 'oslo_db/tests')
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_models.py | 50 |
1 files changed, 50 insertions, 0 deletions
diff --git a/oslo_db/tests/sqlalchemy/test_models.py b/oslo_db/tests/sqlalchemy/test_models.py index 60e8c55..fc12b0b 100644 --- a/oslo_db/tests/sqlalchemy/test_models.py +++ b/oslo_db/tests/sqlalchemy/test_models.py @@ -14,10 +14,13 @@ # under the License. import collections +import datetime +import mock from oslotest import base as oslo_test from sqlalchemy import Column from sqlalchemy import Integer, String +from sqlalchemy import event from sqlalchemy.ext.declarative import declarative_base from oslo_db.sqlalchemy import models @@ -179,3 +182,50 @@ class TimestampMixinTest(oslo_test.BaseTestCase): for method in methods: self.assertTrue(hasattr(models.TimestampMixin, method), "Method %s() is not found" % method) + + +class SoftDeletedModel(BASE, models.ModelBase, models.SoftDeleteMixin): + __tablename__ = 'test_model_soft_deletes' + + id = Column('id', Integer, primary_key=True) + smth = Column('smth', String(255)) + + +class SoftDeleteMixinTest(test_base.DbTestCase): + def setUp(self): + super(SoftDeleteMixinTest, self).setUp() + + t = BASE.metadata.tables['test_model_soft_deletes'] + t.create(self.engine) + self.addCleanup(t.drop, self.engine) + + self.session = self.sessionmaker(autocommit=False) + self.addCleanup(self.session.close) + + @mock.patch('oslo_utils.timeutils.utcnow') + def test_soft_delete(self, mock_utcnow): + dt = datetime.datetime.utcnow().replace(microsecond=0) + mock_utcnow.return_value = dt + + m = SoftDeletedModel(id=123456, smth='test') + self.session.add(m) + self.session.commit() + self.assertEqual(0, m.deleted) + self.assertIs(None, m.deleted_at) + + m.soft_delete(self.session) + self.assertEqual(123456, m.deleted) + self.assertIs(dt, m.deleted_at) + + def test_soft_delete_coerce_deleted_to_integer(self): + def listener(conn, cur, stmt, params, context, executemany): + if 'insert' in stmt.lower(): # ignore SELECT 1 and BEGIN + self.assertNotIn('False', str(params)) + + event.listen(self.engine, 'before_cursor_execute', listener) + self.addCleanup(event.remove, + self.engine, 'before_cursor_execute', listener) + + m = SoftDeletedModel(id=1, smth='test', deleted=False) + self.session.add(m) + self.session.commit() |