summaryrefslogtreecommitdiff
path: root/test/dialect/postgresql/test_query.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/dialect/postgresql/test_query.py')
-rw-r--r--test/dialect/postgresql/test_query.py133
1 files changed, 0 insertions, 133 deletions
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index b8129f1e3..47a12afec 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -595,139 +595,6 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
(33, 'd4')])
-class ServerSideCursorsTest(fixtures.TestBase, AssertsExecutionResults):
-
- __requires__ = 'psycopg2_compatibility',
-
- def _fixture(self, server_side_cursors):
- self.engine = engines.testing_engine(
- options={'server_side_cursors': server_side_cursors}
- )
- return self.engine
-
- def tearDown(self):
- engines.testing_reaper.close_all()
- self.engine.dispose()
-
- def test_global_string(self):
- engine = self._fixture(True)
- result = engine.execute('select 1')
- assert result.cursor.name
-
- def test_global_text(self):
- engine = self._fixture(True)
- result = engine.execute(text('select 1'))
- assert result.cursor.name
-
- def test_global_expr(self):
- engine = self._fixture(True)
- result = engine.execute(select([1]))
- assert result.cursor.name
-
- def test_global_off_explicit(self):
- engine = self._fixture(False)
- result = engine.execute(text('select 1'))
-
- # It should be off globally ...
-
- assert not result.cursor.name
-
- def test_stmt_option(self):
- engine = self._fixture(False)
-
- s = select([1]).execution_options(stream_results=True)
- result = engine.execute(s)
-
- # ... but enabled for this one.
-
- assert result.cursor.name
-
- def test_conn_option(self):
- engine = self._fixture(False)
-
- # and this one
- result = \
- engine.connect().execution_options(stream_results=True).\
- execute('select 1'
- )
- assert result.cursor.name
-
- def test_stmt_enabled_conn_option_disabled(self):
- engine = self._fixture(False)
-
- s = select([1]).execution_options(stream_results=True)
-
- # not this one
- result = \
- engine.connect().execution_options(stream_results=False).\
- execute(s)
- assert not result.cursor.name
-
- def test_stmt_option_disabled(self):
- engine = self._fixture(True)
- s = select([1]).execution_options(stream_results=False)
- result = engine.execute(s)
- assert not result.cursor.name
-
- def test_aliases_and_ss(self):
- engine = self._fixture(False)
- s1 = select([1]).execution_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_expr(self):
- engine = self._fixture(True)
- s1 = select([1], for_update=True)
- result = engine.execute(s1)
- assert result.cursor.name
-
- def test_for_update_string(self):
- engine = self._fixture(True)
- result = engine.execute('SELECT 1 FOR UPDATE')
- assert result.cursor.name
-
- def test_text_no_ss(self):
- engine = self._fixture(False)
- s = text('select 42')
- result = engine.execute(s)
- assert not result.cursor.name
-
- def test_text_ss_option(self):
- engine = self._fixture(False)
- s = text('select 42').execution_options(stream_results=True)
- result = engine.execute(s)
- assert result.cursor.name
-
- @testing.provide_metadata
- def test_roundtrip(self):
- md = self.metadata
-
- engine = self._fixture(True)
- test_table = Table('test_table', md,
- Column('id', Integer, primary_key=True),
- Column('data', String(50)))
- test_table.create(checkfirst=True)
- test_table.insert().execute(data='data1')
- nextid = engine.execute(Sequence('test_table_id_seq'))
- test_table.insert().execute(id=nextid, data='data2')
- eq_(test_table.select().execute().fetchall(), [(1, 'data1'
- ), (2, 'data2')])
- test_table.update().where(
- test_table.c.id == 2).values(
- data=test_table.c.data +
- ' updated').execute()
- eq_(test_table.select().execute().fetchall(),
- [(1, 'data1'), (2, 'data2 updated')])
- test_table.delete().execute()
- eq_(select([func.count('*')]).select_from(test_table).scalar(), 0)
-
-
class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
__only_on__ = 'postgresql >= 8.3'