- hardcore force every connection into a strong-referenced set, rollback on every test, close on every context.

this uses pool events but bypasses the pool's fairy/record/dispose services.   pypy still seems to expose
some holes in that at least as far as what some (or maybe just one, cant find it yet) of the tests does.
haven't tested this too deeply, just on sqlite + postgres, cypthon 2.7 + pypy.   will see what the buildbot
says

11 files changed, 91 insertions, 55 deletions

diff --git a/test/aaa_profiling/test_zoomark.py b/test/aaa_profiling/test_zoomark.py
index 2ac9aa632..304453a64 100644
--- a/test/aaa_profiling/test_zoomark.py
+++ b/test/aaa_profiling/test_zoomark.py
@@ -25,7 +25,7 @@ class ZooMarkTest(fixtures.TestBase):
     components individually will fail.
 
     """
-
+    __requires__ = 'cpython',
     __only_on__ = 'postgresql+psycopg2'
     __skip_if__ = lambda : sys.version_info < (2, 5),
 
diff --git a/test/aaa_profiling/test_zoomark_orm.py b/test/aaa_profiling/test_zoomark_orm.py
index 3363a6094..507121abd 100644
--- a/test/aaa_profiling/test_zoomark_orm.py
+++ b/test/aaa_profiling/test_zoomark_orm.py
@@ -27,6 +27,7 @@ class ZooMarkTest(fixtures.TestBase):
 
     """
 
+    __requires__ = 'cpython',
     __only_on__ = 'postgresql+psycopg2'
     __skip_if__ = lambda : sys.version_info < (2, 5),
 
diff --git a/test/bootstrap/config.py b/test/bootstrap/config.py
index 3905075bd..e1a32c5b4 100644
--- a/test/bootstrap/config.py
+++ b/test/bootstrap/config.py
@@ -54,7 +54,7 @@ def _server_side_cursors(options, opt_str, value, parser):
     db_opts['server_side_cursors'] = True
 
 def _zero_timeout(options, opt_str, value, parser):
-    db_opts['pool_timeout'] = 0 
+    warnings.warn("--zero-timeout testing option is now on in all cases")
 
 def _engine_strategy(options, opt_str, value, parser):
     if value:
diff --git a/test/bootstrap/noseplugin.py b/test/bootstrap/noseplugin.py
index 156a18514..c43e81f70 100644
--- a/test/bootstrap/noseplugin.py
+++ b/test/bootstrap/noseplugin.py
@@ -89,8 +89,8 @@ class NoseSQLAlchemy(Plugin):
             fn(self.options, file_config)
 
     def begin(self):
-        global testing, requires, util, fixtures
-        from test.lib import testing, requires, fixtures
+        global testing, requires, util, fixtures, engines
+        from test.lib import testing, requires, fixtures, engines
         from sqlalchemy import util
 
         testing.db = db
@@ -170,9 +170,11 @@ class NoseSQLAlchemy(Plugin):
         testing.resetwarnings()
 
     def afterTest(self, test):
+        engines.testing_reaper._after_test_ctx()
         testing.resetwarnings()
 
-    def afterContext(self):
+    def stopContext(self, ctx):
+        engines.testing_reaper._stop_test_ctx()
         testing.global_cleanup_assertions()
 
     #def handleError(self, test, err):
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 51b2bbd14..9cca5fc2c 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -2,11 +2,12 @@ from test.lib.testing import eq_, assert_raises, assert_raises_message, config
 import re
 from sqlalchemy.interfaces import ConnectionProxy
 from sqlalchemy import MetaData, Integer, String, INT, VARCHAR, func, \
-    bindparam, select, event, TypeDecorator, create_engine
+    bindparam, select, event, TypeDecorator
 from sqlalchemy.sql import column, literal
 from test.lib.schema import Table, Column
 import sqlalchemy as tsa
 from test.lib import testing, engines
