1. Éric Lemoine
  2. sqlalchemy


Mike Bayer  committed 38baf8d

- given that Oracle's issue is not just open connections, but *closed* ones too, go really
heavy handed and start marking tests as "requires.ad_hoc_engines", add a flag --low-connections
that will switch the engine reaper mechanism to use as *few* distinct engines and connections
as possible, many engine tests that really need their own engines are just skipped.

  • Participants
  • Parent commits ca1c1d1
  • Branches default

Comments (0)

Files changed (6)

File test/bootstrap/config.py

View file
 pre_configure = []
 post_configure = []
+def _setup_options(opt, file_config):
+    global options
+    options = opt
 def _monkeypatch_cdecimal(options, file_config):
     if options.cdecimal:
         import sys

File test/bootstrap/noseplugin.py

View file
             help="Use mock pool (asserts only one connection used)")
         opt("--zero-timeout", action="callback", callback=_zero_timeout,
             help="Set pool_timeout to zero, applies to QueuePool only")
+        opt("--low-connections", action="store_true", dest="low_connections",
+            help="Use a low number of distinct connections - i.e. for Oracle TNS"
+        )
         opt("--enginestrategy", action="callback", type="string",
             help="Engine strategy (plain or threadlocal, defaults to plain)")
     def stopContext(self, ctx):
-        testing.global_cleanup_assertions()
+        if not config.options.low_connections:
+            testing.global_cleanup_assertions()
     #def handleError(self, test, err):

File test/engine/test_execute.py

View file
         eq_(testing.db.execute(users_autoinc.select()).fetchall(), [(1,
+    @testing.requires.ad_hoc_engines
     def test_engine_level_options(self):
         eng = engines.testing_engine(options={'execution_options'
                 : {'foo': 'bar'}})
         eq_(conn.execute("select count(*) from users").scalar(), 3)
 class LoggingNameTest(fixtures.TestBase):
+    __requires__ = 'ad_hoc_engines',
     def _assert_names_in_execute(self, eng, eng_name, pool_name):
         for name in [b.name for b in self.buf.buffer]:
 class EchoTest(fixtures.TestBase):
+    __requires__ = 'ad_hoc_engines',
     def setup(self):
         self.level = logging.getLogger('sqlalchemy.engine').level
 class EngineEventsTest(fixtures.TestBase):
+    __requires__ = 'ad_hoc_engines', 
     def tearDown(self):
     the deprecated ConnectionProxy interface.
+    __requires__ = 'ad_hoc_engines', 
     @testing.uses_deprecated(r'.*Use event.listen')
     @testing.fails_on('firebird', 'Data type unknown')

File test/engine/test_transaction.py

View file
 class TLTransactionTest(fixtures.TestBase):
+    __requires__ = ('ad_hoc_engines', )
     def setup_class(cls):
         self.assert_(len(errors) != 0)
 class IsolationLevelTest(fixtures.TestBase):
-    __requires__ = ('isolation_level',)
+    __requires__ = ('isolation_level', 'ad_hoc_engines')
     def _default_isolation_level(self):
         if testing.against('sqlite'):

File test/lib/engines.py

View file
         #    self._safe(conn.rollback)
     def _stop_test_ctx(self):
+        if config.options.low_connections:
+            self._stop_test_ctx_minimal()
+        else:
+            self._stop_test_ctx_aggressive()
+    def _stop_test_ctx_minimal(self):
+        from test.lib import testing
+        self.close_all()
+        for rec in self.testing_engines.keys():
+            if rec is not testing.db:
+                rec.dispose()
+    def _stop_test_ctx_aggressive(self):
         for conn in self.conns:

File test/lib/requires.py

View file
 from sqlalchemy import util
+from test.lib import config
 import testing
 import sys
         skip_if(lambda: not _has_sqlite())
+def ad_hoc_engines(fn):
+    """Test environment must allow ad-hoc engine/connection creation.
+    DBs that scale poorly for many connections, even when closed, i.e.
+    Oracle, may use the "--low-connections" option which flags this requirement
+    as not present.
+    """
+    return _chain_decorators_on(
+        fn,
+        skip_if(lambda: config.options.low_connections)
+    )