From abccc0624228def744b0382e84f01cf95e0d3aed Mon Sep 17 00:00:00 2001
From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Sat, 16 Jan 2010 22:44:04 +0000
Subject: - added "statement_options()" to Query, to so options can be passed
 to the resulting statement. Currently only Select-statements have these
 options, and the only option used is "stream_results", and the only dialect
 which knows "stream_results" is psycopg2. - Query.yield_per() will set the
 "stream_results" statement option automatically. - Added
 "statement_options()" to Selects, which set statement specific options. These
 enable e.g. dialect specific options such as whether to enable using server
 side cursors, etc. - The psycopg2 now respects the statement option
 "stream_results". This option overrides the connection setting
 "server_side_cursors". If true, server side cursors will be used for the
 statement. If false, they will not be used, even if "server_side_cursors" is
 true on the connection. [ticket:1619] - added a "frozendict" from
 http://code.activestate.com/recipes/414283/, adding more default collections
 as immutable class vars on Query, Insert, Select

---
 test/dialect/test_postgresql.py | 72 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 72 insertions(+)

(limited to 'test/dialect/test_postgresql.py')

diff --git a/test/dialect/test_postgresql.py b/test/dialect/test_postgresql.py
index 9833042fe..5d6bcaf6d 100644
--- a/test/dialect/test_postgresql.py
+++ b/test/dialect/test_postgresql.py
@@ -1448,6 +1448,78 @@ class ServerSideCursorsTest(TestBase, AssertsExecutionResults):
 
         result = ss_engine.execute(select([1]))
         assert result.cursor.name
+
+    def test_uses_ss_when_explicitly_enabled(self):
+        engine = engines.testing_engine(options={'server_side_cursors':False})
+        result = engine.execute(text("select 1"))
+        # It should be off globally ...
+        assert not result.cursor.name
+
+        s = select([1]).statement_options(stream_results=True)
+        result = engine.execute(s)
+        # ... but enabled for this one.
+        assert result.cursor.name
+
+    def test_ss_explicitly_disabled(self):
+        s = select([1]).statement_options(stream_results=False)
+        result = ss_engine.execute(s)
+        assert not result.cursor.name
+
+    def test_aliases_and_ss(self):
+        engine = engines.testing_engine(options={'server_side_cursors':False})
+        s1 = select([1]).statement_options(stream_results=True).alias()
+        result = engine.execute(s1)
+        assert result.cursor.name
+
+        # s1's options shouldn't affect s2 when s2 is used as a from_obj.
+        s2 = select([1], from_obj=s1)
+        result = engine.execute(s2)
+        assert not result.cursor.name
+
+    def test_for_update_and_ss(self):
+        s1 = select([1], for_update=True)
+        result = ss_engine.execute(s1)
+        assert result.cursor.name
+
+        result = ss_engine.execute('SELECT 1 FOR UPDATE')
+        assert result.cursor.name
+
+    def test_orm_queries_with_ss(self):
+        metadata = MetaData(testing.db)
+        class Foo(object): pass
+        footable = Table('foobar', metadata,
+            Column('id', Integer, primary_key=True),
+        )
+        mapper(Foo, footable)
+        metadata.create_all()
+        try:
+            sess = create_session()
+
+            engine = engines.testing_engine(options={'server_side_cursors':False})
+            result = engine.execute(sess.query(Foo).statement)
+            assert not result.cursor.name, result.cursor.name
+            result.close()
+
+            q = sess.query(Foo).statement_options(stream_results=True)
+            result = engine.execute(q.statement)
+            assert result.cursor.name
+            result.close()
+
+            result = sess.query(Foo).statement_options(stream_results=True).subquery().execute()
+            assert result.cursor.name
+            result.close()
+        finally:
+            metadata.drop_all()
+            
+    def test_text_with_ss(self):
+        engine = engines.testing_engine(options={'server_side_cursors':False})
+        s = text('select 42')
+        result = engine.execute(s)
+        assert not result.cursor.name
+        s = text('select 42', statement_options=dict(stream_results=True))
+        result = engine.execute(s)
+        assert result.cursor.name
+
         
     def test_roundtrip(self):
         test_table = Table('test_table', MetaData(ss_engine),
-- 
cgit v1.2.1