]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
added a "live" test of recycle to further ensure #1133 is not valid
authorMike Bayer <mike_mp@zzzcomputing.com>
Sun, 10 Aug 2008 05:15:11 +0000 (05:15 +0000)
committerMike Bayer <mike_mp@zzzcomputing.com>
Sun, 10 Aug 2008 05:15:11 +0000 (05:15 +0000)
test/engine/reconnect.py
test/testlib/engines.py

index b774ed7ae690d5e1ac86a1b40e0baa7085439aa2..7cdd1a799ae8ef008b2ab04c63da5cd5b50cb833 100644 (file)
@@ -2,7 +2,7 @@ import testenv; testenv.configure_for_tests()
 import sys, weakref
 from sqlalchemy import create_engine, exceptions, select, MetaData, Table, Column, Integer, String
 from testlib import *
-
+import time
 
 class MockDisconnect(Exception):
     pass
@@ -275,6 +275,22 @@ class RealReconnectTest(TestBase):
         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
index f5694df57e55c92fc1630ad7861f49f6df4e4bab..e0d3488bd7c52000f4a16372a420ee0478133b00 100644 (file)
@@ -87,7 +87,10 @@ class ReconnectFixture(object):
 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