from sqlalchemy import create_engine, exceptions, select, MetaData, Table, Column, Integer, String
from testlib import *
import time
+import gc
class MockDisconnect(Exception):
pass
assert id(db.pool) != pid
# ensure all connections closed (pool was recycled)
+ gc.collect()
assert len(dbapi.connections) == 0
conn =db.connect()
pass
# assert was invalidated
+ gc.collect()
assert len(dbapi.connections) == 0
assert not conn.closed
assert conn.invalidated
assert conn.invalidated
# ensure all connections closed (pool was recycled)
+ gc.collect()
assert len(dbapi.connections) == 0
# test reconnects