import sys, weakref
from sqlalchemy import create_engine, exceptions, select, MetaData, Table, Column, Integer, String
from testlib import *
-
+import time
class MockDisconnect(Exception):
pass
self.assertEquals(conn.execute(select([1])).scalar(), 1)
assert not conn.invalidated
+class RecycleTest(TestBase):
+ def test_basic(self):
+ for threadlocal in (False, True):
+ engine = engines.reconnecting_engine(options={'pool_recycle':1, 'pool_threadlocal':threadlocal})
+
+ conn = engine.connect()
+ self.assertEquals(conn.execute(select([1])).scalar(), 1)
+ conn.close()
+
+ engine.test_shutdown()
+ time.sleep(2)
+
+ conn = engine.connect()
+ self.assertEquals(conn.execute(select([1])).scalar(), 1)
+ conn.close()
+
class InvalidateDuringResultTest(TestBase):
def setUp(self):
global meta, table, engine
def reconnecting_engine(url=None, options=None):
url = url or config.db_url
dbapi = config.db.dialect.dbapi
- engine = testing_engine(url, {'module':ReconnectFixture(dbapi)})
+ if not options:
+ options = {}
+ options['module'] = ReconnectFixture(dbapi)
+ engine = testing_engine(url, options)
engine.test_shutdown = engine.dialect.dbapi.shutdown
return engine