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)

2 files changed, 73 insertions, 0 deletions

diff --git a/test/dialect/postgresql/test_on_conflict.py b/test/dialect/postgresql/test_on_conflict.py
index 7c83f2826..c3e1b9158 100644
--- a/test/dialect/postgresql/test_on_conflict.py
+++ b/test/dialect/postgresql/test_on_conflict.py
@@ -12,6 +12,7 @@ class OnConflictTest(fixtures.TablesTest):
 
     __only_on__ = 'postgresql >= 9.5',
     __backend__ = True
+    run_define_tables = 'each'
 
     @classmethod
     def define_tables(cls, metadata):
@@ -79,6 +80,7 @@ class OnConflictTest(fixtures.TablesTest):
         with testing.db.connect() as conn:
             result = conn.execute(
                 insert(users).on_conflict_do_nothing(),
+
                 dict(id=1, name='name1')
             )
             eq_(result.inserted_primary_key, [1])
@@ -96,6 +98,33 @@ class OnConflictTest(fixtures.TablesTest):
                 [(1, 'name1')]
             )
 
+    def test_on_conflict_do_nothing_connectionless(self):
+        users = self.tables.users_xtra
+
+        with testing.db.connect() as conn:
+            result = conn.execute(
+                insert(users).on_conflict_do_nothing(
+                    constraint='uq_login_email'),
+
+                dict(name='name1', login_email='email1')
+            )
+            eq_(result.inserted_primary_key, [1])
+            eq_(result.returned_defaults, (1,))
+
+        result = testing.db.execute(
+            insert(users).on_conflict_do_nothing(
+                constraint='uq_login_email'
+            ),
+            dict(name='name2', login_email='email1')
+        )
+        eq_(result.inserted_primary_key, None)
+        eq_(result.returned_defaults, None)
+
+        eq_(
+            testing.db.execute(users.select().where(users.c.id == 1)).fetchall(),
+            [(1, 'name1', 'email1', None)]
+        )
+
     @testing.provide_metadata
     def test_on_conflict_do_nothing_target(self):
         users = self.tables.users
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([