- add a connect=True key to connection record to support

pre-loading of _ConnectionRecord objects
- ensure _ConnectionRecord.close() leaves the record in a good
state for reopening

Change-Id: I4eeb8e29f643950afb9eaab94d815e2a115db3d6

2 files changed, 51 insertions, 2 deletions

diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 4bd8f60ec..087f22c60 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -445,9 +445,10 @@ class _ConnectionRecord(object):
 
     """
 
-    def __init__(self, pool):
+    def __init__(self, pool, connect=True):
         self.__pool = pool
-        self.__connect(first_connect_check=True)
+        if connect:
+            self.__connect(first_connect_check=True)
         self.finalize_callback = deque()
 
     connection = None
@@ -590,6 +591,7 @@ class _ConnectionRecord(object):
         if self.__pool.dispatch.close:
             self.__pool.dispatch.close(self.connection, self)
         self.__pool._close_connection(self.connection)
+        self.connection = None
 
     def __connect(self, first_connect_check=False):
         pool = self.__pool
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 057289199..56a3d6d23 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -232,6 +232,53 @@ class PoolTest(PoolTestBase):
         assert not c2.info
         assert 'foo2' in c.info
 
+    def test_rec_unconnected(self):
+        # test production of a _ConnectionRecord with an
+        # initally unconnected state.
+
+        dbapi = MockDBAPI()
+        p1 = pool.Pool(
+            creator=lambda: dbapi.connect('foo.db')
+        )
+
+        r1 = pool._ConnectionRecord(p1, connect=False)
+
+        assert not r1.connection
+        c1 = r1.get_connection()
+        is_(c1, r1.connection)
+
+    def test_rec_close_reopen(self):
+        # test that _ConnectionRecord.close() allows
+        # the record to be reusable
+        dbapi = MockDBAPI()
+        p1 = pool.Pool(
+            creator=lambda: dbapi.connect('foo.db')
+        )
+
+        r1 = pool._ConnectionRecord(p1)
+
+        c1 = r1.connection
+        c2 = r1.get_connection()
+        is_(c1, c2)
+
+        r1.close()
+
+        assert not r1.connection
+        eq_(
+            c1.mock_calls,
+            [call.close()]
+        )
+
+        c2 = r1.get_connection()
+
+        is_not_(c1, c2)
+        is_(c2, r1.connection)
+
+        eq_(
+            c2.mock_calls,
+            []
+        )
+
 
 class PoolDialectTest(PoolTestBase):
     def _dialect(self):