- 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

1 files changed, 72 insertions, 0 deletions

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),