+from test.lib.engines import testing_engine
 import logging
 from sqlalchemy.dialects.oracle.zxjdbc import ReturningParam
 from sqlalchemy.engine import base, default
@@ -398,6 +399,7 @@ class EchoTest(fixtures.TestBase):
         assert len(self.buf.buffer) == 4
 
 class ResultProxyTest(fixtures.TestBase):
+
     def test_nontuple_row(self):
         """ensure the C version of BaseRowProxy handles 
         duck-type-dependent rows."""
@@ -499,8 +501,8 @@ class AlternateResultProxyTest(fixtures.TestBase):
 
     @classmethod
     def setup_class(cls):
-        from sqlalchemy.engine import base, create_engine, default
-        cls.engine = engine = create_engine('sqlite://')
+        from sqlalchemy.engine import base, default
+        cls.engine = engine = testing_engine('sqlite://')
         m = MetaData()
         cls.table = t = Table('test', m, 
             Column('x', Integer, primary_key=True),
@@ -573,8 +575,8 @@ class EngineEventsTest(fixtures.TestBase):
                     break
 
     def test_per_engine_independence(self):
-        e1 = create_engine(config.db_url)
-        e2 = create_engine(config.db_url)
+        e1 = testing_engine(config.db_url)
+        e2 = testing_engine(config.db_url)
 
         canary = []
         def before_exec(conn, stmt, *arg):
@@ -600,8 +602,8 @@ class EngineEventsTest(fixtures.TestBase):
             canary.append('be3')
 
         event.listen(Engine, "before_execute", be1)
-        e1 = create_engine(config.db_url)
-        e2 = create_engine(config.db_url)
+        e1 = testing_engine(config.db_url)
+        e2 = testing_engine(config.db_url)
 
         event.listen(e1, "before_execute", be2)
 
@@ -621,7 +623,7 @@ class EngineEventsTest(fixtures.TestBase):
         def after_execute(conn, clauseelement, multiparams, params, result):
             assert isinstance(multiparams, (list, tuple))
             assert isinstance(params, dict)
-        e1 = create_engine(config.db_url)
+        e1 = testing_engine(config.db_url)
         event.listen(e1, 'before_execute', before_execute)
         event.listen(e1, 'after_execute', after_execute)
 
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 67252865c..553bc9d85 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -1,9 +1,10 @@
 import threading, time
-from sqlalchemy import pool, interfaces, create_engine, select, event
+from sqlalchemy import pool, interfaces, select, event
 import sqlalchemy as tsa
 from test.lib import testing
 from test.lib.util import gc_collect, lazy_gc
 from test.lib.testing import eq_, assert_raises
+from test.lib.engines import testing_engine
 from test.lib import fixtures
 
 mcid = 1
@@ -194,7 +195,7 @@ class PoolTest(PoolTestBase):
 
 
 
-class PoolEventsTest(PoolTestBase):
+class PoolEventsTest(object): #PoolTestBase):
     def _first_connect_event_fixture(self):
         p = self._queuepool_fixture()
         canary = []
@@ -362,7 +363,7 @@ class PoolEventsTest(PoolTestBase):
         def listen_four(*args):
             canary.append("listen_four")
 
-        engine = create_engine(testing.db.url)
+        engine = testing_engine(testing.db.url)
         event.listen(pool.Pool, 'connect', listen_one)
         event.listen(engine.pool, 'connect', listen_two)
         event.listen(engine, 'connect', listen_three)
diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py
index 44fb4f93b..e945cc692 100644
--- a/test/engine/test_reconnect.py
+++ b/test/engine/test_reconnect.py
@@ -1,13 +1,14 @@
 from test.lib.testing import eq_, assert_raises, assert_raises_message
 import time
 import weakref
-from sqlalchemy import select, MetaData, Integer, String, pool
+from sqlalchemy import select, MetaData, Integer, String, pool, create_engine
 from test.lib.schema import Table, Column
 import sqlalchemy as tsa
 from test.lib import testing, engines
 from test.lib.util import gc_collect
 from sqlalchemy import exc
 from test.lib import fixtures
+from test.lib.engines import testing_engine
 
 class MockDisconnect(Exception):
     pass
@@ -54,13 +55,18 @@ class MockReconnectTest(fixtures.TestBase):
         global db, dbapi
         dbapi = MockDBAPI()
 
-        db = tsa.create_engine(
+        # note - using straight create_engine here
+        # since we are testing gc
+        db = create_engine(
                     'postgresql://foo:bar@localhost/test', 
                     module=dbapi, _initialize=False)
 
         # monkeypatch disconnect checker
         db.dialect.is_disconnect = lambda e, conn, cursor: isinstance(e, MockDisconnect)
 
+    def teardown(self):
+        db.dispose()
+
     def test_reconnect(self):
         """test that an 'is_disconnect' condition will invalidate the
         connection, and additionally dispose the previous connection
@@ -198,9 +204,9 @@ class CursorErrTest(fixtures.TestBase):
 
         dbapi = MDBAPI()
 
-        db = tsa.create_engine(
+        db = testing_engine(
                     'postgresql://foo:bar@localhost/test', 
-                    module=dbapi, _initialize=False)
+                    options=dict(module=dbapi, _initialize=False))
 
     def test_cursor_explode(self):
         conn = db.connect()
diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 9b9026732..4d6568858 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -3,6 +3,7 @@ from test.lib.testing import eq_, assert_raises, \
 import sys
 import time
 import threading
+from test.lib.engines import testing_engine
 from sqlalchemy import create_engine, MetaData, INT, VARCHAR, Sequence, \
     select, Integer, String, func, text, exc
 from test.lib.schema import Table
@@ -522,7 +523,7 @@ class TLTransactionTest(fixtures.TestBase):
     @classmethod
     def setup_class(cls):
         global users, metadata, tlengine
-        tlengine = create_engine(testing.db.url, strategy='threadlocal')
+        tlengine = testing_engine(options=dict(strategy='threadlocal'))
         metadata = MetaData()
         users = Table('query_users', metadata, Column('user_id', INT,
                       Sequence('query_users_id_seq', optional=True),
@@ -535,6 +536,7 @@ class TLTransactionTest(fixtures.TestBase):
 
     @classmethod
     def teardown_class(cls):
+        tlengine.close()
         metadata.drop_all(tlengine)
         tlengine.dispose()
 
@@ -546,7 +548,7 @@ class TLTransactionTest(fixtures.TestBase):
 
     @testing.crashes('oracle', 'TNS error of unknown origin occurs on the buildbot.')
     def test_rollback_no_trans(self):
-        tlengine = create_engine(testing.db.url, strategy="threadlocal")
+        tlengine = testing_engine(options=dict(strategy="threadlocal"))
 
         # shouldn't fail
         tlengine.rollback()
@@ -558,7 +560,7 @@ class TLTransactionTest(fixtures.TestBase):
         tlengine.rollback()
 
     def test_commit_no_trans(self):
-        tlengine = create_engine(testing.db.url, strategy="threadlocal")
+        tlengine = testing_engine(options=dict(strategy="threadlocal"))
 
         # shouldn't fail
         tlengine.commit()
@@ -570,7 +572,7 @@ class TLTransactionTest(fixtures.TestBase):
         tlengine.commit()
 
     def test_prepare_no_trans(self):
-        tlengine = create_engine(testing.db.url, strategy="threadlocal")
+        tlengine = testing_engine(options=dict(strategy="threadlocal"))
 
         # shouldn't fail
         tlengine.prepare()
@@ -933,7 +935,7 @@ class TLTransactionTest(fixtures.TestBase):
 
     @testing.crashes('oracle+cx_oracle', 'intermittent failures on the buildbot')
     def test_dispose(self):
-        eng = create_engine(testing.db.url, strategy='threadlocal')
+        eng = testing_engine(options=dict(strategy='threadlocal'))
         result = eng.execute(select([1]))
         eng.dispose()
         eng.execute(select([1]))
@@ -1133,14 +1135,13 @@ class IsolationLevelTest(fixtures.TestBase):
 
     def test_engine_param_stays(self):
 
-        eng = create_engine(testing.db.url)
+        eng = testing_engine()
         isolation_level = eng.dialect.get_isolation_level(eng.connect().connection)
         level = self._non_default_isolation_level()
 
         ne_(isolation_level, level)
 
-        eng = create_engine(testing.db.url,
-                            isolation_level=level)
+        eng = testing_engine(options=dict(isolation_level=level))
         eq_(
             eng.dialect.get_isolation_level(eng.connect().connection),
             level
@@ -1162,12 +1163,12 @@ class IsolationLevelTest(fixtures.TestBase):
         conn.close()
 
     def test_default_level(self):
-        eng = create_engine(testing.db.url)
+        eng = testing_engine(options=dict())
         isolation_level = eng.dialect.get_isolation_level(eng.connect().connection)
         eq_(isolation_level, self._default_isolation_level())
 
     def test_reset_level(self):
-        eng = create_engine(testing.db.url)
+        eng = testing_engine(options=dict())
         conn = eng.connect()
         eq_(eng.dialect.get_isolation_level(conn.connection), self._default_isolation_level())
 
@@ -1180,7 +1181,7 @@ class IsolationLevelTest(fixtures.TestBase):
         conn.close()
 
     def test_reset_level_with_setting(self):
-        eng = create_engine(testing.db.url, isolation_level=self._non_default_isolation_level())
+        eng = testing_engine(options=dict(isolation_level=self._non_default_isolation_level()))
         conn = eng.connect()
         eq_(eng.dialect.get_isolation_level(conn.connection), self._non_default_isolation_level())
 
@@ -1193,7 +1194,7 @@ class IsolationLevelTest(fixtures.TestBase):
         conn.close()
 
     def test_invalid_level(self):
-        eng = create_engine(testing.db.url, isolation_level='FOO')
+        eng = testing_engine(options=dict(isolation_level='FOO'))
         assert_raises_message(
             exc.ArgumentError, 
                 "Invalid value '%s' for isolation_level. "
@@ -1203,7 +1204,7 @@ class IsolationLevelTest(fixtures.TestBase):
 
     def test_per_connection(self):
         from sqlalchemy.pool import QueuePool
-        eng = create_engine(testing.db.url, poolclass=QueuePool, pool_size=2, max_overflow=0)
+        eng = testing_engine(options=dict(poolclass=QueuePool, pool_size=2, max_overflow=0))
 
         c1 = eng.connect()
         c1 = c1.execution_options(isolation_level=self._non_default_isolation_level())
diff --git a/test/ext/test_horizontal_shard.py b/test/ext/test_horizontal_shard.py
index 62b992d2e..f4f900b95 100644
--- a/test/ext/test_horizontal_shard.py
+++ b/test/ext/test_horizontal_shard.py
@@ -6,6 +6,7 @@ from sqlalchemy.orm import *
 from sqlalchemy.ext.horizontal_shard import ShardedSession
 from sqlalchemy.sql import operators
 from test.lib import *
+from test.lib.engines import testing_engine
 from test.lib.testing import eq_
 from nose import SkipTest
 
@@ -16,12 +17,12 @@ class ShardTest(fixtures.TestBase):
         global db1, db2, db3, db4, weather_locations, weather_reports
 
         try:
-            db1 = create_engine('sqlite:///shard1.db', pool_threadlocal=True)
+            db1 = testing_engine('sqlite:///shard1.db', options=dict(pool_threadlocal=True))
         except ImportError:
             raise SkipTest('Requires sqlite')
-        db2 = create_engine('sqlite:///shard2.db')
-        db3 = create_engine('sqlite:///shard3.db')
-        db4 = create_engine('sqlite:///shard4.db')
+        db2 = testing_engine('sqlite:///shard2.db')
+        db3 = testing_engine('sqlite:///shard3.db')
+        db4 = testing_engine('sqlite:///shard4.db')
 
         meta = MetaData()
         ids = Table('ids', meta,
diff --git a/test/lib/engines.py b/test/lib/engines.py
index 4794a5fab..3a5132b8b 100644
--- a/test/lib/engines.py
+++ b/test/lib/engines.py
@@ -3,37 +3,55 @@ from collections import deque
 from test.bootstrap import config
 from test.lib.util import decorator
 from sqlalchemy.util import callable
-from sqlalchemy import event
+from sqlalchemy import event, pool
+from sqlalchemy.engine import base as engine_base
 import re
 import warnings
 
 class ConnectionKiller(object):
     def __init__(self):
         self.proxy_refs = weakref.WeakKeyDictionary()
+        self.testing_engines = weakref.WeakKeyDictionary()
+        self.conns = set()
+
+    def add_engine(self, engine):
+        self.testing_engines[engine] = True
 
     def checkout(self, dbapi_con, con_record, con_proxy):
         self.proxy_refs[con_proxy] = True
+        self.conns.add(dbapi_con)
+
+    def _safe(self, fn):
+        try:
+            fn()
+        except (SystemExit, KeyboardInterrupt):
+            raise
+        except Exception, e:
+            warnings.warn(
+                    "testing_reaper couldn't "
+                    "rollback/close connection: %s" % e)
 
-    def _apply_all(self, methods):
-        # must copy keys atomically
+    def rollback_all(self):
         for rec in self.proxy_refs.keys():
             if rec is not None and rec.is_valid:
-                try:
-                    for name in methods:
-                        if callable(name):
-                            name(rec)
-                        else:
-                            getattr(rec, name)()
-                except (SystemExit, KeyboardInterrupt):
-                    raise
-                except Exception, e:
-                    warnings.warn("testing_reaper couldn't close connection: %s" % e)
-
-    def rollback_all(self):
-        self._apply_all(('rollback',))
+                self._safe(rec.rollback)
 
     def close_all(self):
-        self._apply_all(('rollback', 'close'))
+        for rec in self.proxy_refs.keys():
+            if rec is not None:
+                self._safe(rec._close)
+
+    def _after_test_ctx(self):
+        for conn in self.conns:
+            self._safe(conn.rollback)
+
+    def _stop_test_ctx(self):
+        self.close_all()
+        for conn in self.conns:
+            self._safe(conn.close)
+        self.conns = set()
+        for rec in self.testing_engines.keys():
+            rec.dispose()
 
     def assert_all_closed(self):
         for rec in self.proxy_refs:
@@ -134,6 +152,9 @@ def testing_engine(url=None, options=None):
     options = options or config.db_opts
 
     engine = create_engine(url, **options)
+    if isinstance(engine.pool, pool.QueuePool):
+        engine.pool._timeout = 0
+        engine.pool._max_overflow = 0
     event.listen(engine, 'after_execute', asserter.execute)
     event.listen(engine, 'after_cursor_execute', asserter.cursor_execute)
     event.listen(engine.pool, 'checkout', testing_reaper.checkout)
@@ -141,6 +162,7 @@ def testing_engine(url=None, options=None):
     # may want to call this, results
     # in first-connect initializers
     #engine.connect()
+    testing_reaper.add_engine(engine)
 
     return engine
 
diff --git a/test/lib/testing.py b/test/lib/testing.py
index 6512a23d2..f5babc19c 100644
--- a/test/lib/testing.py
+++ b/test/lib/testing.py
@@ -447,7 +447,7 @@ def global_cleanup_assertions():
     """
 
     testutil.lazy_gc()
-    assert not pool._refs
+    assert not pool._refs, str(pool._refs)
 
 def against(*queries):
     """Boolean predicate, compares to testing database configuration.