ResultProxy won't autoclose connection until state flag is set

Changed the mechanics of :class:`.ResultProxy` to unconditionally
delay the "autoclose" step until the :class:`.Connection` is done
with the object; in the case where Postgresql ON CONFLICT with
RETURNING returns no rows, autoclose was occurring in this previously
non-existent use case, causing the usual autocommit behavior that
occurs unconditionally upon INSERT/UPDATE/DELETE to fail.

Change-Id: I235a25daf4381b31f523331f810ea04450349722
Fixes: #3955
(cherry picked from commit 8ee363e4917b0dcd64a83b6d26e465c9e61e0ea5)
(cherry picked from commit f52fb5282a046d26b6ee2778e03b995eb117c2ee)

1 files changed, 44 insertions, 0 deletions

diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index de677be9a..48fe28861 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -489,6 +489,50 @@ class ResultProxyTest(fixtures.TablesTest):
             result.fetchone
         )
 
+    def test_connectionless_autoclose_rows_exhausted(self):
+        users = self.tables.users
+        users.insert().execute(
+            dict(user_id=1, user_name='john'),
+        )
+
+        result = testing.db.execute("select * from users")
+        connection = result.connection
+        assert not connection.closed
+        eq_(result.fetchone(), (1, 'john'))
+        assert not connection.closed
+        eq_(result.fetchone(), None)
+        assert connection.closed
+
+    @testing.requires.returning
+    def test_connectionless_autoclose_crud_rows_exhausted(self):
+        users = self.tables.users
+        stmt = users.insert().values(user_id=1, user_name='john').\
+            returning(users.c.user_id)
+        result = testing.db.execute(stmt)
+        connection = result.connection
+        assert not connection.closed
+        eq_(result.fetchone(), (1, ))
+        assert not connection.closed
+        eq_(result.fetchone(), None)
+        assert connection.closed
+
+    def test_connectionless_autoclose_no_rows(self):
+        result = testing.db.execute("select * from users")
+        connection = result.connection
+        assert not connection.closed
+        eq_(result.fetchone(), None)
+        assert connection.closed
+
+    def test_connectionless_autoclose_no_metadata(self):
+        result = testing.db.execute("update users set user_id=5")
+        connection = result.connection
+        assert connection.closed
+        assert_raises_message(
+            exc.ResourceClosedError,
+            "This result object does not return rows.",
+            result.fetchone
+        )
+
     def test_row_case_sensitive(self):
         row = testing.db.execute(
             select([