)
metadata.create_all()
+ @engines.close_first
def tearDown(self):
testing.db.connect().execute(users.delete())
+
def tearDownAll(self):
metadata.drop_all()
class ExecuteTest(TestBase):
-
+ @engines.close_first
+ def tearDown(self):
+ pass
+
def test_standalone_execute(self):
x = testing.db.func.current_date().execute().scalar()
y = testing.db.func.current_date().select().execute().scalar()
conn.close()
assert (x == y == z) is True
+ @engines.close_first
def test_update(self):
"""
Tests sending functions and SQL expressions to the VALUES and SET
maxlen = testing.db.dialect.max_identifier_length
testing.db.dialect.max_identifier_length = IDENT_LENGTH
+ @engines.close_first
def tearDown(self):
table1.delete().execute()
)
metadata.create_all()
+ @engines.close_first
def tearDown(self):
addresses.delete().execute()
users.delete().execute()
dict(col2="t3col2r2", col3="bbb", col4="aaa"),
dict(col2="t3col2r3", col3="ccc", col4="bbb"),
])
-
+
+ @engines.close_first
+ def tearDown(self):
+ pass
+
def tearDownAll(self):
metadata.drop_all()
def tearDownAll(self):
metadata.drop_all()
+ @engines.close_first
def tearDown(self):
unicode_table.delete().execute()
)
binary_table.create()
+ @engines.close_first
def tearDown(self):
binary_table.delete().execute()
def tearDownAll(self):
metadata.drop_all()
+ @engines.close_first
def tearDown(self):
numeric_table.delete().execute()
)
metadata.create_all()
+ @engines.close_first
def tearDown(self):
interval_table.delete().execute()
)
metadata.create_all()
+ @engines.close_first
def tearDown(self):
if metadata.tables:
t3.delete().execute()
testing_reaper = ConnectionKiller()
def drop_all_tables(metadata):
- testing_reaper.rollback_all()
+ testing_reaper.close_all()
metadata.drop_all()
def assert_conns_closed(fn):
testing_reaper.rollback_all()
return _function_named(decorated, fn.__name__)
+def close_first(fn):
+ """Decorator that closes all connections before fn execution."""
+ def decorated(*args, **kw):
+ testing_reaper.close_all()
+ fn(*args, **kw)
+ return _function_named(decorated, fn.__name__)
+
+
def close_open_connections(fn):
"""Decorator that closes all connections after fn execution."""