diff options
Diffstat (limited to 'lib/sqlalchemy/testing/suite/test_results.py')
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_results.py | 242 |
1 files changed, 119 insertions, 123 deletions
diff --git a/lib/sqlalchemy/testing/suite/test_results.py b/lib/sqlalchemy/testing/suite/test_results.py index f464d47eb..247f05cf5 100644 --- a/lib/sqlalchemy/testing/suite/test_results.py +++ b/lib/sqlalchemy/testing/suite/test_results.py @@ -15,14 +15,18 @@ class RowFetchTest(fixtures.TablesTest): @classmethod def define_tables(cls, metadata): - Table('plain_pk', metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)) - ) - Table('has_dates', metadata, - Column('id', Integer, primary_key=True), - Column('today', DateTime) - ) + Table( + "plain_pk", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), + ) + Table( + "has_dates", + metadata, + Column("id", Integer, primary_key=True), + Column("today", DateTime), + ) @classmethod def insert_data(cls): @@ -32,65 +36,51 @@ class RowFetchTest(fixtures.TablesTest): {"id": 1, "data": "d1"}, {"id": 2, "data": "d2"}, {"id": 3, "data": "d3"}, - ] + ], ) config.db.execute( cls.tables.has_dates.insert(), - [ - {"id": 1, "today": datetime.datetime(2006, 5, 12, 12, 0, 0)} - ] + [{"id": 1, "today": datetime.datetime(2006, 5, 12, 12, 0, 0)}], ) def test_via_string(self): row = config.db.execute( - self.tables.plain_pk.select(). - order_by(self.tables.plain_pk.c.id) + self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id) ).first() - eq_( - row['id'], 1 - ) - eq_( - row['data'], "d1" - ) + eq_(row["id"], 1) + eq_(row["data"], "d1") def test_via_int(self): row = config.db.execute( - self.tables.plain_pk.select(). - order_by(self.tables.plain_pk.c.id) + self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id) ).first() - eq_( - row[0], 1 - ) - eq_( - row[1], "d1" - ) + eq_(row[0], 1) + eq_(row[1], "d1") def test_via_col_object(self): row = config.db.execute( - self.tables.plain_pk.select(). - order_by(self.tables.plain_pk.c.id) + self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id) ).first() - eq_( - row[self.tables.plain_pk.c.id], 1 - ) - eq_( - row[self.tables.plain_pk.c.data], "d1" - ) + eq_(row[self.tables.plain_pk.c.id], 1) + eq_(row[self.tables.plain_pk.c.data], "d1") @requirements.duplicate_names_in_cursor_description def test_row_with_dupe_names(self): result = config.db.execute( - select([self.tables.plain_pk.c.data, - self.tables.plain_pk.c.data.label('data')]). - order_by(self.tables.plain_pk.c.id) + select( + [ + self.tables.plain_pk.c.data, + self.tables.plain_pk.c.data.label("data"), + ] + ).order_by(self.tables.plain_pk.c.id) ) row = result.first() - eq_(result.keys(), ['data', 'data']) - eq_(row, ('d1', 'd1')) + eq_(result.keys(), ["data", "data"]) + eq_(row, ("d1", "d1")) def test_row_w_scalar_select(self): """test that a scalar select as a column is returned as such @@ -101,11 +91,11 @@ class RowFetchTest(fixtures.TablesTest): """ datetable = self.tables.has_dates - s = select([datetable.alias('x').c.today]).as_scalar() - s2 = select([datetable.c.id, s.label('somelabel')]) + s = select([datetable.alias("x").c.today]).as_scalar() + s2 = select([datetable.c.id, s.label("somelabel")]) row = config.db.execute(s2).first() - eq_(row['somelabel'], datetime.datetime(2006, 5, 12, 12, 0, 0)) + eq_(row["somelabel"], datetime.datetime(2006, 5, 12, 12, 0, 0)) class PercentSchemaNamesTest(fixtures.TablesTest): @@ -117,29 +107,31 @@ class PercentSchemaNamesTest(fixtures.TablesTest): """ - __requires__ = ('percent_schema_names', ) + __requires__ = ("percent_schema_names",) __backend__ = True @classmethod def define_tables(cls, metadata): - cls.tables.percent_table = Table('percent%table', metadata, - Column("percent%", Integer), - Column( - "spaces % more spaces", Integer), - ) + cls.tables.percent_table = Table( + "percent%table", + metadata, + Column("percent%", Integer), + Column("spaces % more spaces", Integer), + ) cls.tables.lightweight_percent_table = sql.table( - 'percent%table', sql.column("percent%"), - sql.column("spaces % more spaces") + "percent%table", + sql.column("percent%"), + sql.column("spaces % more spaces"), ) def test_single_roundtrip(self): percent_table = self.tables.percent_table for params in [ - {'percent%': 5, 'spaces % more spaces': 12}, - {'percent%': 7, 'spaces % more spaces': 11}, - {'percent%': 9, 'spaces % more spaces': 10}, - {'percent%': 11, 'spaces % more spaces': 9} + {"percent%": 5, "spaces % more spaces": 12}, + {"percent%": 7, "spaces % more spaces": 11}, + {"percent%": 9, "spaces % more spaces": 10}, + {"percent%": 11, "spaces % more spaces": 9}, ]: config.db.execute(percent_table.insert(), params) self._assert_table() @@ -147,14 +139,15 @@ class PercentSchemaNamesTest(fixtures.TablesTest): def test_executemany_roundtrip(self): percent_table = self.tables.percent_table config.db.execute( - percent_table.insert(), - {'percent%': 5, 'spaces % more spaces': 12} + percent_table.insert(), {"percent%": 5, "spaces % more spaces": 12} ) config.db.execute( percent_table.insert(), - [{'percent%': 7, 'spaces % more spaces': 11}, - {'percent%': 9, 'spaces % more spaces': 10}, - {'percent%': 11, 'spaces % more spaces': 9}] + [ + {"percent%": 7, "spaces % more spaces": 11}, + {"percent%": 9, "spaces % more spaces": 10}, + {"percent%": 11, "spaces % more spaces": 9}, + ], ) self._assert_table() @@ -163,85 +156,81 @@ class PercentSchemaNamesTest(fixtures.TablesTest): lightweight_percent_table = self.tables.lightweight_percent_table for table in ( - percent_table, - percent_table.alias(), - lightweight_percent_table, - lightweight_percent_table.alias()): + percent_table, + percent_table.alias(), + lightweight_percent_table, + lightweight_percent_table.alias(), + ): eq_( list( config.db.execute( - table.select().order_by(table.c['percent%']) + table.select().order_by(table.c["percent%"]) ) ), - [ - (5, 12), - (7, 11), - (9, 10), - (11, 9) - ] + [(5, 12), (7, 11), (9, 10), (11, 9)], ) eq_( list( config.db.execute( - table.select(). - where(table.c['spaces % more spaces'].in_([9, 10])). - order_by(table.c['percent%']), + table.select() + .where(table.c["spaces % more spaces"].in_([9, 10])) + .order_by(table.c["percent%"]) ) ), - [ - (9, 10), - (11, 9) - ] + [(9, 10), (11, 9)], ) - row = config.db.execute(table.select(). - order_by(table.c['percent%'])).first() - eq_(row['percent%'], 5) - eq_(row['spaces % more spaces'], 12) + row = config.db.execute( + table.select().order_by(table.c["percent%"]) + ).first() + eq_(row["percent%"], 5) + eq_(row["spaces % more spaces"], 12) - eq_(row[table.c['percent%']], 5) - eq_(row[table.c['spaces % more spaces']], 12) + eq_(row[table.c["percent%"]], 5) + eq_(row[table.c["spaces % more spaces"]], 12) config.db.execute( percent_table.update().values( - {percent_table.c['spaces % more spaces']: 15} + {percent_table.c["spaces % more spaces"]: 15} ) ) eq_( list( config.db.execute( - percent_table. - select(). - order_by(percent_table.c['percent%']) + percent_table.select().order_by( + percent_table.c["percent%"] + ) ) ), - [(5, 15), (7, 15), (9, 15), (11, 15)] + [(5, 15), (7, 15), (9, 15), (11, 15)], ) -class ServerSideCursorsTest(fixtures.TestBase, testing.AssertsExecutionResults): +class ServerSideCursorsTest( + fixtures.TestBase, testing.AssertsExecutionResults +): - __requires__ = ('server_side_cursors', ) + __requires__ = ("server_side_cursors",) __backend__ = True def _is_server_side(self, cursor): if self.engine.dialect.driver == "psycopg2": return cursor.name - elif self.engine.dialect.driver == 'pymysql': - sscursor = __import__('pymysql.cursors').cursors.SSCursor + elif self.engine.dialect.driver == "pymysql": + sscursor = __import__("pymysql.cursors").cursors.SSCursor return isinstance(cursor, sscursor) elif self.engine.dialect.driver == "mysqldb": - sscursor = __import__('MySQLdb.cursors').cursors.SSCursor + sscursor = __import__("MySQLdb.cursors").cursors.SSCursor return isinstance(cursor, sscursor) else: return False def _fixture(self, server_side_cursors): self.engine = engines.testing_engine( - options={'server_side_cursors': server_side_cursors} + options={"server_side_cursors": server_side_cursors} ) return self.engine @@ -251,12 +240,12 @@ class ServerSideCursorsTest(fixtures.TestBase, testing.AssertsExecutionResults): def test_global_string(self): engine = self._fixture(True) - result = engine.execute('select 1') + result = engine.execute("select 1") assert self._is_server_side(result.cursor) def test_global_text(self): engine = self._fixture(True) - result = engine.execute(text('select 1')) + result = engine.execute(text("select 1")) assert self._is_server_side(result.cursor) def test_global_expr(self): @@ -266,7 +255,7 @@ class ServerSideCursorsTest(fixtures.TestBase, testing.AssertsExecutionResults): def test_global_off_explicit(self): engine = self._fixture(False) - result = engine.execute(text('select 1')) + result = engine.execute(text("select 1")) # It should be off globally ... @@ -286,10 +275,11 @@ class ServerSideCursorsTest(fixtures.TestBase, testing.AssertsExecutionResults): engine = self._fixture(False) # and this one - result = \ - engine.connect().execution_options(stream_results=True).\ - execute('select 1' - ) + result = ( + engine.connect() + .execution_options(stream_results=True) + .execute("select 1") + ) assert self._is_server_side(result.cursor) def test_stmt_enabled_conn_option_disabled(self): @@ -298,9 +288,9 @@ class ServerSideCursorsTest(fixtures.TestBase, testing.AssertsExecutionResults): s = select([1]).execution_options(stream_results=True) # not this one - result = \ - engine.connect().execution_options(stream_results=False).\ - execute(s) + result = ( + engine.connect().execution_options(stream_results=False).execute(s) + ) assert not self._is_server_side(result.cursor) def test_stmt_option_disabled(self): @@ -329,18 +319,18 @@ class ServerSideCursorsTest(fixtures.TestBase, testing.AssertsExecutionResults): def test_for_update_string(self): engine = self._fixture(True) - result = engine.execute('SELECT 1 FOR UPDATE') + result = engine.execute("SELECT 1 FOR UPDATE") assert self._is_server_side(result.cursor) def test_text_no_ss(self): engine = self._fixture(False) - s = text('select 42') + s = text("select 42") result = engine.execute(s) assert not self._is_server_side(result.cursor) def test_text_ss_option(self): engine = self._fixture(False) - s = text('select 42').execution_options(stream_results=True) + s = text("select 42").execution_options(stream_results=True) result = engine.execute(s) assert self._is_server_side(result.cursor) @@ -349,19 +339,25 @@ class ServerSideCursorsTest(fixtures.TestBase, testing.AssertsExecutionResults): 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 = 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') - test_table.insert().execute(data='data2') - eq_(test_table.select().order_by(test_table.c.id).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().order_by(test_table.c.id).execute().fetchall(), - [(1, 'data1'), (2, 'data2 updated')]) + test_table.insert().execute(data="data1") + test_table.insert().execute(data="data2") + eq_( + test_table.select().order_by(test_table.c.id).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().order_by(test_table.c.id).execute().fetchall(), + [(1, "data1"), (2, "data2 updated")], + ) test_table.delete().execute() - eq_(select([func.count('*')]).select_from(test_table).scalar(), 0) + eq_(select([func.count("*")]).select_from(test_table).scalar(), 0) |