except tsa.exc.TimeoutError:
assert int(time.time() - now) == 2
+ @testing.requires.threading_with_mock
def test_timeout_race(self):
# test a race condition where the initial connecting threads all race
# to queue.Empty, then block on the mutex. each thread consumes a
lazy_gc()
assert not pool._refs
+ @testing.requires.threading_with_mock
def test_waiters_handled(self):
"""test that threads waiting for connections are
handled when the pool is replaced.
time.sleep(2)
eq_(len(success), 12)
+ @testing.requires.threading_with_mock
@testing.requires.python26
def test_notify_waiters(self):
dbapi = MockDBAPI()
c3 = p.connect()
assert c3.connection is c2_con
+ @testing.requires.threading_with_mock
def test_no_overflow(self):
self._test_overflow(40, 0)
+ @testing.requires.threading_with_mock
def test_max_overflow(self):
self._test_overflow(40, 5)
class SingletonThreadPoolTest(PoolTestBase):
+ @testing.requires.threading_with_mock
def test_cleanup(self):
self._test_cleanup(False)
+ @testing.requires.threading_with_mock
def test_cleanup_no_gc(self):
self._test_cleanup(True)
"Not supported on MySQL + Windows"
)
+ def threading_with_mock(self, fn):
+ """Mark tests that use threading and mock at the same time, so they
+ can be excluded using "-a '!threading_with_mock'" - stability
+ issues have been observed with coverage + python 3.3
+
+ """
+ fn.threading_with_mock = True
+ return fn
+
@property
def english_locale_on_postgresql(self):
return skip_if(lambda: against('postgresql') \