Don't double-process ResultMetaData for BufferedColumnResultProxy

Fixed a bug in the result proxy used mainly by Oracle when binary and
other LOB types are in play, such that when query / statement caching
were used, the type-level result processors, notably that required by
the binary type itself but also any other processor, would become lost
after the first run of the statement due to it being removed from the
cached result metadata.

Change-Id: I751940866cffb4f48de46edc8137482eab59790c
Fixes: #3699
(cherry picked from commit f3bc60bdd809235cbeb3f414717ac0e273269cf9)

1 files changed, 47 insertions, 5 deletions

diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index f79318236..67815a930 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -6,7 +6,7 @@ from sqlalchemy import util
 from sqlalchemy import (
     exc, sql, func, select, String, Integer, MetaData, ForeignKey,
     VARCHAR, INT, CHAR, text, type_coerce, literal_column,
-    TypeDecorator, table, column)
+    TypeDecorator, table, column, literal)
 from sqlalchemy.engine import result as _result
 from sqlalchemy.testing.schema import Table, Column
 import operator
@@ -1433,18 +1433,60 @@ class AlternateResultProxyTest(fixtures.TablesTest):
             r.fetchall
         )
 
-    def test_plain(self):
+    def test_basic_plain(self):
         self._test_proxy(_result.ResultProxy)
 
-    def test_buffered_row_result_proxy(self):
+    def test_basic_buffered_row_result_proxy(self):
         self._test_proxy(_result.BufferedRowResultProxy)
 
-    def test_fully_buffered_result_proxy(self):
+    def test_basic_fully_buffered_result_proxy(self):
         self._test_proxy(_result.FullyBufferedResultProxy)
 
-    def test_buffered_column_result_proxy(self):
+    def test_basic_buffered_column_result_proxy(self):
         self._test_proxy(_result.BufferedColumnResultProxy)
 
+    def test_resultprocessor_plain(self):
+        self._test_result_processor(_result.ResultProxy, False)
+
+    def test_resultprocessor_plain_cached(self):
+        self._test_result_processor(_result.ResultProxy, True)
+
+    def test_resultprocessor_buffered_column(self):
+        self._test_result_processor(_result.BufferedColumnResultProxy, False)
+
+    def test_resultprocessor_buffered_column_cached(self):
+        self._test_result_processor(_result.BufferedColumnResultProxy, True)
+
+    def test_resultprocessor_buffered_row(self):
+        self._test_result_processor(_result.BufferedRowResultProxy, False)
+
+    def test_resultprocessor_buffered_row_cached(self):
+        self._test_result_processor(_result.BufferedRowResultProxy, True)
+
+    def test_resultprocessor_fully_buffered(self):
+        self._test_result_processor(_result.FullyBufferedResultProxy, False)
+
+    def test_resultprocessor_fully_buffered_cached(self):
+        self._test_result_processor(_result.FullyBufferedResultProxy, True)
+
+    def _test_result_processor(self, cls, use_cache):
+        class MyType(TypeDecorator):
+            impl = String()
+
+            def process_result_value(self, value, dialect):
+                return "HI " + value
+
+        with self._proxy_fixture(cls):
+            with self.engine.connect() as conn:
+                if use_cache:
+                    cache = {}
+                    conn = conn.execution_options(compiled_cache=cache)
+
+                stmt = select([literal("THERE", type_=MyType())])
+                for i in range(2):
+                    r = conn.execute(stmt)
+                    eq_(r.scalar(), "HI THERE")
+
     def test_buffered_row_growth(self):
         with self._proxy_fixture(_result.BufferedRowResultProxy):
             with self.engine.connect() as conn: