diff options
Diffstat (limited to 'test/dialect/postgresql/test_query.py')
-rw-r--r-- | test/dialect/postgresql/test_query.py | 133 |
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' |