From c91fd822bc9816827d0aab4699e304ab49ed8280 Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Sun, 12 Jan 2014 17:34:20 -0500 Subject: - add new event PoolEvents.invalidate(). allows interception of invalidation events including auto-invalidation, which is useful both for tests here as well as detecting failure conditions within the "reset" or "close" cases. - rename the argument for PoolEvents.reset() to dbapi_connection and connection_record to be consistent with everything else. - add new documentation sections on invalidation, including auto-invalidation and the invalidation process within the pool. - add _ConnectionFairy and _ConnectionRecord to the pool documentation. Establish docs for common _ConnectionFairy/_ConnectionRecord methods and accessors and have PoolEvents docs refer to _ConnectionRecord, since it is passed to all events. Rename a few _ConnectionFairy methods that are actually private to pool such as _checkout(), _checkin() and _checkout_existing(); there should not be any external code calling these --- test/engine/test_pool.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) (limited to 'test/engine') diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py index eb70bdf7f..10f490b48 100644 --- a/test/engine/test_pool.py +++ b/test/engine/test_pool.py @@ -308,6 +308,13 @@ class PoolEventsTest(PoolTestBase): return p, canary + def _invalidate_event_fixture(self): + p = self._queuepool_fixture() + canary = Mock() + event.listen(p, 'invalidate', canary) + + return p, canary + def test_first_connect_event(self): p, canary = self._first_connect_event_fixture() @@ -411,6 +418,31 @@ class PoolEventsTest(PoolTestBase): c1.close() eq_(canary, ['reset']) + def test_invalidate_event_no_exception(self): + p, canary = self._invalidate_event_fixture() + + c1 = p.connect() + c1.close() + assert not canary.called + c1 = p.connect() + dbapi_con = c1.connection + c1.invalidate() + assert canary.call_args_list[0][0][0] is dbapi_con + assert canary.call_args_list[0][0][2] is None + + def test_invalidate_event_exception(self): + p, canary = self._invalidate_event_fixture() + + c1 = p.connect() + c1.close() + assert not canary.called + c1 = p.connect() + dbapi_con = c1.connection + exc = Exception("hi") + c1.invalidate(exc) + assert canary.call_args_list[0][0][0] is dbapi_con + assert canary.call_args_list[0][0][2] is exc + def test_checkin_event_gc(self): p, canary = self._checkin_event_fixture() -- cgit v1.2.1