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