diff options
Diffstat (limited to 'lib/sqlalchemy/testing/suite')
| -rw-r--r-- | lib/sqlalchemy/testing/suite/__init__.py | 1 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_cte.py | 132 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_ddl.py | 50 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_dialect.py | 54 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_insert.py | 221 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_reflection.py | 762 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_results.py | 242 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_select.py | 483 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_sequence.py | 136 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_types.py | 702 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_update_delete.py | 37 |
11 files changed, 1403 insertions, 1417 deletions
diff --git a/lib/sqlalchemy/testing/suite/__init__.py b/lib/sqlalchemy/testing/suite/__init__.py index 748d9722d..a4e142c5a 100644 --- a/lib/sqlalchemy/testing/suite/__init__.py +++ b/lib/sqlalchemy/testing/suite/__init__.py @@ -1,4 +1,3 @@ - from sqlalchemy.testing.suite.test_cte import * from sqlalchemy.testing.suite.test_dialect import * from sqlalchemy.testing.suite.test_ddl import * diff --git a/lib/sqlalchemy/testing/suite/test_cte.py b/lib/sqlalchemy/testing/suite/test_cte.py index cc72278e6..d2f35933b 100644 --- a/lib/sqlalchemy/testing/suite/test_cte.py +++ b/lib/sqlalchemy/testing/suite/test_cte.py @@ -10,22 +10,28 @@ from ..schema import Table, Column class CTETest(fixtures.TablesTest): __backend__ = True - __requires__ = 'ctes', + __requires__ = ("ctes",) - run_inserts = 'each' - run_deletes = 'each' + run_inserts = "each" + run_deletes = "each" @classmethod def define_tables(cls, metadata): - Table("some_table", metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)), - Column("parent_id", ForeignKey("some_table.id"))) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), + Column("parent_id", ForeignKey("some_table.id")), + ) - Table("some_other_table", metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)), - Column("parent_id", Integer)) + Table( + "some_other_table", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), + Column("parent_id", Integer), + ) @classmethod def insert_data(cls): @@ -36,28 +42,33 @@ class CTETest(fixtures.TablesTest): {"id": 2, "data": "d2", "parent_id": 1}, {"id": 3, "data": "d3", "parent_id": 1}, {"id": 4, "data": "d4", "parent_id": 3}, - {"id": 5, "data": "d5", "parent_id": 3} - ] + {"id": 5, "data": "d5", "parent_id": 3}, + ], ) def test_select_nonrecursive_round_trip(self): some_table = self.tables.some_table with config.db.connect() as conn: - cte = select([some_table]).where( - some_table.c.data.in_(["d2", "d3", "d4"])).cte("some_cte") + cte = ( + select([some_table]) + .where(some_table.c.data.in_(["d2", "d3", "d4"])) + .cte("some_cte") + ) result = conn.execute( select([cte.c.data]).where(cte.c.data.in_(["d4", "d5"])) ) - eq_(result.fetchall(), [("d4", )]) + eq_(result.fetchall(), [("d4",)]) def test_select_recursive_round_trip(self): some_table = self.tables.some_table with config.db.connect() as conn: - cte = select([some_table]).where( - some_table.c.data.in_(["d2", "d3", "d4"])).cte( - "some_cte", recursive=True) + cte = ( + select([some_table]) + .where(some_table.c.data.in_(["d2", "d3", "d4"])) + .cte("some_cte", recursive=True) + ) cte_alias = cte.alias("c1") st1 = some_table.alias() @@ -67,12 +78,13 @@ class CTETest(fixtures.TablesTest): select([st1]).where(st1.c.id == cte_alias.c.parent_id) ) result = conn.execute( - select([cte.c.data]).where( - cte.c.data != "d2").order_by(cte.c.data.desc()) + select([cte.c.data]) + .where(cte.c.data != "d2") + .order_by(cte.c.data.desc()) ) eq_( result.fetchall(), - [('d4',), ('d3',), ('d3',), ('d1',), ('d1',), ('d1',)] + [("d4",), ("d3",), ("d3",), ("d1",), ("d1",), ("d1",)], ) def test_insert_from_select_round_trip(self): @@ -80,20 +92,21 @@ class CTETest(fixtures.TablesTest): some_other_table = self.tables.some_other_table with config.db.connect() as conn: - cte = select([some_table]).where( - some_table.c.data.in_(["d2", "d3", "d4"]) - ).cte("some_cte") + cte = ( + select([some_table]) + .where(some_table.c.data.in_(["d2", "d3", "d4"])) + .cte("some_cte") + ) conn.execute( some_other_table.insert().from_select( - ["id", "data", "parent_id"], - select([cte]) + ["id", "data", "parent_id"], select([cte]) ) ) eq_( conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), - [(2, "d2", 1), (3, "d3", 1), (4, "d4", 3)] + [(2, "d2", 1), (3, "d3", 1), (4, "d4", 3)], ) @testing.requires.ctes_with_update_delete @@ -105,27 +118,31 @@ class CTETest(fixtures.TablesTest): with config.db.connect() as conn: conn.execute( some_other_table.insert().from_select( - ['id', 'data', 'parent_id'], - select([some_table]) + ["id", "data", "parent_id"], select([some_table]) ) ) - cte = select([some_table]).where( - some_table.c.data.in_(["d2", "d3", "d4"]) - ).cte("some_cte") + cte = ( + select([some_table]) + .where(some_table.c.data.in_(["d2", "d3", "d4"])) + .cte("some_cte") + ) conn.execute( - some_other_table.update().values(parent_id=5).where( - some_other_table.c.data == cte.c.data - ) + some_other_table.update() + .values(parent_id=5) + .where(some_other_table.c.data == cte.c.data) ) eq_( conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), [ - (1, "d1", None), (2, "d2", 5), - (3, "d3", 5), (4, "d4", 5), (5, "d5", 3) - ] + (1, "d1", None), + (2, "d2", 5), + (3, "d3", 5), + (4, "d4", 5), + (5, "d5", 3), + ], ) @testing.requires.ctes_with_update_delete @@ -137,14 +154,15 @@ class CTETest(fixtures.TablesTest): with config.db.connect() as conn: conn.execute( some_other_table.insert().from_select( - ['id', 'data', 'parent_id'], - select([some_table]) + ["id", "data", "parent_id"], select([some_table]) ) ) - cte = select([some_table]).where( - some_table.c.data.in_(["d2", "d3", "d4"]) - ).cte("some_cte") + cte = ( + select([some_table]) + .where(some_table.c.data.in_(["d2", "d3", "d4"])) + .cte("some_cte") + ) conn.execute( some_other_table.delete().where( some_other_table.c.data == cte.c.data @@ -154,9 +172,7 @@ class CTETest(fixtures.TablesTest): conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), - [ - (1, "d1", None), (5, "d5", 3) - ] + [(1, "d1", None), (5, "d5", 3)], ) @testing.requires.ctes_with_update_delete @@ -168,26 +184,26 @@ class CTETest(fixtures.TablesTest): with config.db.connect() as conn: conn.execute( some_other_table.insert().from_select( - ['id', 'data', 'parent_id'], - select([some_table]) + ["id", "data", "parent_id"], select([some_table]) ) ) - cte = select([some_table]).where( - some_table.c.data.in_(["d2", "d3", "d4"]) - ).cte("some_cte") + cte = ( + select([some_table]) + .where(some_table.c.data.in_(["d2", "d3", "d4"])) + .cte("some_cte") + ) conn.execute( some_other_table.delete().where( - some_other_table.c.data == - select([cte.c.data]).where( - cte.c.id == some_other_table.c.id) + some_other_table.c.data + == select([cte.c.data]).where( + cte.c.id == some_other_table.c.id + ) ) ) eq_( conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), - [ - (1, "d1", None), (5, "d5", 3) - ] + [(1, "d1", None), (5, "d5", 3)], ) diff --git a/lib/sqlalchemy/testing/suite/test_ddl.py b/lib/sqlalchemy/testing/suite/test_ddl.py index 1d8010c8a..7c44388d4 100644 --- a/lib/sqlalchemy/testing/suite/test_ddl.py +++ b/lib/sqlalchemy/testing/suite/test_ddl.py @@ -1,5 +1,3 @@ - - from .. import fixtures, config, util from ..config import requirements from ..assertions import eq_ @@ -11,55 +9,47 @@ class TableDDLTest(fixtures.TestBase): __backend__ = True def _simple_fixture(self): - return Table('test_table', self.metadata, - Column('id', Integer, primary_key=True, - autoincrement=False), - Column('data', String(50)) - ) + return Table( + "test_table", + self.metadata, + Column("id", Integer, primary_key=True, autoincrement=False), + Column("data", String(50)), + ) def _underscore_fixture(self): - return Table('_test_table', self.metadata, - Column('id', Integer, primary_key=True, - autoincrement=False), - Column('_data', String(50)) - ) + return Table( + "_test_table", + self.metadata, + Column("id", Integer, primary_key=True, autoincrement=False), + Column("_data", String(50)), + ) def _simple_roundtrip(self, table): with config.db.begin() as conn: - conn.execute(table.insert().values((1, 'some data'))) + conn.execute(table.insert().values((1, "some data"))) result = conn.execute(table.select()) - eq_( - result.first(), - (1, 'some data') - ) + eq_(result.first(), (1, "some data")) @requirements.create_table @util.provide_metadata def test_create_table(self): table = self._simple_fixture() - table.create( - config.db, checkfirst=False - ) + table.create(config.db, checkfirst=False) self._simple_roundtrip(table) @requirements.drop_table @util.provide_metadata def test_drop_table(self): table = self._simple_fixture() - table.create( - config.db, checkfirst=False - ) - table.drop( - config.db, checkfirst=False - ) + table.create(config.db, checkfirst=False) + table.drop(config.db, checkfirst=False) @requirements.create_table @util.provide_metadata def test_underscore_names(self): table = self._underscore_fixture() - table.create( - config.db, checkfirst=False - ) + table.create(config.db, checkfirst=False) self._simple_roundtrip(table) -__all__ = ('TableDDLTest', ) + +__all__ = ("TableDDLTest",) diff --git a/lib/sqlalchemy/testing/suite/test_dialect.py b/lib/sqlalchemy/testing/suite/test_dialect.py index 2c5dd0e36..5e589f3b8 100644 --- a/lib/sqlalchemy/testing/suite/test_dialect.py +++ b/lib/sqlalchemy/testing/suite/test_dialect.py @@ -15,16 +15,19 @@ class ExceptionTest(fixtures.TablesTest): specific exceptions from real round trips, we need to be conservative. """ - run_deletes = 'each' + + run_deletes = "each" __backend__ = True @classmethod def define_tables(cls, metadata): - Table('manual_pk', metadata, - Column('id', Integer, primary_key=True, autoincrement=False), - Column('data', String(50)) - ) + Table( + "manual_pk", + metadata, + Column("id", Integer, primary_key=True, autoincrement=False), + Column("data", String(50)), + ) @requirements.duplicate_key_raises_integrity_error def test_integrity_error(self): @@ -33,15 +36,14 @@ class ExceptionTest(fixtures.TablesTest): trans = conn.begin() conn.execute( - self.tables.manual_pk.insert(), - {'id': 1, 'data': 'd1'} + self.tables.manual_pk.insert(), {"id": 1, "data": "d1"} ) assert_raises( exc.IntegrityError, conn.execute, self.tables.manual_pk.insert(), - {'id': 1, 'data': 'd1'} + {"id": 1, "data": "d1"}, ) trans.rollback() @@ -49,38 +51,39 @@ class ExceptionTest(fixtures.TablesTest): class AutocommitTest(fixtures.TablesTest): - run_deletes = 'each' + run_deletes = "each" - __requires__ = 'autocommit', + __requires__ = ("autocommit",) __backend__ = True @classmethod def define_tables(cls, metadata): - Table('some_table', metadata, - Column('id', Integer, primary_key=True, autoincrement=False), - Column('data', String(50)), - test_needs_acid=True - ) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True, autoincrement=False), + Column("data", String(50)), + test_needs_acid=True, + ) def _test_conn_autocommits(self, conn, autocommit): trans = conn.begin() conn.execute( - self.tables.some_table.insert(), - {"id": 1, "data": "some data"} + self.tables.some_table.insert(), {"id": 1, "data": "some data"} ) trans.rollback() eq_( conn.scalar(select([self.tables.some_table.c.id])), - 1 if autocommit else None + 1 if autocommit else None, ) conn.execute(self.tables.some_table.delete()) def test_autocommit_on(self): conn = config.db.connect() - c2 = conn.execution_options(isolation_level='AUTOCOMMIT') + c2 = conn.execution_options(isolation_level="AUTOCOMMIT") self._test_conn_autocommits(c2, True) conn.invalidate() self._test_conn_autocommits(conn, False) @@ -98,7 +101,7 @@ class EscapingTest(fixtures.TestBase): """ m = self.metadata - t = Table('t', m, Column('data', String(50))) + t = Table("t", m, Column("data", String(50))) t.create(config.db) with config.db.begin() as conn: conn.execute(t.insert(), dict(data="some % value")) @@ -107,14 +110,17 @@ class EscapingTest(fixtures.TestBase): eq_( conn.scalar( select([t.c.data]).where( - t.c.data == literal_column("'some % value'")) + t.c.data == literal_column("'some % value'") + ) ), - "some % value" + "some % value", ) eq_( conn.scalar( select([t.c.data]).where( - t.c.data == literal_column("'some %% other value'")) - ), "some %% other value" + t.c.data == literal_column("'some %% other value'") + ) + ), + "some %% other value", ) diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py index c0b6b18eb..6257451eb 100644 --- a/lib/sqlalchemy/testing/suite/test_insert.py +++ b/lib/sqlalchemy/testing/suite/test_insert.py @@ -10,53 +10,48 @@ from ..schema import Table, Column class LastrowidTest(fixtures.TablesTest): - run_deletes = 'each' + run_deletes = "each" __backend__ = True - __requires__ = 'implements_get_lastrowid', 'autoincrement_insert' + __requires__ = "implements_get_lastrowid", "autoincrement_insert" __engine_options__ = {"implicit_returning": False} @classmethod def define_tables(cls, metadata): - Table('autoinc_pk', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(50)) - ) + Table( + "autoinc_pk", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(50)), + ) - Table('manual_pk', metadata, - Column('id', Integer, primary_key=True, autoincrement=False), - Column('data', String(50)) - ) + Table( + "manual_pk", + metadata, + Column("id", Integer, primary_key=True, autoincrement=False), + Column("data", String(50)), + ) def _assert_round_trip(self, table, conn): row = conn.execute(table.select()).first() - eq_( - row, - (config.db.dialect.default_sequence_base, "some data") - ) + eq_(row, (config.db.dialect.default_sequence_base, "some data")) def test_autoincrement_on_insert(self): - config.db.execute( - self.tables.autoinc_pk.insert(), - data="some data" - ) + config.db.execute(self.tables.autoinc_pk.insert(), data="some data") self._assert_round_trip(self.tables.autoinc_pk, config.db) def test_last_inserted_id(self): r = config.db.execute( - self.tables.autoinc_pk.insert(), - data="some data" + self.tables.autoinc_pk.insert(), data="some data" ) pk = config.db.scalar(select([self.tables.autoinc_pk.c.id])) - eq_( - r.inserted_primary_key, - [pk] - ) + eq_(r.inserted_primary_key, [pk]) # failed on pypy1.9 but seems to be OK on pypy 2.1 # @exclusions.fails_if(lambda: util.pypy, @@ -65,50 +60,57 @@ class LastrowidTest(fixtures.TablesTest): @requirements.dbapi_lastrowid def test_native_lastrowid_autoinc(self): r = config.db.execute( - self.tables.autoinc_pk.insert(), - data="some data" + self.tables.autoinc_pk.insert(), data="some data" ) lastrowid = r.lastrowid pk = config.db.scalar(select([self.tables.autoinc_pk.c.id])) - eq_( - lastrowid, pk - ) + eq_(lastrowid, pk) class InsertBehaviorTest(fixtures.TablesTest): - run_deletes = 'each' + run_deletes = "each" __backend__ = True @classmethod def define_tables(cls, metadata): - Table('autoinc_pk', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(50)) - ) - Table('manual_pk', metadata, - Column('id', Integer, primary_key=True, autoincrement=False), - Column('data', String(50)) - ) - Table('includes_defaults', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(50)), - Column('x', Integer, default=5), - Column('y', Integer, - default=literal_column("2", type_=Integer) + literal(2))) + Table( + "autoinc_pk", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(50)), + ) + Table( + "manual_pk", + metadata, + Column("id", Integer, primary_key=True, autoincrement=False), + Column("data", String(50)), + ) + Table( + "includes_defaults", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(50)), + Column("x", Integer, default=5), + Column( + "y", + Integer, + default=literal_column("2", type_=Integer) + literal(2), + ), + ) def test_autoclose_on_insert(self): if requirements.returning.enabled: engine = engines.testing_engine( - options={'implicit_returning': False}) + options={"implicit_returning": False} + ) else: engine = config.db - r = engine.execute( - self.tables.autoinc_pk.insert(), - data="some data" - ) + r = engine.execute(self.tables.autoinc_pk.insert(), data="some data") assert r._soft_closed assert not r.closed assert r.is_insert @@ -117,8 +119,7 @@ class InsertBehaviorTest(fixtures.TablesTest): @requirements.returning def test_autoclose_on_insert_implicit_returning(self): r = config.db.execute( - self.tables.autoinc_pk.insert(), - data="some data" + self.tables.autoinc_pk.insert(), data="some data" ) assert r._soft_closed assert not r.closed @@ -127,15 +128,14 @@ class InsertBehaviorTest(fixtures.TablesTest): @requirements.empty_inserts def test_empty_insert(self): - r = config.db.execute( - self.tables.autoinc_pk.insert(), - ) + r = config.db.execute(self.tables.autoinc_pk.insert()) assert r._soft_closed assert not r.closed r = config.db.execute( - self.tables.autoinc_pk.select(). - where(self.tables.autoinc_pk.c.id != None) + self.tables.autoinc_pk.select().where( + self.tables.autoinc_pk.c.id != None + ) ) assert len(r.fetchall()) @@ -150,15 +150,15 @@ class InsertBehaviorTest(fixtures.TablesTest): dict(id=1, data="data1"), dict(id=2, data="data2"), dict(id=3, data="data3"), - ] + ], ) result = config.db.execute( - dest_table.insert(). - from_select( + dest_table.insert().from_select( ("data",), - select([src_table.c.data]). - where(src_table.c.data.in_(["data2", "data3"])) + select([src_table.c.data]).where( + src_table.c.data.in_(["data2", "data3"]) + ), ) ) @@ -167,7 +167,7 @@ class InsertBehaviorTest(fixtures.TablesTest): result = config.db.execute( select([dest_table.c.data]).order_by(dest_table.c.data) ) - eq_(result.fetchall(), [("data2", ), ("data3", )]) + eq_(result.fetchall(), [("data2",), ("data3",)]) @requirements.insert_from_select def test_insert_from_select_autoinc_no_rows(self): @@ -175,11 +175,11 @@ class InsertBehaviorTest(fixtures.TablesTest): dest_table = self.tables.autoinc_pk result = config.db.execute( - dest_table.insert(). - from_select( + dest_table.insert().from_select( ("data",), - select([src_table.c.data]). - where(src_table.c.data.in_(["data2", "data3"])) + select([src_table.c.data]).where( + src_table.c.data.in_(["data2", "data3"]) + ), ) ) eq_(result.inserted_primary_key, [None]) @@ -199,23 +199,23 @@ class InsertBehaviorTest(fixtures.TablesTest): dict(id=1, data="data1"), dict(id=2, data="data2"), dict(id=3, data="data3"), - ] + ], ) config.db.execute( - table.insert(inline=True). - from_select(("id", "data",), - select([table.c.id + 5, table.c.data]). - where(table.c.data.in_(["data2", "data3"])) - ), + table.insert(inline=True).from_select( + ("id", "data"), + select([table.c.id + 5, table.c.data]).where( + table.c.data.in_(["data2", "data3"]) + ), + ) ) eq_( config.db.execute( select([table.c.data]).order_by(table.c.data) ).fetchall(), - [("data1", ), ("data2", ), ("data2", ), - ("data3", ), ("data3", )] + [("data1",), ("data2",), ("data2",), ("data3",), ("data3",)], ) @requirements.insert_from_select @@ -227,56 +227,60 @@ class InsertBehaviorTest(fixtures.TablesTest): dict(id=1, data="data1"), dict(id=2, data="data2"), dict(id=3, data="data3"), - ] + ], ) config.db.execute( - table.insert(inline=True). - from_select(("id", "data",), - select([table.c.id + 5, table.c.data]). - where(table.c.data.in_(["data2", "data3"])) - ), + table.insert(inline=True).from_select( + ("id", "data"), + select([table.c.id + 5, table.c.data]).where( + table.c.data.in_(["data2", "data3"]) + ), + ) ) eq_( config.db.execute( select([table]).order_by(table.c.data, table.c.id) ).fetchall(), - [(1, 'data1', 5, 4), (2, 'data2', 5, 4), - (7, 'data2', 5, 4), (3, 'data3', 5, 4), (8, 'data3', 5, 4)] + [ + (1, "data1", 5, 4), + (2, "data2", 5, 4), + (7, "data2", 5, 4), + (3, "data3", 5, 4), + (8, "data3", 5, 4), + ], ) class ReturningTest(fixtures.TablesTest): - run_create_tables = 'each' - __requires__ = 'returning', 'autoincrement_insert' + run_create_tables = "each" + __requires__ = "returning", "autoincrement_insert" __backend__ = True __engine_options__ = {"implicit_returning": True} def _assert_round_trip(self, table, conn): row = conn.execute(table.select()).first() - eq_( - row, - (config.db.dialect.default_sequence_base, "some data") - ) + eq_(row, (config.db.dialect.default_sequence_base, "some data")) @classmethod def define_tables(cls, metadata): - Table('autoinc_pk', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('data', String(50)) - ) + Table( + "autoinc_pk", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("data", String(50)), + ) @requirements.fetch_rows_post_commit def test_explicit_returning_pk_autocommit(self): engine = config.db table = self.tables.autoinc_pk r = engine.execute( - table.insert().returning( - table.c.id), - data="some data" + table.insert().returning(table.c.id), data="some data" ) pk = r.first()[0] fetched_pk = config.db.scalar(select([table.c.id])) @@ -287,9 +291,7 @@ class ReturningTest(fixtures.TablesTest): table = self.tables.autoinc_pk with engine.begin() as conn: r = conn.execute( - table.insert().returning( - table.c.id), - data="some data" + table.insert().returning(table.c.id), data="some data" ) pk = r.first()[0] fetched_pk = config.db.scalar(select([table.c.id])) @@ -297,23 +299,16 @@ class ReturningTest(fixtures.TablesTest): def test_autoincrement_on_insert_implcit_returning(self): - config.db.execute( - self.tables.autoinc_pk.insert(), - data="some data" - ) + config.db.execute(self.tables.autoinc_pk.insert(), data="some data") self._assert_round_trip(self.tables.autoinc_pk, config.db) def test_last_inserted_id_implicit_returning(self): r = config.db.execute( - self.tables.autoinc_pk.insert(), - data="some data" + self.tables.autoinc_pk.insert(), data="some data" ) pk = config.db.scalar(select([self.tables.autoinc_pk.c.id])) - eq_( - r.inserted_primary_key, - [pk] - ) + eq_(r.inserted_primary_key, [pk]) -__all__ = ('LastrowidTest', 'InsertBehaviorTest', 'ReturningTest') +__all__ = ("LastrowidTest", "InsertBehaviorTest", "ReturningTest") diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py index 00a5aac01..bfed5f1ab 100644 --- a/lib/sqlalchemy/testing/suite/test_reflection.py +++ b/lib/sqlalchemy/testing/suite/test_reflection.py @@ -1,5 +1,3 @@ - - import sqlalchemy as sa from sqlalchemy import exc as sa_exc from sqlalchemy import types as sql_types @@ -26,10 +24,12 @@ class HasTableTest(fixtures.TablesTest): @classmethod def define_tables(cls, metadata): - Table('test_table', metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)) - ) + Table( + "test_table", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), + ) def test_has_table(self): with config.db.begin() as conn: @@ -46,8 +46,10 @@ class ComponentReflectionTest(fixtures.TablesTest): def setup_bind(cls): if config.requirements.independent_connections.enabled: from sqlalchemy import pool + return engines.testing_engine( - options=dict(poolclass=pool.StaticPool)) + options=dict(poolclass=pool.StaticPool) + ) else: return config.db @@ -65,86 +67,109 @@ class ComponentReflectionTest(fixtures.TablesTest): schema_prefix = "" if testing.requires.self_referential_foreign_keys.enabled: - users = Table('users', metadata, - Column('user_id', sa.INT, primary_key=True), - Column('test1', sa.CHAR(5), nullable=False), - Column('test2', sa.Float(5), nullable=False), - Column('parent_user_id', sa.Integer, - sa.ForeignKey('%susers.user_id' % - schema_prefix, - name='user_id_fk')), - schema=schema, - test_needs_fk=True, - ) + users = Table( + "users", + metadata, + Column("user_id", sa.INT, primary_key=True), + Column("test1", sa.CHAR(5), nullable=False), + Column("test2", sa.Float(5), nullable=False), + Column( + "parent_user_id", + sa.Integer, + sa.ForeignKey( + "%susers.user_id" % schema_prefix, name="user_id_fk" + ), + ), + schema=schema, + test_needs_fk=True, + ) else: - users = Table('users', metadata, - Column('user_id', sa.INT, primary_key=True), - Column('test1', sa.CHAR(5), nullable=False), - Column('test2', sa.Float(5), nullable=False), - schema=schema, - test_needs_fk=True, - ) - - Table("dingalings", metadata, - Column('dingaling_id', sa.Integer, primary_key=True), - Column('address_id', sa.Integer, - sa.ForeignKey('%semail_addresses.address_id' % - schema_prefix)), - Column('data', sa.String(30)), - schema=schema, - test_needs_fk=True, - ) - Table('email_addresses', metadata, - Column('address_id', sa.Integer), - Column('remote_user_id', sa.Integer, - sa.ForeignKey(users.c.user_id)), - Column('email_address', sa.String(20)), - sa.PrimaryKeyConstraint('address_id', name='email_ad_pk'), - schema=schema, - test_needs_fk=True, - ) - Table('comment_test', metadata, - Column('id', sa.Integer, primary_key=True, comment='id comment'), - Column('data', sa.String(20), comment='data % comment'), - Column( - 'd2', sa.String(20), - comment=r"""Comment types type speedily ' " \ '' Fun!"""), - schema=schema, - comment=r"""the test % ' " \ table comment""") + users = Table( + "users", + metadata, + Column("user_id", sa.INT, primary_key=True), + Column("test1", sa.CHAR(5), nullable=False), + Column("test2", sa.Float(5), nullable=False), + schema=schema, + test_needs_fk=True, + ) + + Table( + "dingalings", + metadata, + Column("dingaling_id", sa.Integer, primary_key=True), + Column( + "address_id", + sa.Integer, + sa.ForeignKey("%semail_addresses.address_id" % schema_prefix), + ), + Column("data", sa.String(30)), + schema=schema, + test_needs_fk=True, + ) + Table( + "email_addresses", + metadata, + Column("address_id", sa.Integer), + Column( + "remote_user_id", sa.Integer, sa.ForeignKey(users.c.user_id) + ), + Column("email_address", sa.String(20)), + sa.PrimaryKeyConstraint("address_id", name="email_ad_pk"), + schema=schema, + test_needs_fk=True, + ) + Table( + "comment_test", + metadata, + Column("id", sa.Integer, primary_key=True, comment="id comment"), + Column("data", sa.String(20), comment="data % comment"), + Column( + "d2", + sa.String(20), + comment=r"""Comment types type speedily ' " \ '' Fun!""", + ), + schema=schema, + comment=r"""the test % ' " \ table comment""", + ) if testing.requires.cross_schema_fk_reflection.enabled: if schema is None: Table( - 'local_table', metadata, - Column('id', sa.Integer, primary_key=True), - Column('data', sa.String(20)), + "local_table", + metadata, + Column("id", sa.Integer, primary_key=True), + Column("data", sa.String(20)), Column( - 'remote_id', + "remote_id", ForeignKey( - '%s.remote_table_2.id' % - testing.config.test_schema) + "%s.remote_table_2.id" % testing.config.test_schema + ), ), test_needs_fk=True, - schema=config.db.dialect.default_schema_name + schema=config.db.dialect.default_schema_name, ) else: Table( - 'remote_table', metadata, - Column('id', sa.Integer, primary_key=True), + "remote_table", + metadata, + Column("id", sa.Integer, primary_key=True), Column( - 'local_id', + "local_id", ForeignKey( - '%s.local_table.id' % - config.db.dialect.default_schema_name) + "%s.local_table.id" + % config.db.dialect.default_schema_name + ), ), - Column('data', sa.String(20)), + Column("data", sa.String(20)), schema=schema, test_needs_fk=True, ) Table( - 'remote_table_2', metadata, - Column('id', sa.Integer, primary_key=True), - Column('data', sa.String(20)), + "remote_table_2", + metadata, + Column("id", sa.Integer, primary_key=True), + Column("data", sa.String(20)), schema=schema, test_needs_fk=True, ) @@ -155,19 +180,21 @@ class ComponentReflectionTest(fixtures.TablesTest): if not schema: # test_needs_fk is at the moment to force MySQL InnoDB noncol_idx_test_nopk = Table( - 'noncol_idx_test_nopk', metadata, - Column('q', sa.String(5)), + "noncol_idx_test_nopk", + metadata, + Column("q", sa.String(5)), test_needs_fk=True, ) noncol_idx_test_pk = Table( - 'noncol_idx_test_pk', metadata, - Column('id', sa.Integer, primary_key=True), - Column('q', sa.String(5)), + "noncol_idx_test_pk", + metadata, + Column("id", sa.Integer, primary_key=True), + Column("q", sa.String(5)), test_needs_fk=True, ) - Index('noncol_idx_nopk', noncol_idx_test_nopk.c.q.desc()) - Index('noncol_idx_pk', noncol_idx_test_pk.c.q.desc()) + Index("noncol_idx_nopk", noncol_idx_test_nopk.c.q.desc()) + Index("noncol_idx_pk", noncol_idx_test_pk.c.q.desc()) if testing.requires.view_column_reflection.enabled: cls.define_views(metadata, schema) @@ -180,34 +207,35 @@ class ComponentReflectionTest(fixtures.TablesTest): # temp table fixture if testing.against("oracle"): kw = { - 'prefixes': ["GLOBAL TEMPORARY"], - 'oracle_on_commit': 'PRESERVE ROWS' + "prefixes": ["GLOBAL TEMPORARY"], + "oracle_on_commit": "PRESERVE ROWS", } else: - kw = { - 'prefixes': ["TEMPORARY"], - } + kw = {"prefixes": ["TEMPORARY"]} user_tmp = Table( - "user_tmp", metadata, + "user_tmp", + metadata, Column("id", sa.INT, primary_key=True), - Column('name', sa.VARCHAR(50)), - Column('foo', sa.INT), - sa.UniqueConstraint('name', name='user_tmp_uq'), + Column("name", sa.VARCHAR(50)), + Column("foo", sa.INT), + sa.UniqueConstraint("name", name="user_tmp_uq"), sa.Index("user_tmp_ix", "foo"), **kw ) - if testing.requires.view_reflection.enabled and \ - testing.requires.temporary_views.enabled: - event.listen( - user_tmp, "after_create", - DDL("create temporary view user_tmp_v as " - "select * from user_tmp") - ) + if ( + testing.requires.view_reflection.enabled + and testing.requires.temporary_views.enabled + ): event.listen( - user_tmp, "before_drop", - DDL("drop view user_tmp_v") + user_tmp, + "after_create", + DDL( + "create temporary view user_tmp_v as " + "select * from user_tmp" + ), ) + event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v")) @classmethod def define_index(cls, metadata, users): @@ -216,23 +244,19 @@ class ComponentReflectionTest(fixtures.TablesTest): @classmethod def define_views(cls, metadata, schema): - for table_name in ('users', 'email_addresses'): + for table_name in ("users", "email_addresses"): fullname = table_name if schema: fullname = "%s.%s" % (schema, table_name) - view_name = fullname + '_v' + view_name = fullname + "_v" query = "CREATE VIEW %s AS SELECT * FROM %s" % ( - view_name, fullname) - - event.listen( - metadata, - "after_create", - DDL(query) + view_name, + fullname, ) + + event.listen(metadata, "after_create", DDL(query)) event.listen( - metadata, - "before_drop", - DDL("DROP VIEW %s" % view_name) + metadata, "before_drop", DDL("DROP VIEW %s" % view_name) ) @testing.requires.schema_reflection @@ -244,9 +268,9 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.requires.schema_reflection def test_dialect_initialize(self): engine = engines.testing_engine() - assert not hasattr(engine.dialect, 'default_schema_name') + assert not hasattr(engine.dialect, "default_schema_name") inspect(engine) - assert hasattr(engine.dialect, 'default_schema_name') + assert hasattr(engine.dialect, "default_schema_name") @testing.requires.schema_reflection def test_get_default_schema_name(self): @@ -254,40 +278,49 @@ class ComponentReflectionTest(fixtures.TablesTest): eq_(insp.default_schema_name, testing.db.dialect.default_schema_name) @testing.provide_metadata - def _test_get_table_names(self, schema=None, table_type='table', - order_by=None): + def _test_get_table_names( + self, schema=None, table_type="table", order_by=None + ): _ignore_tables = [ - 'comment_test', 'noncol_idx_test_pk', 'noncol_idx_test_nopk', - 'local_table', 'remote_table', 'remote_table_2' + "comment_test", + "noncol_idx_test_pk", + "noncol_idx_test_nopk", + "local_table", + "remote_table", + "remote_table_2", ] meta = self.metadata - users, addresses, dingalings = self.tables.users, \ - self.tables.email_addresses, self.tables.dingalings + users, addresses, dingalings = ( + self.tables.users, + self.tables.email_addresses, + self.tables.dingalings, + ) insp = inspect(meta.bind) - if table_type == 'view': + if table_type == "view": table_names = insp.get_view_names(schema) table_names.sort() - answer = ['email_addresses_v', 'users_v'] + answer = ["email_addresses_v", "users_v"] eq_(sorted(table_names), answer) else: table_names = [ - t for t in insp.get_table_names( - schema, - order_by=order_by) if t not in _ignore_tables] + t + for t in insp.get_table_names(schema, order_by=order_by) + if t not in _ignore_tables + ] - if order_by == 'foreign_key': - answer = ['users', 'email_addresses', 'dingalings'] + if order_by == "foreign_key": + answer = ["users", "email_addresses", "dingalings"] eq_(table_names, answer) else: - answer = ['dingalings', 'email_addresses', 'users'] + answer = ["dingalings", "email_addresses", "users"] eq_(sorted(table_names), answer) @testing.requires.temp_table_names def test_get_temp_table_names(self): insp = inspect(self.bind) temp_table_names = insp.get_temp_table_names() - eq_(sorted(temp_table_names), ['user_tmp']) + eq_(sorted(temp_table_names), ["user_tmp"]) @testing.requires.view_reflection @testing.requires.temp_table_names @@ -295,7 +328,7 @@ class ComponentReflectionTest(fixtures.TablesTest): def test_get_temp_view_names(self): insp = inspect(self.bind) temp_table_names = insp.get_temp_view_names() - eq_(sorted(temp_table_names), ['user_tmp_v']) + eq_(sorted(temp_table_names), ["user_tmp_v"]) @testing.requires.table_reflection def test_get_table_names(self): @@ -304,7 +337,7 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.requires.table_reflection @testing.requires.foreign_key_constraint_reflection def test_get_table_names_fks(self): - self._test_get_table_names(order_by='foreign_key') + self._test_get_table_names(order_by="foreign_key") @testing.requires.comment_reflection def test_get_comments(self): @@ -320,26 +353,24 @@ class ComponentReflectionTest(fixtures.TablesTest): eq_( insp.get_table_comment("comment_test", schema=schema), - {"text": r"""the test % ' " \ table comment"""} + {"text": r"""the test % ' " \ table comment"""}, ) - eq_( - insp.get_table_comment("users", schema=schema), - {"text": None} - ) + eq_(insp.get_table_comment("users", schema=schema), {"text": None}) eq_( [ - {"name": rec['name'], "comment": rec['comment']} - for rec in - insp.get_columns("comment_test", schema=schema) + {"name": rec["name"], "comment": rec["comment"]} + for rec in insp.get_columns("comment_test", schema=schema) ], [ - {'comment': 'id comment', 'name': 'id'}, - {'comment': 'data % comment', 'name': 'data'}, - {'comment': r"""Comment types type speedily ' " \ '' Fun!""", - 'name': 'd2'} - ] + {"comment": "id comment", "name": "id"}, + {"comment": "data % comment", "name": "data"}, + { + "comment": r"""Comment types type speedily ' " \ '' Fun!""", + "name": "d2", + }, + ], ) @testing.requires.table_reflection @@ -349,30 +380,33 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.requires.view_column_reflection def test_get_view_names(self): - self._test_get_table_names(table_type='view') + self._test_get_table_names(table_type="view") @testing.requires.view_column_reflection @testing.requires.schemas def test_get_view_names_with_schema(self): self._test_get_table_names( - testing.config.test_schema, table_type='view') + testing.config.test_schema, table_type="view" + ) @testing.requires.table_reflection @testing.requires.view_column_reflection def test_get_tables_and_views(self): self._test_get_table_names() - self._test_get_table_names(table_type='view') + self._test_get_table_names(table_type="view") - def _test_get_columns(self, schema=None, table_type='table'): + def _test_get_columns(self, schema=None, table_type="table"): meta = MetaData(testing.db) - users, addresses, dingalings = self.tables.users, \ - self.tables.email_addresses, self.tables.dingalings - table_names = ['users', 'email_addresses'] - if table_type == 'view': - table_names = ['users_v', 'email_addresses_v'] + users, addresses, dingalings = ( + self.tables.users, + self.tables.email_addresses, + self.tables.dingalings, + ) + table_names = ["users", "email_addresses"] + if table_type == "view": + table_names = ["users_v", "email_addresses_v"] insp = inspect(meta.bind) - for table_name, table in zip(table_names, (users, - addresses)): + for table_name, table in zip(table_names, (users, addresses)): schema_name = schema cols = insp.get_columns(table_name, schema=schema_name) self.assert_(len(cols) > 0, len(cols)) @@ -380,36 +414,46 @@ class ComponentReflectionTest(fixtures.TablesTest): # should be in order for i, col in enumerate(table.columns): - eq_(col.name, cols[i]['name']) - ctype = cols[i]['type'].__class__ + eq_(col.name, cols[i]["name"]) + ctype = cols[i]["type"].__class__ ctype_def = col.type if isinstance(ctype_def, sa.types.TypeEngine): ctype_def = ctype_def.__class__ # Oracle returns Date for DateTime. - if testing.against('oracle') and ctype_def \ - in (sql_types.Date, sql_types.DateTime): + if testing.against("oracle") and ctype_def in ( + sql_types.Date, + sql_types.DateTime, + ): ctype_def = sql_types.Date # assert that the desired type and return type share # a base within one of the generic types. - self.assert_(len(set(ctype.__mro__). - intersection(ctype_def.__mro__). - intersection([ - sql_types.Integer, - sql_types.Numeric, - sql_types.DateTime, - sql_types.Date, - sql_types.Time, - sql_types.String, - sql_types._Binary, - ])) > 0, '%s(%s), %s(%s)' % - (col.name, col.type, cols[i]['name'], ctype)) + self.assert_( + len( + set(ctype.__mro__) + .intersection(ctype_def.__mro__) + .intersection( + [ + sql_types.Integer, + sql_types.Numeric, + sql_types.DateTime, + sql_types.Date, + sql_types.Time, + sql_types.String, + sql_types._Binary, + ] + ) + ) + > 0, + "%s(%s), %s(%s)" + % (col.name, col.type, cols[i]["name"], ctype), + ) if not col.primary_key: - assert cols[i]['default'] is None + assert cols[i]["default"] is None @testing.requires.table_reflection def test_get_columns(self): @@ -417,24 +461,20 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.provide_metadata def _type_round_trip(self, *types): - t = Table('t', self.metadata, - *[ - Column('t%d' % i, type_) - for i, type_ in enumerate(types) - ] - ) + t = Table( + "t", + self.metadata, + *[Column("t%d" % i, type_) for i, type_ in enumerate(types)] + ) t.create() return [ - c['type'] for c in - inspect(self.metadata.bind).get_columns('t') + c["type"] for c in inspect(self.metadata.bind).get_columns("t") ] @testing.requires.table_reflection def test_numeric_reflection(self): - for typ in self._type_round_trip( - sql_types.Numeric(18, 5), - ): + for typ in self._type_round_trip(sql_types.Numeric(18, 5)): assert isinstance(typ, sql_types.Numeric) eq_(typ.precision, 18) eq_(typ.scale, 5) @@ -448,16 +488,19 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.requires.table_reflection @testing.provide_metadata def test_nullable_reflection(self): - t = Table('t', self.metadata, - Column('a', Integer, nullable=True), - Column('b', Integer, nullable=False)) + t = Table( + "t", + self.metadata, + Column("a", Integer, nullable=True), + Column("b", Integer, nullable=False), + ) t.create() eq_( dict( - (col['name'], col['nullable']) - for col in inspect(self.metadata.bind).get_columns('t') + (col["name"], col["nullable"]) + for col in inspect(self.metadata.bind).get_columns("t") ), - {"a": True, "b": False} + {"a": True, "b": False}, ) @testing.requires.table_reflection @@ -470,32 +513,30 @@ class ComponentReflectionTest(fixtures.TablesTest): meta = MetaData(self.bind) user_tmp = self.tables.user_tmp insp = inspect(meta.bind) - cols = insp.get_columns('user_tmp') + cols = insp.get_columns("user_tmp") self.assert_(len(cols) > 0, len(cols)) for i, col in enumerate(user_tmp.columns): - eq_(col.name, cols[i]['name']) + eq_(col.name, cols[i]["name"]) @testing.requires.temp_table_reflection @testing.requires.view_column_reflection @testing.requires.temporary_views def test_get_temp_view_columns(self): insp = inspect(self.bind) - cols = insp.get_columns('user_tmp_v') - eq_( - [col['name'] for col in cols], - ['id', 'name', 'foo'] - ) + cols = insp.get_columns("user_tmp_v") + eq_([col["name"] for col in cols], ["id", "name", "foo"]) @testing.requires.view_column_reflection def test_get_view_columns(self): - self._test_get_columns(table_type='view') + self._test_get_columns(table_type="view") @testing.requires.view_column_reflection @testing.requires.schemas def test_get_view_columns_with_schema(self): self._test_get_columns( - schema=testing.config.test_schema, table_type='view') + schema=testing.config.test_schema, table_type="view" + ) @testing.provide_metadata def _test_get_pk_constraint(self, schema=None): @@ -504,15 +545,15 @@ class ComponentReflectionTest(fixtures.TablesTest): insp = inspect(meta.bind) users_cons = insp.get_pk_constraint(users.name, schema=schema) - users_pkeys = users_cons['constrained_columns'] - eq_(users_pkeys, ['user_id']) + users_pkeys = users_cons["constrained_columns"] + eq_(users_pkeys, ["user_id"]) addr_cons = insp.get_pk_constraint(addresses.name, schema=schema) - addr_pkeys = addr_cons['constrained_columns'] - eq_(addr_pkeys, ['address_id']) + addr_pkeys = addr_cons["constrained_columns"] + eq_(addr_pkeys, ["address_id"]) with testing.requires.reflects_pk_names.fail_if(): - eq_(addr_cons['name'], 'email_ad_pk') + eq_(addr_cons["name"], "email_ad_pk") @testing.requires.primary_key_constraint_reflection def test_get_pk_constraint(self): @@ -534,44 +575,46 @@ class ComponentReflectionTest(fixtures.TablesTest): sa_exc.SADeprecationWarning, "Call to deprecated method get_primary_keys." " Use get_pk_constraint instead.", - insp.get_primary_keys, users.name + insp.get_primary_keys, + users.name, ) @testing.provide_metadata def _test_get_foreign_keys(self, schema=None): meta = self.metadata - users, addresses, dingalings = self.tables.users, \ - self.tables.email_addresses, self.tables.dingalings + users, addresses, dingalings = ( + self.tables.users, + self.tables.email_addresses, + self.tables.dingalings, + ) insp = inspect(meta.bind) expected_schema = schema # users if testing.requires.self_referential_foreign_keys.enabled: - users_fkeys = insp.get_foreign_keys(users.name, - schema=schema) + users_fkeys = insp.get_foreign_keys(users.name, schema=schema) fkey1 = users_fkeys[0] with testing.requires.named_constraints.fail_if(): - eq_(fkey1['name'], "user_id_fk") + eq_(fkey1["name"], "user_id_fk") - eq_(fkey1['referred_schema'], expected_schema) - eq_(fkey1['referred_table'], users.name) - eq_(fkey1['referred_columns'], ['user_id', ]) + eq_(fkey1["referred_schema"], expected_schema) + eq_(fkey1["referred_table"], users.name) + eq_(fkey1["referred_columns"], ["user_id"]) if testing.requires.self_referential_foreign_keys.enabled: - eq_(fkey1['constrained_columns'], ['parent_user_id']) + eq_(fkey1["constrained_columns"], ["parent_user_id"]) # addresses - addr_fkeys = insp.get_foreign_keys(addresses.name, - schema=schema) + addr_fkeys = insp.get_foreign_keys(addresses.name, schema=schema) fkey1 = addr_fkeys[0] with testing.requires.implicitly_named_constraints.fail_if(): - self.assert_(fkey1['name'] is not None) + self.assert_(fkey1["name"] is not None) - eq_(fkey1['referred_schema'], expected_schema) - eq_(fkey1['referred_table'], users.name) - eq_(fkey1['referred_columns'], ['user_id', ]) - eq_(fkey1['constrained_columns'], ['remote_user_id']) + eq_(fkey1["referred_schema"], expected_schema) + eq_(fkey1["referred_table"], users.name) + eq_(fkey1["referred_columns"], ["user_id"]) + eq_(fkey1["constrained_columns"], ["remote_user_id"]) @testing.requires.foreign_key_constraint_reflection def test_get_foreign_keys(self): @@ -586,9 +629,9 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.requires.schemas def test_get_inter_schema_foreign_keys(self): local_table, remote_table, remote_table_2 = self.tables( - '%s.local_table' % testing.db.dialect.default_schema_name, - '%s.remote_table' % testing.config.test_schema, - '%s.remote_table_2' % testing.config.test_schema + "%s.local_table" % testing.db.dialect.default_schema_name, + "%s.remote_table" % testing.config.test_schema, + "%s.remote_table_2" % testing.config.test_schema, ) insp = inspect(config.db) @@ -597,25 +640,25 @@ class ComponentReflectionTest(fixtures.TablesTest): eq_(len(local_fkeys), 1) fkey1 = local_fkeys[0] - eq_(fkey1['referred_schema'], testing.config.test_schema) - eq_(fkey1['referred_table'], remote_table_2.name) - eq_(fkey1['referred_columns'], ['id', ]) - eq_(fkey1['constrained_columns'], ['remote_id']) + eq_(fkey1["referred_schema"], testing.config.test_schema) + eq_(fkey1["referred_table"], remote_table_2.name) + eq_(fkey1["referred_columns"], ["id"]) + eq_(fkey1["constrained_columns"], ["remote_id"]) remote_fkeys = insp.get_foreign_keys( - remote_table.name, schema=testing.config.test_schema) + remote_table.name, schema=testing.config.test_schema + ) eq_(len(remote_fkeys), 1) fkey2 = remote_fkeys[0] - assert fkey2['referred_schema'] in ( + assert fkey2["referred_schema"] in ( None, - testing.db.dialect.default_schema_name + testing.db.dialect.default_schema_name, ) - eq_(fkey2['referred_table'], local_table.name) - eq_(fkey2['referred_columns'], ['id', ]) - eq_(fkey2['constrained_columns'], ['local_id']) - + eq_(fkey2["referred_table"], local_table.name) + eq_(fkey2["referred_columns"], ["id"]) + eq_(fkey2["constrained_columns"], ["local_id"]) @testing.requires.foreign_key_constraint_option_reflection_ondelete def test_get_foreign_key_options_ondelete(self): @@ -630,26 +673,32 @@ class ComponentReflectionTest(fixtures.TablesTest): meta = self.metadata Table( - 'x', meta, - Column('id', Integer, primary_key=True), - test_needs_fk=True - ) - - Table('table', meta, - Column('id', Integer, primary_key=True), - Column('x_id', Integer, sa.ForeignKey('x.id', name='xid')), - Column('test', String(10)), - test_needs_fk=True) - - Table('user', meta, - Column('id', Integer, primary_key=True), - Column('name', String(50), nullable=False), - Column('tid', Integer), - sa.ForeignKeyConstraint( - ['tid'], ['table.id'], - name='myfk', - **options), - test_needs_fk=True) + "x", + meta, + Column("id", Integer, primary_key=True), + test_needs_fk=True, + ) + + Table( + "table", + meta, + Column("id", Integer, primary_key=True), + Column("x_id", Integer, sa.ForeignKey("x.id", name="xid")), + Column("test", String(10)), + test_needs_fk=True, + ) + + Table( + "user", + meta, + Column("id", Integer, primary_key=True), + Column("name", String(50), nullable=False), + Column("tid", Integer), + sa.ForeignKeyConstraint( + ["tid"], ["table.id"], name="myfk", **options + ), + test_needs_fk=True, + ) meta.create_all() @@ -657,49 +706,44 @@ class ComponentReflectionTest(fixtures.TablesTest): # test 'options' is always present for a backend # that can reflect these, since alembic looks for this - opts = insp.get_foreign_keys('table')[0]['options'] + opts = insp.get_foreign_keys("table")[0]["options"] - eq_( - dict( - (k, opts[k]) - for k in opts if opts[k] - ), - {} - ) + eq_(dict((k, opts[k]) for k in opts if opts[k]), {}) - opts = insp.get_foreign_keys('user')[0]['options'] - eq_( - dict( - (k, opts[k]) - for k in opts if opts[k] - ), - options - ) + opts = insp.get_foreign_keys("user")[0]["options"] + eq_(dict((k, opts[k]) for k in opts if opts[k]), options) def _assert_insp_indexes(self, indexes, expected_indexes): - index_names = [d['name'] for d in indexes] + index_names = [d["name"] for d in indexes] for e_index in expected_indexes: - assert e_index['name'] in index_names - index = indexes[index_names.index(e_index['name'])] + assert e_index["name"] in index_names + index = indexes[index_names.index(e_index["name"])] for key in e_index: eq_(e_index[key], index[key]) @testing.provide_metadata def _test_get_indexes(self, schema=None): meta = self.metadata - users, addresses, dingalings = self.tables.users, \ - self.tables.email_addresses, self.tables.dingalings + users, addresses, dingalings = ( + self.tables.users, + self.tables.email_addresses, + self.tables.dingalings, + ) # The database may decide to create indexes for foreign keys, etc. # so there may be more indexes than expected. insp = inspect(meta.bind) - indexes = insp.get_indexes('users', schema=schema) + indexes = insp.get_indexes("users", schema=schema) expected_indexes = [ - {'unique': False, - 'column_names': ['test1', 'test2'], - 'name': 'users_t_idx'}, - {'unique': False, - 'column_names': ['user_id', 'test2', 'test1'], - 'name': 'users_all_idx'} + { + "unique": False, + "column_names": ["test1", "test2"], + "name": "users_t_idx", + }, + { + "unique": False, + "column_names": ["user_id", "test2", "test1"], + "name": "users_all_idx", + }, ] self._assert_insp_indexes(indexes, expected_indexes) @@ -721,10 +765,7 @@ class ComponentReflectionTest(fixtures.TablesTest): # reflecting an index that has "x DESC" in it as the column. # the DB may or may not give us "x", but make sure we get the index # back, it has a name, it's connected to the table. - expected_indexes = [ - {'unique': False, - 'name': ixname} - ] + expected_indexes = [{"unique": False, "name": ixname}] self._assert_insp_indexes(indexes, expected_indexes) t = Table(tname, meta, autoload_with=meta.bind) @@ -748,24 +789,30 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.requires.unique_constraint_reflection def test_get_temp_table_unique_constraints(self): insp = inspect(self.bind) - reflected = insp.get_unique_constraints('user_tmp') + reflected = insp.get_unique_constraints("user_tmp") for refl in reflected: # Different dialects handle duplicate index and constraints # differently, so ignore this flag - refl.pop('duplicates_index', None) - eq_(reflected, [{'column_names': ['name'], 'name': 'user_tmp_uq'}]) + refl.pop("duplicates_index", None) + eq_(reflected, [{"column_names": ["name"], "name": "user_tmp_uq"}]) @testing.requires.temp_table_reflection def test_get_temp_table_indexes(self): insp = inspect(self.bind) - indexes = insp.get_indexes('user_tmp') + indexes = insp.get_indexes("user_tmp") for ind in indexes: - ind.pop('dialect_options', None) + ind.pop("dialect_options", None) eq_( # TODO: we need to add better filtering for indexes/uq constraints # that are doubled up - [idx for idx in indexes if idx['name'] == 'user_tmp_ix'], - [{'unique': False, 'column_names': ['foo'], 'name': 'user_tmp_ix'}] + [idx for idx in indexes if idx["name"] == "user_tmp_ix"], + [ + { + "unique": False, + "column_names": ["foo"], + "name": "user_tmp_ix", + } + ], ) @testing.requires.unique_constraint_reflection @@ -783,36 +830,37 @@ class ComponentReflectionTest(fixtures.TablesTest): # CREATE TABLE? uniques = sorted( [ - {'name': 'unique_a', 'column_names': ['a']}, - {'name': 'unique_a_b_c', 'column_names': ['a', 'b', 'c']}, - {'name': 'unique_c_a_b', 'column_names': ['c', 'a', 'b']}, - {'name': 'unique_asc_key', 'column_names': ['asc', 'key']}, - {'name': 'i.have.dots', 'column_names': ['b']}, - {'name': 'i have spaces', 'column_names': ['c']}, + {"name": "unique_a", "column_names": ["a"]}, + {"name": "unique_a_b_c", "column_names": ["a", "b", "c"]}, + {"name": "unique_c_a_b", "column_names": ["c", "a", "b"]}, + {"name": "unique_asc_key", "column_names": ["asc", "key"]}, + {"name": "i.have.dots", "column_names": ["b"]}, + {"name": "i have spaces", "column_names": ["c"]}, ], - key=operator.itemgetter('name') + key=operator.itemgetter("name"), ) orig_meta = self.metadata table = Table( - 'testtbl', orig_meta, - Column('a', sa.String(20)), - Column('b', sa.String(30)), - Column('c', sa.Integer), + "testtbl", + orig_meta, + Column("a", sa.String(20)), + Column("b", sa.String(30)), + Column("c", sa.Integer), # reserved identifiers - Column('asc', sa.String(30)), - Column('key', sa.String(30)), - schema=schema + Column("asc", sa.String(30)), + Column("key", sa.String(30)), + schema=schema, ) for uc in uniques: table.append_constraint( - sa.UniqueConstraint(*uc['column_names'], name=uc['name']) + sa.UniqueConstraint(*uc["column_names"], name=uc["name"]) ) orig_meta.create_all() inspector = inspect(orig_meta.bind) reflected = sorted( - inspector.get_unique_constraints('testtbl', schema=schema), - key=operator.itemgetter('name') + inspector.get_unique_constraints("testtbl", schema=schema), + key=operator.itemgetter("name"), ) names_that_duplicate_index = set() @@ -820,25 +868,31 @@ class ComponentReflectionTest(fixtures.TablesTest): for orig, refl in zip(uniques, reflected): # Different dialects handle duplicate index and constraints # differently, so ignore this flag - dupe = refl.pop('duplicates_index', None) + dupe = refl.pop("duplicates_index", None) if dupe: names_that_duplicate_index.add(dupe) eq_(orig, refl) reflected_metadata = MetaData() reflected = Table( - 'testtbl', reflected_metadata, autoload_with=orig_meta.bind, - schema=schema) + "testtbl", + reflected_metadata, + autoload_with=orig_meta.bind, + schema=schema, + ) # test "deduplicates for index" logic. MySQL and Oracle # "unique constraints" are actually unique indexes (with possible # exception of a unique that is a dupe of another one in the case # of Oracle). make sure # they aren't duplicated. idx_names = set([idx.name for idx in reflected.indexes]) - uq_names = set([ - uq.name for uq in reflected.constraints - if isinstance(uq, sa.UniqueConstraint)]).difference( - ['unique_c_a_b']) + uq_names = set( + [ + uq.name + for uq in reflected.constraints + if isinstance(uq, sa.UniqueConstraint) + ] + ).difference(["unique_c_a_b"]) assert not idx_names.intersection(uq_names) if names_that_duplicate_index: @@ -858,47 +912,52 @@ class ComponentReflectionTest(fixtures.TablesTest): def _test_get_check_constraints(self, schema=None): orig_meta = self.metadata Table( - 'sa_cc', orig_meta, - Column('a', Integer()), - sa.CheckConstraint('a > 1 AND a < 5', name='cc1'), - sa.CheckConstraint('a = 1 OR (a > 2 AND a < 5)', name='cc2'), - schema=schema + "sa_cc", + orig_meta, + Column("a", Integer()), + sa.CheckConstraint("a > 1 AND a < 5", name="cc1"), + sa.CheckConstraint("a = 1 OR (a > 2 AND a < 5)", name="cc2"), + schema=schema, ) orig_meta.create_all() inspector = inspect(orig_meta.bind) reflected = sorted( - inspector.get_check_constraints('sa_cc', schema=schema), - key=operator.itemgetter('name') + inspector.get_check_constraints("sa_cc", schema=schema), + key=operator.itemgetter("name"), ) # trying to minimize effect of quoting, parenthesis, etc. # may need to add more to this as new dialects get CHECK # constraint reflection support def normalize(sqltext): - return " ".join(re.findall(r"and|\d|=|a|or|<|>", sqltext.lower(), re.I)) + return " ".join( + re.findall(r"and|\d|=|a|or|<|>", sqltext.lower(), re.I) + ) reflected = [ - {"name": item["name"], - "sqltext": normalize(item["sqltext"])} + {"name": item["name"], "sqltext": normalize(item["sqltext"])} for item in reflected ] eq_( reflected, [ - {'name': 'cc1', 'sqltext': 'a > 1 and a < 5'}, - {'name': 'cc2', 'sqltext': 'a = 1 or a > 2 and a < 5'} - ] + {"name": "cc1", "sqltext": "a > 1 and a < 5"}, + {"name": "cc2", "sqltext": "a = 1 or a > 2 and a < 5"}, + ], ) @testing.provide_metadata def _test_get_view_definition(self, schema=None): meta = self.metadata - users, addresses, dingalings = self.tables.users, \ - self.tables.email_addresses, self.tables.dingalings - view_name1 = 'users_v' - view_name2 = 'email_addresses_v' + users, addresses, dingalings = ( + self.tables.users, + self.tables.email_addresses, + self.tables.dingalings, + ) + view_name1 = "users_v" + view_name2 = "email_addresses_v" insp = inspect(meta.bind) v1 = insp.get_view_definition(view_name1, schema=schema) self.assert_(v1) @@ -918,18 +977,21 @@ class ComponentReflectionTest(fixtures.TablesTest): @testing.provide_metadata def _test_get_table_oid(self, table_name, schema=None): meta = self.metadata - users, addresses, dingalings = self.tables.users, \ - self.tables.email_addresses, self.tables.dingalings + users, addresses, dingalings = ( + self.tables.users, + self.tables.email_addresses, + self.tables.dingalings, + ) insp = inspect(meta.bind) oid = insp.get_table_oid(table_name, schema) self.assert_(isinstance(oid, int)) def test_get_table_oid(self): - self._test_get_table_oid('users') + self._test_get_table_oid("users") @testing.requires.schemas def test_get_table_oid_with_schema(self): - self._test_get_table_oid('users', schema=testing.config.test_schema) + self._test_get_table_oid("users", schema=testing.config.test_schema) @testing.requires.table_reflection @testing.provide_metadata @@ -950,49 +1012,53 @@ class ComponentReflectionTest(fixtures.TablesTest): insp = inspect(meta.bind) for tname, cname in [ - ('users', 'user_id'), - ('email_addresses', 'address_id'), - ('dingalings', 'dingaling_id'), + ("users", "user_id"), + ("email_addresses", "address_id"), + ("dingalings", "dingaling_id"), ]: cols = insp.get_columns(tname) - id_ = {c['name']: c for c in cols}[cname] - assert id_.get('autoincrement', True) + id_ = {c["name"]: c for c in cols}[cname] + assert id_.get("autoincrement", True) class NormalizedNameTest(fixtures.TablesTest): - __requires__ = 'denormalized_names', + __requires__ = ("denormalized_names",) __backend__ = True @classmethod def define_tables(cls, metadata): Table( - quoted_name('t1', quote=True), metadata, - Column('id', Integer, primary_key=True), + quoted_name("t1", quote=True), + metadata, + Column("id", Integer, primary_key=True), ) Table( - quoted_name('t2', quote=True), metadata, - Column('id', Integer, primary_key=True), - Column('t1id', ForeignKey('t1.id')) + quoted_name("t2", quote=True), + metadata, + Column("id", Integer, primary_key=True), + Column("t1id", ForeignKey("t1.id")), ) def test_reflect_lowercase_forced_tables(self): m2 = MetaData(testing.db) - t2_ref = Table(quoted_name('t2', quote=True), m2, autoload=True) - t1_ref = m2.tables['t1'] + t2_ref = Table(quoted_name("t2", quote=True), m2, autoload=True) + t1_ref = m2.tables["t1"] assert t2_ref.c.t1id.references(t1_ref.c.id) m3 = MetaData(testing.db) - m3.reflect(only=lambda name, m: name.lower() in ('t1', 't2')) - assert m3.tables['t2'].c.t1id.references(m3.tables['t1'].c.id) + m3.reflect(only=lambda name, m: name.lower() in ("t1", "t2")) + assert m3.tables["t2"].c.t1id.references(m3.tables["t1"].c.id) def test_get_table_names(self): tablenames = [ - t for t in inspect(testing.db).get_table_names() - if t.lower() in ("t1", "t2")] + t + for t in inspect(testing.db).get_table_names() + if t.lower() in ("t1", "t2") + ] eq_(tablenames[0].upper(), tablenames[0].lower()) eq_(tablenames[1].upper(), tablenames[1].lower()) -__all__ = ('ComponentReflectionTest', 'HasTableTest', 'NormalizedNameTest') +__all__ = ("ComponentReflectionTest", "HasTableTest", "NormalizedNameTest") 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) diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py index 73ce02492..032b68eb6 100644 --- a/lib/sqlalchemy/testing/suite/test_select.py +++ b/lib/sqlalchemy/testing/suite/test_select.py @@ -16,10 +16,12 @@ class CollateTest(fixtures.TablesTest): @classmethod def define_tables(cls, metadata): - Table("some_table", metadata, - Column('id', Integer, primary_key=True), - Column('data', String(100)) - ) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(100)), + ) @classmethod def insert_data(cls): @@ -28,26 +30,21 @@ class CollateTest(fixtures.TablesTest): [ {"id": 1, "data": "collate data1"}, {"id": 2, "data": "collate data2"}, - ] + ], ) def _assert_result(self, select, result): - eq_( - config.db.execute(select).fetchall(), - result - ) + eq_(config.db.execute(select).fetchall(), result) @testing.requires.order_by_collation def test_collate_order_by(self): collation = testing.requires.get_order_by_collation(testing.config) self._assert_result( - select([self.tables.some_table]). - order_by(self.tables.some_table.c.data.collate(collation).asc()), - [ - (1, "collate data1"), - (2, "collate data2"), - ] + select([self.tables.some_table]).order_by( + self.tables.some_table.c.data.collate(collation).asc() + ), + [(1, "collate data1"), (2, "collate data2")], ) @@ -59,17 +56,20 @@ class OrderByLabelTest(fixtures.TablesTest): setting. """ + __backend__ = True @classmethod def define_tables(cls, metadata): - Table("some_table", metadata, - Column('id', Integer, primary_key=True), - Column('x', Integer), - Column('y', Integer), - Column('q', String(50)), - Column('p', String(50)) - ) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True), + Column("x", Integer), + Column("y", Integer), + Column("q", String(50)), + Column("p", String(50)), + ) @classmethod def insert_data(cls): @@ -79,65 +79,55 @@ class OrderByLabelTest(fixtures.TablesTest): {"id": 1, "x": 1, "y": 2, "q": "q1", "p": "p3"}, {"id": 2, "x": 2, "y": 3, "q": "q2", "p": "p2"}, {"id": 3, "x": 3, "y": 4, "q": "q3", "p": "p1"}, - ] + ], ) def _assert_result(self, select, result): - eq_( - config.db.execute(select).fetchall(), - result - ) + eq_(config.db.execute(select).fetchall(), result) def test_plain(self): table = self.tables.some_table - lx = table.c.x.label('lx') - self._assert_result( - select([lx]).order_by(lx), - [(1, ), (2, ), (3, )] - ) + lx = table.c.x.label("lx") + self._assert_result(select([lx]).order_by(lx), [(1,), (2,), (3,)]) def test_composed_int(self): table = self.tables.some_table - lx = (table.c.x + table.c.y).label('lx') - self._assert_result( - select([lx]).order_by(lx), - [(3, ), (5, ), (7, )] - ) + lx = (table.c.x + table.c.y).label("lx") + self._assert_result(select([lx]).order_by(lx), [(3,), (5,), (7,)]) def test_composed_multiple(self): table = self.tables.some_table - lx = (table.c.x + table.c.y).label('lx') - ly = (func.lower(table.c.q) + table.c.p).label('ly') + lx = (table.c.x + table.c.y).label("lx") + ly = (func.lower(table.c.q) + table.c.p).label("ly") self._assert_result( select([lx, ly]).order_by(lx, ly.desc()), - [(3, util.u('q1p3')), (5, util.u('q2p2')), (7, util.u('q3p1'))] + [(3, util.u("q1p3")), (5, util.u("q2p2")), (7, util.u("q3p1"))], ) def test_plain_desc(self): table = self.tables.some_table - lx = table.c.x.label('lx') + lx = table.c.x.label("lx") self._assert_result( - select([lx]).order_by(lx.desc()), - [(3, ), (2, ), (1, )] + select([lx]).order_by(lx.desc()), [(3,), (2,), (1,)] ) def test_composed_int_desc(self): table = self.tables.some_table - lx = (table.c.x + table.c.y).label('lx') + lx = (table.c.x + table.c.y).label("lx") self._assert_result( - select([lx]).order_by(lx.desc()), - [(7, ), (5, ), (3, )] + select([lx]).order_by(lx.desc()), [(7,), (5,), (3,)] ) @testing.requires.group_by_complex_expression def test_group_by_composed(self): table = self.tables.some_table - expr = (table.c.x + table.c.y).label('lx') - stmt = select([func.count(table.c.id), expr]).group_by(expr).order_by(expr) - self._assert_result( - stmt, - [(1, 3), (1, 5), (1, 7)] + expr = (table.c.x + table.c.y).label("lx") + stmt = ( + select([func.count(table.c.id), expr]) + .group_by(expr) + .order_by(expr) ) + self._assert_result(stmt, [(1, 3), (1, 5), (1, 7)]) class LimitOffsetTest(fixtures.TablesTest): @@ -145,10 +135,13 @@ class LimitOffsetTest(fixtures.TablesTest): @classmethod def define_tables(cls, metadata): - Table("some_table", metadata, - Column('id', Integer, primary_key=True), - Column('x', Integer), - Column('y', Integer)) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True), + Column("x", Integer), + Column("y", Integer), + ) @classmethod def insert_data(cls): @@ -159,20 +152,17 @@ class LimitOffsetTest(fixtures.TablesTest): {"id": 2, "x": 2, "y": 3}, {"id": 3, "x": 3, "y": 4}, {"id": 4, "x": 4, "y": 5}, - ] + ], ) def _assert_result(self, select, result, params=()): - eq_( - config.db.execute(select, params).fetchall(), - result - ) + eq_(config.db.execute(select, params).fetchall(), result) def test_simple_limit(self): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).limit(2), - [(1, 1, 2), (2, 2, 3)] + [(1, 1, 2), (2, 2, 3)], ) @testing.requires.offset @@ -180,7 +170,7 @@ class LimitOffsetTest(fixtures.TablesTest): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).offset(2), - [(3, 3, 4), (4, 4, 5)] + [(3, 3, 4), (4, 4, 5)], ) @testing.requires.offset @@ -188,7 +178,7 @@ class LimitOffsetTest(fixtures.TablesTest): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).limit(2).offset(1), - [(2, 2, 3), (3, 3, 4)] + [(2, 2, 3), (3, 3, 4)], ) @testing.requires.offset @@ -198,41 +188,40 @@ class LimitOffsetTest(fixtures.TablesTest): table = self.tables.some_table stmt = select([table]).order_by(table.c.id).limit(2).offset(1) sql = stmt.compile( - dialect=config.db.dialect, - compile_kwargs={"literal_binds": True}) + dialect=config.db.dialect, compile_kwargs={"literal_binds": True} + ) sql = str(sql) - self._assert_result( - sql, - [(2, 2, 3), (3, 3, 4)] - ) + self._assert_result(sql, [(2, 2, 3), (3, 3, 4)]) @testing.requires.bound_limit_offset def test_bound_limit(self): table = self.tables.some_table self._assert_result( - select([table]).order_by(table.c.id).limit(bindparam('l')), + select([table]).order_by(table.c.id).limit(bindparam("l")), [(1, 1, 2), (2, 2, 3)], - params={"l": 2} + params={"l": 2}, ) @testing.requires.bound_limit_offset def test_bound_offset(self): table = self.tables.some_table self._assert_result( - select([table]).order_by(table.c.id).offset(bindparam('o')), + select([table]).order_by(table.c.id).offset(bindparam("o")), [(3, 3, 4), (4, 4, 5)], - params={"o": 2} + params={"o": 2}, ) @testing.requires.bound_limit_offset def test_bound_limit_offset(self): table = self.tables.some_table self._assert_result( - select([table]).order_by(table.c.id). - limit(bindparam("l")).offset(bindparam("o")), + select([table]) + .order_by(table.c.id) + .limit(bindparam("l")) + .offset(bindparam("o")), [(2, 2, 3), (3, 3, 4)], - params={"l": 2, "o": 1} + params={"l": 2, "o": 1}, ) @@ -241,10 +230,13 @@ class CompoundSelectTest(fixtures.TablesTest): @classmethod def define_tables(cls, metadata): - Table("some_table", metadata, - Column('id', Integer, primary_key=True), - Column('x', Integer), - Column('y', Integer)) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True), + Column("x", Integer), + Column("y", Integer), + ) @classmethod def insert_data(cls): @@ -255,14 +247,11 @@ class CompoundSelectTest(fixtures.TablesTest): {"id": 2, "x": 2, "y": 3}, {"id": 3, "x": 3, "y": 4}, {"id": 4, "x": 4, "y": 5}, - ] + ], ) def _assert_result(self, select, result, params=()): - eq_( - config.db.execute(select, params).fetchall(), - result - ) + eq_(config.db.execute(select, params).fetchall(), result) def test_plain_union(self): table = self.tables.some_table @@ -270,10 +259,7 @@ class CompoundSelectTest(fixtures.TablesTest): s2 = select([table]).where(table.c.id == 3) u1 = union(s1, s2) - self._assert_result( - u1.order_by(u1.c.id), - [(2, 2, 3), (3, 3, 4)] - ) + self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) def test_select_from_plain_union(self): table = self.tables.some_table @@ -281,80 +267,88 @@ class CompoundSelectTest(fixtures.TablesTest): s2 = select([table]).where(table.c.id == 3) u1 = union(s1, s2).alias().select() - self._assert_result( - u1.order_by(u1.c.id), - [(2, 2, 3), (3, 3, 4)] - ) + self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) @testing.requires.order_by_col_from_union @testing.requires.parens_in_union_contained_select_w_limit_offset def test_limit_offset_selectable_in_unions(self): table = self.tables.some_table - s1 = select([table]).where(table.c.id == 2).\ - limit(1).order_by(table.c.id) - s2 = select([table]).where(table.c.id == 3).\ - limit(1).order_by(table.c.id) + s1 = ( + select([table]) + .where(table.c.id == 2) + .limit(1) + .order_by(table.c.id) + ) + s2 = ( + select([table]) + .where(table.c.id == 3) + .limit(1) + .order_by(table.c.id) + ) u1 = union(s1, s2).limit(2) - self._assert_result( - u1.order_by(u1.c.id), - [(2, 2, 3), (3, 3, 4)] - ) + self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) @testing.requires.parens_in_union_contained_select_wo_limit_offset def test_order_by_selectable_in_unions(self): table = self.tables.some_table - s1 = select([table]).where(table.c.id == 2).\ - order_by(table.c.id) - s2 = select([table]).where(table.c.id == 3).\ - order_by(table.c.id) + s1 = select([table]).where(table.c.id == 2).order_by(table.c.id) + s2 = select([table]).where(table.c.id == 3).order_by(table.c.id) u1 = union(s1, s2).limit(2) - self._assert_result( - u1.order_by(u1.c.id), - [(2, 2, 3), (3, 3, 4)] - ) + self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) def test_distinct_selectable_in_unions(self): table = self.tables.some_table - s1 = select([table]).where(table.c.id == 2).\ - distinct() - s2 = select([table]).where(table.c.id == 3).\ - distinct() + s1 = select([table]).where(table.c.id == 2).distinct() + s2 = select([table]).where(table.c.id == 3).distinct() u1 = union(s1, s2).limit(2) - self._assert_result( - u1.order_by(u1.c.id), - [(2, 2, 3), (3, 3, 4)] - ) + self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) @testing.requires.parens_in_union_contained_select_w_limit_offset def test_limit_offset_in_unions_from_alias(self): table = self.tables.some_table - s1 = select([table]).where(table.c.id == 2).\ - limit(1).order_by(table.c.id) - s2 = select([table]).where(table.c.id == 3).\ - limit(1).order_by(table.c.id) + s1 = ( + select([table]) + .where(table.c.id == 2) + .limit(1) + .order_by(table.c.id) + ) + s2 = ( + select([table]) + .where(table.c.id == 3) + .limit(1) + .order_by(table.c.id) + ) # this necessarily has double parens u1 = union(s1, s2).alias() self._assert_result( - u1.select().limit(2).order_by(u1.c.id), - [(2, 2, 3), (3, 3, 4)] + u1.select().limit(2).order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] ) def test_limit_offset_aliased_selectable_in_unions(self): table = self.tables.some_table - s1 = select([table]).where(table.c.id == 2).\ - limit(1).order_by(table.c.id).alias().select() - s2 = select([table]).where(table.c.id == 3).\ - limit(1).order_by(table.c.id).alias().select() + s1 = ( + select([table]) + .where(table.c.id == 2) + .limit(1) + .order_by(table.c.id) + .alias() + .select() + ) + s2 = ( + select([table]) + .where(table.c.id == 3) + .limit(1) + .order_by(table.c.id) + .alias() + .select() + ) u1 = union(s1, s2).limit(2) - self._assert_result( - u1.order_by(u1.c.id), - [(2, 2, 3), (3, 3, 4)] - ) + self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) class ExpandingBoundInTest(fixtures.TablesTest): @@ -362,11 +356,14 @@ class ExpandingBoundInTest(fixtures.TablesTest): @classmethod def define_tables(cls, metadata): - Table("some_table", metadata, - Column('id', Integer, primary_key=True), - Column('x', Integer), - Column('y', Integer), - Column('z', String(50))) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True), + Column("x", Integer), + Column("y", Integer), + Column("z", String(50)), + ) @classmethod def insert_data(cls): @@ -377,178 +374,184 @@ class ExpandingBoundInTest(fixtures.TablesTest): {"id": 2, "x": 2, "y": 3, "z": "z2"}, {"id": 3, "x": 3, "y": 4, "z": "z3"}, {"id": 4, "x": 4, "y": 5, "z": "z4"}, - ] + ], ) def _assert_result(self, select, result, params=()): - eq_( - config.db.execute(select, params).fetchall(), - result - ) + eq_(config.db.execute(select, params).fetchall(), result) def test_multiple_empty_sets(self): # test that any anonymous aliasing used by the dialect # is fine with duplicates table = self.tables.some_table - stmt = select([table.c.id]).where( - table.c.x.in_(bindparam('q', expanding=True))).where( - table.c.y.in_(bindparam('p', expanding=True)) - ).order_by(table.c.id) - - self._assert_result( - stmt, - [], - params={"q": [], "p": []}, + stmt = ( + select([table.c.id]) + .where(table.c.x.in_(bindparam("q", expanding=True))) + .where(table.c.y.in_(bindparam("p", expanding=True))) + .order_by(table.c.id) ) + self._assert_result(stmt, [], params={"q": [], "p": []}) + @testing.requires.tuple_in def test_empty_heterogeneous_tuples(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - tuple_(table.c.x, table.c.z).in_( - bindparam('q', expanding=True))).order_by(table.c.id) - - self._assert_result( - stmt, - [], - params={"q": []}, + stmt = ( + select([table.c.id]) + .where( + tuple_(table.c.x, table.c.z).in_( + bindparam("q", expanding=True) + ) + ) + .order_by(table.c.id) ) + self._assert_result(stmt, [], params={"q": []}) + @testing.requires.tuple_in def test_empty_homogeneous_tuples(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - tuple_(table.c.x, table.c.y).in_( - bindparam('q', expanding=True))).order_by(table.c.id) - - self._assert_result( - stmt, - [], - params={"q": []}, + stmt = ( + select([table.c.id]) + .where( + tuple_(table.c.x, table.c.y).in_( + bindparam("q", expanding=True) + ) + ) + .order_by(table.c.id) ) + self._assert_result(stmt, [], params={"q": []}) + def test_bound_in_scalar(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - table.c.x.in_(bindparam('q', expanding=True))).order_by(table.c.id) - - self._assert_result( - stmt, - [(2, ), (3, ), (4, )], - params={"q": [2, 3, 4]}, + stmt = ( + select([table.c.id]) + .where(table.c.x.in_(bindparam("q", expanding=True))) + .order_by(table.c.id) ) + self._assert_result(stmt, [(2,), (3,), (4,)], params={"q": [2, 3, 4]}) + @testing.requires.tuple_in def test_bound_in_two_tuple(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - tuple_(table.c.x, table.c.y).in_( - bindparam('q', expanding=True))).order_by(table.c.id) + stmt = ( + select([table.c.id]) + .where( + tuple_(table.c.x, table.c.y).in_( + bindparam("q", expanding=True) + ) + ) + .order_by(table.c.id) + ) self._assert_result( - stmt, - [(2, ), (3, ), (4, )], - params={"q": [(2, 3), (3, 4), (4, 5)]}, + stmt, [(2,), (3,), (4,)], params={"q": [(2, 3), (3, 4), (4, 5)]} ) @testing.requires.tuple_in def test_bound_in_heterogeneous_two_tuple(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - tuple_(table.c.x, table.c.z).in_( - bindparam('q', expanding=True))).order_by(table.c.id) + stmt = ( + select([table.c.id]) + .where( + tuple_(table.c.x, table.c.z).in_( + bindparam("q", expanding=True) + ) + ) + .order_by(table.c.id) + ) self._assert_result( stmt, - [(2, ), (3, ), (4, )], + [(2,), (3,), (4,)], params={"q": [(2, "z2"), (3, "z3"), (4, "z4")]}, ) def test_empty_set_against_integer(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - table.c.x.in_(bindparam('q', expanding=True))).order_by(table.c.id) - - self._assert_result( - stmt, - [], - params={"q": []}, + stmt = ( + select([table.c.id]) + .where(table.c.x.in_(bindparam("q", expanding=True))) + .order_by(table.c.id) ) + self._assert_result(stmt, [], params={"q": []}) + def test_empty_set_against_integer_negation(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - table.c.x.notin_(bindparam('q', expanding=True)) - ).order_by(table.c.id) - - self._assert_result( - stmt, - [(1, ), (2, ), (3, ), (4, )], - params={"q": []}, + stmt = ( + select([table.c.id]) + .where(table.c.x.notin_(bindparam("q", expanding=True))) + .order_by(table.c.id) ) + self._assert_result(stmt, [(1,), (2,), (3,), (4,)], params={"q": []}) + def test_empty_set_against_string(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - table.c.z.in_(bindparam('q', expanding=True))).order_by(table.c.id) - - self._assert_result( - stmt, - [], - params={"q": []}, + stmt = ( + select([table.c.id]) + .where(table.c.z.in_(bindparam("q", expanding=True))) + .order_by(table.c.id) ) + self._assert_result(stmt, [], params={"q": []}) + def test_empty_set_against_string_negation(self): table = self.tables.some_table - stmt = select([table.c.id]).where( - table.c.z.notin_(bindparam('q', expanding=True)) - ).order_by(table.c.id) - - self._assert_result( - stmt, - [(1, ), (2, ), (3, ), (4, )], - params={"q": []}, + stmt = ( + select([table.c.id]) + .where(table.c.z.notin_(bindparam("q", expanding=True))) + .order_by(table.c.id) ) + self._assert_result(stmt, [(1,), (2,), (3,), (4,)], params={"q": []}) + def test_null_in_empty_set_is_false(self): - stmt = select([ - case( - [ - ( - null().in_(bindparam('foo', value=(), expanding=True)), - true() - ) - ], - else_=false() - ) - ]) - in_( - config.db.execute(stmt).fetchone()[0], - (False, 0) + stmt = select( + [ + case( + [ + ( + null().in_( + bindparam("foo", value=(), expanding=True) + ), + true(), + ) + ], + else_=false(), + ) + ] ) + in_(config.db.execute(stmt).fetchone()[0], (False, 0)) class LikeFunctionsTest(fixtures.TablesTest): __backend__ = True - run_inserts = 'once' + run_inserts = "once" run_deletes = None @classmethod def define_tables(cls, metadata): - Table("some_table", metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50))) + Table( + "some_table", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), + ) @classmethod def insert_data(cls): @@ -565,7 +568,7 @@ class LikeFunctionsTest(fixtures.TablesTest): {"id": 8, "data": "ab9cdefg"}, {"id": 9, "data": "abcde#fg"}, {"id": 10, "data": "abcd9fg"}, - ] + ], ) def _test(self, expr, expected): @@ -573,8 +576,10 @@ class LikeFunctionsTest(fixtures.TablesTest): with config.db.connect() as conn: rows = { - value for value, in - conn.execute(select([some_table.c.id]).where(expr)) + value + for value, in conn.execute( + select([some_table.c.id]).where(expr) + ) } eq_(rows, expected) @@ -591,7 +596,8 @@ class LikeFunctionsTest(fixtures.TablesTest): col = self.tables.some_table.c.data self._test( col.startswith(literal_column("'ab%c'")), - {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}) + {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, + ) def test_startswith_escape(self): col = self.tables.some_table.c.data @@ -608,8 +614,9 @@ class LikeFunctionsTest(fixtures.TablesTest): def test_endswith_sqlexpr(self): col = self.tables.some_table.c.data - self._test(col.endswith(literal_column("'e%fg'")), - {1, 2, 3, 4, 5, 6, 7, 8, 9}) + self._test( + col.endswith(literal_column("'e%fg'")), {1, 2, 3, 4, 5, 6, 7, 8, 9} + ) def test_endswith_autoescape(self): col = self.tables.some_table.c.data @@ -640,5 +647,3 @@ class LikeFunctionsTest(fixtures.TablesTest): col = self.tables.some_table.c.data self._test(col.contains("b%cd", autoescape=True, escape="#"), {3}) self._test(col.contains("b#cd", autoescape=True, escape="#"), {7}) - - diff --git a/lib/sqlalchemy/testing/suite/test_sequence.py b/lib/sqlalchemy/testing/suite/test_sequence.py index f1c00de6b..15a850fe9 100644 --- a/lib/sqlalchemy/testing/suite/test_sequence.py +++ b/lib/sqlalchemy/testing/suite/test_sequence.py @@ -9,140 +9,144 @@ from ..schema import Table, Column class SequenceTest(fixtures.TablesTest): - __requires__ = ('sequences',) + __requires__ = ("sequences",) __backend__ = True - run_create_tables = 'each' + run_create_tables = "each" @classmethod def define_tables(cls, metadata): - Table('seq_pk', metadata, - Column('id', Integer, Sequence('tab_id_seq'), primary_key=True), - Column('data', String(50)) - ) + Table( + "seq_pk", + metadata, + Column("id", Integer, Sequence("tab_id_seq"), primary_key=True), + Column("data", String(50)), + ) - Table('seq_opt_pk', metadata, - Column('id', Integer, Sequence('tab_id_seq', optional=True), - primary_key=True), - Column('data', String(50)) - ) + Table( + "seq_opt_pk", + metadata, + Column( + "id", + Integer, + Sequence("tab_id_seq", optional=True), + primary_key=True, + ), + Column("data", String(50)), + ) def test_insert_roundtrip(self): - config.db.execute( - self.tables.seq_pk.insert(), - data="some data" - ) + config.db.execute(self.tables.seq_pk.insert(), data="some data") self._assert_round_trip(self.tables.seq_pk, config.db) def test_insert_lastrowid(self): - r = config.db.execute( - self.tables.seq_pk.insert(), - data="some data" - ) - eq_( - r.inserted_primary_key, - [1] - ) + r = config.db.execute(self.tables.seq_pk.insert(), data="some data") + eq_(r.inserted_primary_key, [1]) def test_nextval_direct(self): - r = config.db.execute( - self.tables.seq_pk.c.id.default - ) - eq_( - r, 1 - ) + r = config.db.execute(self.tables.seq_pk.c.id.default) + eq_(r, 1) @requirements.sequences_optional def test_optional_seq(self): r = config.db.execute( - self.tables.seq_opt_pk.insert(), - data="some data" - ) - eq_( - r.inserted_primary_key, - [1] + self.tables.seq_opt_pk.insert(), data="some data" ) + eq_(r.inserted_primary_key, [1]) def _assert_round_trip(self, table, conn): row = conn.execute(table.select()).first() - eq_( - row, - (1, "some data") - ) + eq_(row, (1, "some data")) class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase): - __requires__ = ('sequences',) + __requires__ = ("sequences",) __backend__ = True def test_literal_binds_inline_compile(self): table = Table( - 'x', MetaData(), - Column('y', Integer, Sequence('y_seq')), - Column('q', Integer)) + "x", + MetaData(), + Column("y", Integer, Sequence("y_seq")), + Column("q", Integer), + ) stmt = table.insert().values(q=5) seq_nextval = testing.db.dialect.statement_compiler( - statement=None, dialect=testing.db.dialect).visit_sequence( - Sequence("y_seq")) + statement=None, dialect=testing.db.dialect + ).visit_sequence(Sequence("y_seq")) self.assert_compile( stmt, - "INSERT INTO x (y, q) VALUES (%s, 5)" % (seq_nextval, ), + "INSERT INTO x (y, q) VALUES (%s, 5)" % (seq_nextval,), literal_binds=True, - dialect=testing.db.dialect) + dialect=testing.db.dialect, + ) class HasSequenceTest(fixtures.TestBase): - __requires__ = 'sequences', + __requires__ = ("sequences",) __backend__ = True def test_has_sequence(self): - s1 = Sequence('user_id_seq') + s1 = Sequence("user_id_seq") testing.db.execute(schema.CreateSequence(s1)) try: - eq_(testing.db.dialect.has_sequence(testing.db, - 'user_id_seq'), True) + eq_( + testing.db.dialect.has_sequence(testing.db, "user_id_seq"), + True, + ) finally: testing.db.execute(schema.DropSequence(s1)) @testing.requires.schemas def test_has_sequence_schema(self): - s1 = Sequence('user_id_seq', schema=config.test_schema) + s1 = Sequence("user_id_seq", schema=config.test_schema) testing.db.execute(schema.CreateSequence(s1)) try: - eq_(testing.db.dialect.has_sequence( - testing.db, 'user_id_seq', schema=config.test_schema), True) + eq_( + testing.db.dialect.has_sequence( + testing.db, "user_id_seq", schema=config.test_schema + ), + True, + ) finally: testing.db.execute(schema.DropSequence(s1)) def test_has_sequence_neg(self): - eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'), - False) + eq_(testing.db.dialect.has_sequence(testing.db, "user_id_seq"), False) @testing.requires.schemas def test_has_sequence_schemas_neg(self): - eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq', - schema=config.test_schema), - False) + eq_( + testing.db.dialect.has_sequence( + testing.db, "user_id_seq", schema=config.test_schema + ), + False, + ) @testing.requires.schemas def test_has_sequence_default_not_in_remote(self): - s1 = Sequence('user_id_seq') + s1 = Sequence("user_id_seq") testing.db.execute(schema.CreateSequence(s1)) try: - eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq', - schema=config.test_schema), - False) + eq_( + testing.db.dialect.has_sequence( + testing.db, "user_id_seq", schema=config.test_schema + ), + False, + ) finally: testing.db.execute(schema.DropSequence(s1)) @testing.requires.schemas def test_has_sequence_remote_not_in_default(self): - s1 = Sequence('user_id_seq', schema=config.test_schema) + s1 = Sequence("user_id_seq", schema=config.test_schema) testing.db.execute(schema.CreateSequence(s1)) try: - eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'), - False) + eq_( + testing.db.dialect.has_sequence(testing.db, "user_id_seq"), + False, + ) finally: testing.db.execute(schema.DropSequence(s1)) diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py index 27c7bb115..6dfb80915 100644 --- a/lib/sqlalchemy/testing/suite/test_types.py +++ b/lib/sqlalchemy/testing/suite/test_types.py @@ -4,9 +4,24 @@ from .. import fixtures, config from ..assertions import eq_ from ..config import requirements from sqlalchemy import Integer, Unicode, UnicodeText, select, TIMESTAMP -from sqlalchemy import Date, DateTime, Time, MetaData, String, \ - Text, Numeric, Float, literal, Boolean, cast, null, JSON, and_, \ - type_coerce, BigInteger +from sqlalchemy import ( + Date, + DateTime, + Time, + MetaData, + String, + Text, + Numeric, + Float, + literal, + Boolean, + cast, + null, + JSON, + and_, + type_coerce, + BigInteger, +) from ..schema import Table, Column from ... import testing import decimal @@ -24,13 +39,17 @@ class _LiteralRoundTripFixture(object): # into a typed column. we can then SELECT it back as its # official type; ideally we'd be able to use CAST here # but MySQL in particular can't CAST fully - t = Table('t', self.metadata, Column('x', type_)) + t = Table("t", self.metadata, Column("x", type_)) t.create() for value in input_: - ins = t.insert().values(x=literal(value)).compile( - dialect=testing.db.dialect, - compile_kwargs=dict(literal_binds=True) + ins = ( + t.insert() + .values(x=literal(value)) + .compile( + dialect=testing.db.dialect, + compile_kwargs=dict(literal_binds=True), + ) ) testing.db.execute(ins) @@ -42,40 +61,33 @@ class _LiteralRoundTripFixture(object): class _UnicodeFixture(_LiteralRoundTripFixture): - __requires__ = 'unicode_data', + __requires__ = ("unicode_data",) - data = u("Alors vous imaginez ma surprise, au lever du jour, " - "quand une drôle de petite voix m’a réveillé. Elle " - "disait: « S’il vous plaît… dessine-moi un mouton! »") + data = u( + "Alors vous imaginez ma surprise, au lever du jour, " + "quand une drôle de petite voix m’a réveillé. Elle " + "disait: « S’il vous plaît… dessine-moi un mouton! »" + ) @classmethod def define_tables(cls, metadata): - Table('unicode_table', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('unicode_data', cls.datatype), - ) + Table( + "unicode_table", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("unicode_data", cls.datatype), + ) def test_round_trip(self): unicode_table = self.tables.unicode_table - config.db.execute( - unicode_table.insert(), - { - 'unicode_data': self.data, - } - ) + config.db.execute(unicode_table.insert(), {"unicode_data": self.data}) - row = config.db.execute( - select([ - unicode_table.c.unicode_data, - ]) - ).first() + row = config.db.execute(select([unicode_table.c.unicode_data])).first() - eq_( - row, - (self.data, ) - ) + eq_(row, (self.data,)) assert isinstance(row[0], util.text_type) def test_round_trip_executemany(self): @@ -83,44 +95,29 @@ class _UnicodeFixture(_LiteralRoundTripFixture): config.db.execute( unicode_table.insert(), - [ - { - 'unicode_data': self.data, - } - for i in range(3) - ] + [{"unicode_data": self.data} for i in range(3)], ) rows = config.db.execute( - select([ - unicode_table.c.unicode_data, - ]) + select([unicode_table.c.unicode_data]) ).fetchall() - eq_( - rows, - [(self.data, ) for i in range(3)] - ) + eq_(rows, [(self.data,) for i in range(3)]) for row in rows: assert isinstance(row[0], util.text_type) def _test_empty_strings(self): unicode_table = self.tables.unicode_table - config.db.execute( - unicode_table.insert(), - {"unicode_data": u('')} - ) - row = config.db.execute( - select([unicode_table.c.unicode_data]) - ).first() - eq_(row, (u(''),)) + config.db.execute(unicode_table.insert(), {"unicode_data": u("")}) + row = config.db.execute(select([unicode_table.c.unicode_data])).first() + eq_(row, (u(""),)) def test_literal(self): self._literal_round_trip(self.datatype, [self.data], [self.data]) class UnicodeVarcharTest(_UnicodeFixture, fixtures.TablesTest): - __requires__ = 'unicode_data', + __requires__ = ("unicode_data",) __backend__ = True datatype = Unicode(255) @@ -131,7 +128,7 @@ class UnicodeVarcharTest(_UnicodeFixture, fixtures.TablesTest): class UnicodeTextTest(_UnicodeFixture, fixtures.TablesTest): - __requires__ = 'unicode_data', 'text_type' + __requires__ = "unicode_data", "text_type" __backend__ = True datatype = UnicodeText() @@ -142,54 +139,47 @@ class UnicodeTextTest(_UnicodeFixture, fixtures.TablesTest): class TextTest(_LiteralRoundTripFixture, fixtures.TablesTest): - __requires__ = 'text_type', + __requires__ = ("text_type",) __backend__ = True @classmethod def define_tables(cls, metadata): - Table('text_table', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('text_data', Text), - ) + Table( + "text_table", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("text_data", Text), + ) def test_text_roundtrip(self): text_table = self.tables.text_table - config.db.execute( - text_table.insert(), - {"text_data": 'some text'} - ) - row = config.db.execute( - select([text_table.c.text_data]) - ).first() - eq_(row, ('some text',)) + config.db.execute(text_table.insert(), {"text_data": "some text"}) + row = config.db.execute(select([text_table.c.text_data])).first() + eq_(row, ("some text",)) def test_text_empty_strings(self): text_table = self.tables.text_table - config.db.execute( - text_table.insert(), - {"text_data": ''} - ) - row = config.db.execute( - select([text_table.c.text_data]) - ).first() - eq_(row, ('',)) + config.db.execute(text_table.insert(), {"text_data": ""}) + row = config.db.execute(select([text_table.c.text_data])).first() + eq_(row, ("",)) def test_literal(self): self._literal_round_trip(Text, ["some text"], ["some text"]) def test_literal_quoting(self): - data = '''some 'text' hey "hi there" that's text''' + data = """some 'text' hey "hi there" that's text""" self._literal_round_trip(Text, [data], [data]) def test_literal_backslashes(self): - data = r'backslash one \ backslash two \\ end' + data = r"backslash one \ backslash two \\ end" self._literal_round_trip(Text, [data], [data]) def test_literal_percentsigns(self): - data = r'percent % signs %% percent' + data = r"percent % signs %% percent" self._literal_round_trip(Text, [data], [data]) @@ -199,9 +189,7 @@ class StringTest(_LiteralRoundTripFixture, fixtures.TestBase): @requirements.unbounded_varchar def test_nolength_string(self): metadata = MetaData() - foo = Table('foo', metadata, - Column('one', String) - ) + foo = Table("foo", metadata, Column("one", String)) foo.create(config.db) foo.drop(config.db) @@ -210,11 +198,11 @@ class StringTest(_LiteralRoundTripFixture, fixtures.TestBase): self._literal_round_trip(String(40), ["some text"], ["some text"]) def test_literal_quoting(self): - data = '''some 'text' hey "hi there" that's text''' + data = """some 'text' hey "hi there" that's text""" self._literal_round_trip(String(40), [data], [data]) def test_literal_backslashes(self): - data = r'backslash one \ backslash two \\ end' + data = r"backslash one \ backslash two \\ end" self._literal_round_trip(String(40), [data], [data]) @@ -223,44 +211,32 @@ class _DateFixture(_LiteralRoundTripFixture): @classmethod def define_tables(cls, metadata): - Table('date_table', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('date_data', cls.datatype), - ) + Table( + "date_table", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("date_data", cls.datatype), + ) def test_round_trip(self): date_table = self.tables.date_table - config.db.execute( - date_table.insert(), - {'date_data': self.data} - ) + config.db.execute(date_table.insert(), {"date_data": self.data}) - row = config.db.execute( - select([ - date_table.c.date_data, - ]) - ).first() + row = config.db.execute(select([date_table.c.date_data])).first() compare = self.compare or self.data - eq_(row, - (compare, )) + eq_(row, (compare,)) assert isinstance(row[0], type(compare)) def test_null(self): date_table = self.tables.date_table - config.db.execute( - date_table.insert(), - {'date_data': None} - ) + config.db.execute(date_table.insert(), {"date_data": None}) - row = config.db.execute( - select([ - date_table.c.date_data, - ]) - ).first() + row = config.db.execute(select([date_table.c.date_data])).first() eq_(row, (None,)) @testing.requires.datetime_literals @@ -270,48 +246,49 @@ class _DateFixture(_LiteralRoundTripFixture): class DateTimeTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'datetime', + __requires__ = ("datetime",) __backend__ = True datatype = DateTime data = datetime.datetime(2012, 10, 15, 12, 57, 18) class DateTimeMicrosecondsTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'datetime_microseconds', + __requires__ = ("datetime_microseconds",) __backend__ = True datatype = DateTime data = datetime.datetime(2012, 10, 15, 12, 57, 18, 396) + class TimestampMicrosecondsTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'timestamp_microseconds', + __requires__ = ("timestamp_microseconds",) __backend__ = True datatype = TIMESTAMP data = datetime.datetime(2012, 10, 15, 12, 57, 18, 396) class TimeTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'time', + __requires__ = ("time",) __backend__ = True datatype = Time data = datetime.time(12, 57, 18) class TimeMicrosecondsTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'time_microseconds', + __requires__ = ("time_microseconds",) __backend__ = True datatype = Time data = datetime.time(12, 57, 18, 396) class DateTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'date', + __requires__ = ("date",) __backend__ = True datatype = Date data = datetime.date(2012, 10, 15) class DateTimeCoercedToDateTimeTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'date', 'date_coerces_from_datetime' + __requires__ = "date", "date_coerces_from_datetime" __backend__ = True datatype = Date data = datetime.datetime(2012, 10, 15, 12, 57, 18) @@ -319,14 +296,14 @@ class DateTimeCoercedToDateTimeTest(_DateFixture, fixtures.TablesTest): class DateTimeHistoricTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'datetime_historic', + __requires__ = ("datetime_historic",) __backend__ = True datatype = DateTime data = datetime.datetime(1850, 11, 10, 11, 52, 35) class DateHistoricTest(_DateFixture, fixtures.TablesTest): - __requires__ = 'date_historic', + __requires__ = ("date_historic",) __backend__ = True datatype = Date data = datetime.date(1727, 4, 1) @@ -345,26 +322,21 @@ class IntegerTest(_LiteralRoundTripFixture, fixtures.TestBase): def _round_trip(self, datatype, data): metadata = self.metadata int_table = Table( - 'integer_table', metadata, - Column('id', Integer, primary_key=True, - test_needs_autoincrement=True), - Column('integer_data', datatype), + "integer_table", + metadata, + Column( + "id", Integer, primary_key=True, test_needs_autoincrement=True + ), + Column("integer_data", datatype), ) metadata.create_all(config.db) - config.db.execute( - int_table.insert(), - {'integer_data': data} - ) + config.db.execute(int_table.insert(), {"integer_data": data}) - row = config.db.execute( - select([ - int_table.c.integer_data, - ]) - ).first() + row = config.db.execute(select([int_table.c.integer_data])).first() - eq_(row, (data, )) + eq_(row, (data,)) if util.py3k: assert isinstance(row[0], int) @@ -377,12 +349,11 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): @testing.emits_warning(r".*does \*not\* support Decimal objects natively") @testing.provide_metadata - def _do_test(self, type_, input_, output, - filter_=None, check_scale=False): + def _do_test(self, type_, input_, output, filter_=None, check_scale=False): metadata = self.metadata - t = Table('t', metadata, Column('x', type_)) + t = Table("t", metadata, Column("x", type_)) t.create() - t.insert().execute([{'x': x} for x in input_]) + t.insert().execute([{"x": x} for x in input_]) result = {row[0] for row in t.select().execute()} output = set(output) @@ -391,10 +362,7 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): output = set(filter_(x) for x in output) eq_(result, output) if check_scale: - eq_( - [str(x) for x in result], - [str(x) for x in output], - ) + eq_([str(x) for x in result], [str(x) for x in output]) @testing.emits_warning(r".*does \*not\* support Decimal objects natively") def test_render_literal_numeric(self): @@ -416,8 +384,8 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): self._literal_round_trip( Float(4), [15.7563, decimal.Decimal("15.7563")], - [15.7563, ], - filter_=lambda n: n is not None and round(n, 5) or None + [15.7563], + filter_=lambda n: n is not None and round(n, 5) or None, ) @testing.requires.precision_generic_float_type @@ -425,8 +393,8 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): self._do_test( Float(None, decimal_return_scale=7, asdecimal=True), [15.7563827, decimal.Decimal("15.7563827")], - [decimal.Decimal("15.7563827"), ], - check_scale=True + [decimal.Decimal("15.7563827")], + check_scale=True, ) def test_numeric_as_decimal(self): @@ -445,18 +413,12 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): @testing.requires.fetch_null_from_numeric def test_numeric_null_as_decimal(self): - self._do_test( - Numeric(precision=8, scale=4), - [None], - [None], - ) + self._do_test(Numeric(precision=8, scale=4), [None], [None]) @testing.requires.fetch_null_from_numeric def test_numeric_null_as_float(self): self._do_test( - Numeric(precision=8, scale=4, asdecimal=False), - [None], - [None], + Numeric(precision=8, scale=4, asdecimal=False), [None], [None] ) @testing.requires.floats_to_four_decimals @@ -472,15 +434,13 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): Float(precision=8), [15.7563, decimal.Decimal("15.7563")], [15.7563], - filter_=lambda n: n is not None and round(n, 5) or None + filter_=lambda n: n is not None and round(n, 5) or None, ) def test_float_coerce_round_trip(self): expr = 15.7563 - val = testing.db.scalar( - select([literal(expr)]) - ) + val = testing.db.scalar(select([literal(expr)])) eq_(val, expr) # this does not work in MySQL, see #4036, however we choose not @@ -491,34 +451,28 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): def test_decimal_coerce_round_trip(self): expr = decimal.Decimal("15.7563") - val = testing.db.scalar( - select([literal(expr)]) - ) + val = testing.db.scalar(select([literal(expr)])) eq_(val, expr) @testing.emits_warning(r".*does \*not\* support Decimal objects natively") def test_decimal_coerce_round_trip_w_cast(self): expr = decimal.Decimal("15.7563") - val = testing.db.scalar( - select([cast(expr, Numeric(10, 4))]) - ) + val = testing.db.scalar(select([cast(expr, Numeric(10, 4))])) eq_(val, expr) @testing.requires.precision_numerics_general def test_precision_decimal(self): - numbers = set([ - decimal.Decimal("54.234246451650"), - decimal.Decimal("0.004354"), - decimal.Decimal("900.0"), - ]) - - self._do_test( - Numeric(precision=18, scale=12), - numbers, - numbers, + numbers = set( + [ + decimal.Decimal("54.234246451650"), + decimal.Decimal("0.004354"), + decimal.Decimal("900.0"), + ] ) + self._do_test(Numeric(precision=18, scale=12), numbers, numbers) + @testing.requires.precision_numerics_enotation_large def test_enotation_decimal(self): """test exceedingly small decimals. @@ -528,25 +482,23 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): """ - numbers = set([ - decimal.Decimal('1E-2'), - decimal.Decimal('1E-3'), - decimal.Decimal('1E-4'), - decimal.Decimal('1E-5'), - decimal.Decimal('1E-6'), - decimal.Decimal('1E-7'), - decimal.Decimal('1E-8'), - decimal.Decimal("0.01000005940696"), - decimal.Decimal("0.00000005940696"), - decimal.Decimal("0.00000000000696"), - decimal.Decimal("0.70000000000696"), - decimal.Decimal("696E-12"), - ]) - self._do_test( - Numeric(precision=18, scale=14), - numbers, - numbers + numbers = set( + [ + decimal.Decimal("1E-2"), + decimal.Decimal("1E-3"), + decimal.Decimal("1E-4"), + decimal.Decimal("1E-5"), + decimal.Decimal("1E-6"), + decimal.Decimal("1E-7"), + decimal.Decimal("1E-8"), + decimal.Decimal("0.01000005940696"), + decimal.Decimal("0.00000005940696"), + decimal.Decimal("0.00000000000696"), + decimal.Decimal("0.70000000000696"), + decimal.Decimal("696E-12"), + ] ) + self._do_test(Numeric(precision=18, scale=14), numbers, numbers) @testing.requires.precision_numerics_enotation_large def test_enotation_decimal_large(self): @@ -554,41 +506,32 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): """ - numbers = set([ - decimal.Decimal('4E+8'), - decimal.Decimal("5748E+15"), - decimal.Decimal('1.521E+15'), - decimal.Decimal('00000000000000.1E+12'), - ]) - self._do_test( - Numeric(precision=25, scale=2), - numbers, - numbers + numbers = set( + [ + decimal.Decimal("4E+8"), + decimal.Decimal("5748E+15"), + decimal.Decimal("1.521E+15"), + decimal.Decimal("00000000000000.1E+12"), + ] ) + self._do_test(Numeric(precision=25, scale=2), numbers, numbers) @testing.requires.precision_numerics_many_significant_digits def test_many_significant_digits(self): - numbers = set([ - decimal.Decimal("31943874831932418390.01"), - decimal.Decimal("319438950232418390.273596"), - decimal.Decimal("87673.594069654243"), - ]) - self._do_test( - Numeric(precision=38, scale=12), - numbers, - numbers + numbers = set( + [ + decimal.Decimal("31943874831932418390.01"), + decimal.Decimal("319438950232418390.273596"), + decimal.Decimal("87673.594069654243"), + ] ) + self._do_test(Numeric(precision=38, scale=12), numbers, numbers) @testing.requires.precision_numerics_retains_significant_digits def test_numeric_no_decimal(self): - numbers = set([ - decimal.Decimal("1.000") - ]) + numbers = set([decimal.Decimal("1.000")]) self._do_test( - Numeric(precision=5, scale=3), - numbers, - numbers, - check_scale=True + Numeric(precision=5, scale=3), numbers, numbers, check_scale=True ) @@ -597,42 +540,32 @@ class BooleanTest(_LiteralRoundTripFixture, fixtures.TablesTest): @classmethod def define_tables(cls, metadata): - Table('boolean_table', metadata, - Column('id', Integer, primary_key=True, autoincrement=False), - Column('value', Boolean), - Column('unconstrained_value', Boolean(create_constraint=False)), - ) + Table( + "boolean_table", + metadata, + Column("id", Integer, primary_key=True, autoincrement=False), + Column("value", Boolean), + Column("unconstrained_value", Boolean(create_constraint=False)), + ) def test_render_literal_bool(self): - self._literal_round_trip( - Boolean(), - [True, False], - [True, False] - ) + self._literal_round_trip(Boolean(), [True, False], [True, False]) def test_round_trip(self): boolean_table = self.tables.boolean_table config.db.execute( boolean_table.insert(), - { - 'id': 1, - 'value': True, - 'unconstrained_value': False - } + {"id": 1, "value": True, "unconstrained_value": False}, ) row = config.db.execute( - select([ - boolean_table.c.value, - boolean_table.c.unconstrained_value - ]) + select( + [boolean_table.c.value, boolean_table.c.unconstrained_value] + ) ).first() - eq_( - row, - (True, False) - ) + eq_(row, (True, False)) assert isinstance(row[0], bool) def test_null(self): @@ -640,24 +573,16 @@ class BooleanTest(_LiteralRoundTripFixture, fixtures.TablesTest): config.db.execute( boolean_table.insert(), - { - 'id': 1, - 'value': None, - 'unconstrained_value': None - } + {"id": 1, "value": None, "unconstrained_value": None}, ) row = config.db.execute( - select([ - boolean_table.c.value, - boolean_table.c.unconstrained_value - ]) + select( + [boolean_table.c.value, boolean_table.c.unconstrained_value] + ) ).first() - eq_( - row, - (None, None) - ) + eq_(row, (None, None)) def test_whereclause(self): # testing "WHERE <column>" renders a compatible expression @@ -667,92 +592,82 @@ class BooleanTest(_LiteralRoundTripFixture, fixtures.TablesTest): conn.execute( boolean_table.insert(), [ - {'id': 1, 'value': True, 'unconstrained_value': True}, - {'id': 2, 'value': False, 'unconstrained_value': False} - ] + {"id": 1, "value": True, "unconstrained_value": True}, + {"id": 2, "value": False, "unconstrained_value": False}, + ], ) eq_( conn.scalar( select([boolean_table.c.id]).where(boolean_table.c.value) ), - 1 + 1, ) eq_( conn.scalar( select([boolean_table.c.id]).where( - boolean_table.c.unconstrained_value) + boolean_table.c.unconstrained_value + ) ), - 1 + 1, ) eq_( conn.scalar( select([boolean_table.c.id]).where(~boolean_table.c.value) ), - 2 + 2, ) eq_( conn.scalar( select([boolean_table.c.id]).where( - ~boolean_table.c.unconstrained_value) + ~boolean_table.c.unconstrained_value + ) ), - 2 + 2, ) - - class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): - __requires__ = 'json_type', + __requires__ = ("json_type",) __backend__ = True datatype = JSON - data1 = { - "key1": "value1", - "key2": "value2" - } + data1 = {"key1": "value1", "key2": "value2"} data2 = { "Key 'One'": "value1", "key two": "value2", - "key three": "value ' three '" + "key three": "value ' three '", } data3 = { "key1": [1, 2, 3], "key2": ["one", "two", "three"], - "key3": [{"four": "five"}, {"six": "seven"}] + "key3": [{"four": "five"}, {"six": "seven"}], } data4 = ["one", "two", "three"] data5 = { "nested": { - "elem1": [ - {"a": "b", "c": "d"}, - {"e": "f", "g": "h"} - ], - "elem2": { - "elem3": {"elem4": "elem5"} - } + "elem1": [{"a": "b", "c": "d"}, {"e": "f", "g": "h"}], + "elem2": {"elem3": {"elem4": "elem5"}}, } } - data6 = { - "a": 5, - "b": "some value", - "c": {"foo": "bar"} - } + data6 = {"a": 5, "b": "some value", "c": {"foo": "bar"}} @classmethod def define_tables(cls, metadata): - Table('data_table', metadata, - Column('id', Integer, primary_key=True), - Column('name', String(30), nullable=False), - Column('data', cls.datatype), - Column('nulldata', cls.datatype(none_as_null=True)) - ) + Table( + "data_table", + metadata, + Column("id", Integer, primary_key=True), + Column("name", String(30), nullable=False), + Column("data", cls.datatype), + Column("nulldata", cls.datatype(none_as_null=True)), + ) def test_round_trip_data1(self): self._test_round_trip(self.data1) @@ -761,99 +676,82 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): data_table = self.tables.data_table config.db.execute( - data_table.insert(), - {'name': 'row1', 'data': data_element} + data_table.insert(), {"name": "row1", "data": data_element} ) - row = config.db.execute( - select([ - data_table.c.data, - ]) - ).first() + row = config.db.execute(select([data_table.c.data])).first() - eq_(row, (data_element, )) + eq_(row, (data_element,)) def test_round_trip_none_as_sql_null(self): - col = self.tables.data_table.c['nulldata'] + col = self.tables.data_table.c["nulldata"] with config.db.connect() as conn: conn.execute( - self.tables.data_table.insert(), - {"name": "r1", "data": None} + self.tables.data_table.insert(), {"name": "r1", "data": None} ) eq_( conn.scalar( - select([self.tables.data_table.c.name]). - where(col.is_(null())) + select([self.tables.data_table.c.name]).where( + col.is_(null()) + ) ), - "r1" + "r1", ) - eq_( - conn.scalar( - select([col]) - ), - None - ) + eq_(conn.scalar(select([col])), None) def test_round_trip_json_null_as_json_null(self): - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] with config.db.connect() as conn: conn.execute( self.tables.data_table.insert(), - {"name": "r1", "data": JSON.NULL} + {"name": "r1", "data": JSON.NULL}, ) eq_( conn.scalar( - select([self.tables.data_table.c.name]). - where(cast(col, String) == 'null') + select([self.tables.data_table.c.name]).where( + cast(col, String) == "null" + ) ), - "r1" + "r1", ) - eq_( - conn.scalar( - select([col]) - ), - None - ) + eq_(conn.scalar(select([col])), None) def test_round_trip_none_as_json_null(self): - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] with config.db.connect() as conn: conn.execute( - self.tables.data_table.insert(), - {"name": "r1", "data": None} + self.tables.data_table.insert(), {"name": "r1", "data": None} ) eq_( conn.scalar( - select([self.tables.data_table.c.name]). - where(cast(col, String) == 'null') + select([self.tables.data_table.c.name]).where( + cast(col, String) == "null" + ) ), - "r1" + "r1", ) - eq_( - conn.scalar( - select([col]) - ), - None - ) + eq_(conn.scalar(select([col])), None) def _criteria_fixture(self): config.db.execute( self.tables.data_table.insert(), - [{"name": "r1", "data": self.data1}, - {"name": "r2", "data": self.data2}, - {"name": "r3", "data": self.data3}, - {"name": "r4", "data": self.data4}, - {"name": "r5", "data": self.data5}, - {"name": "r6", "data": self.data6}] + [ + {"name": "r1", "data": self.data1}, + {"name": "r2", "data": self.data2}, + {"name": "r3", "data": self.data3}, + {"name": "r4", "data": self.data4}, + {"name": "r5", "data": self.data5}, + {"name": "r6", "data": self.data6}, + ], ) def _test_index_criteria(self, crit, expected, test_literal=True): @@ -861,20 +759,20 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): with config.db.connect() as conn: stmt = select([self.tables.data_table.c.name]).where(crit) - eq_( - conn.scalar(stmt), - expected - ) + eq_(conn.scalar(stmt), expected) if test_literal: - literal_sql = str(stmt.compile( - config.db, compile_kwargs={"literal_binds": True})) + literal_sql = str( + stmt.compile( + config.db, compile_kwargs={"literal_binds": True} + ) + ) eq_(conn.scalar(literal_sql), expected) def test_crit_spaces_in_key(self): name = self.tables.data_table.c.name - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] # limit the rows here to avoid PG error # "cannot extract field from a non-object", which is @@ -882,76 +780,74 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): self._test_index_criteria( and_( name.in_(["r1", "r2", "r3"]), - cast(col["key two"], String) == '"value2"' + cast(col["key two"], String) == '"value2"', ), - "r2" + "r2", ) @config.requirements.json_array_indexes def test_crit_simple_int(self): name = self.tables.data_table.c.name - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] # limit the rows here to avoid PG error # "cannot extract array element from a non-array", which is # fixed in 9.4 but may exist in 9.3 self._test_index_criteria( - and_(name == 'r4', cast(col[1], String) == '"two"'), - "r4" + and_(name == "r4", cast(col[1], String) == '"two"'), "r4" ) def test_crit_mixed_path(self): - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] self._test_index_criteria( - cast(col[("key3", 1, "six")], String) == '"seven"', - "r3" + cast(col[("key3", 1, "six")], String) == '"seven"', "r3" ) def test_crit_string_path(self): - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] self._test_index_criteria( cast(col[("nested", "elem2", "elem3", "elem4")], String) == '"elem5"', - "r5" + "r5", ) def test_crit_against_string_basic(self): name = self.tables.data_table.c.name - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] self._test_index_criteria( - and_(name == 'r6', cast(col["b"], String) == '"some value"'), - "r6" + and_(name == "r6", cast(col["b"], String) == '"some value"'), "r6" ) def test_crit_against_string_coerce_type(self): name = self.tables.data_table.c.name - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] self._test_index_criteria( - and_(name == 'r6', - cast(col["b"], String) == type_coerce("some value", JSON)), + and_( + name == "r6", + cast(col["b"], String) == type_coerce("some value", JSON), + ), "r6", - test_literal=False + test_literal=False, ) def test_crit_against_int_basic(self): name = self.tables.data_table.c.name - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] self._test_index_criteria( - and_(name == 'r6', cast(col["a"], String) == '5'), - "r6" + and_(name == "r6", cast(col["a"], String) == "5"), "r6" ) def test_crit_against_int_coerce_type(self): name = self.tables.data_table.c.name - col = self.tables.data_table.c['data'] + col = self.tables.data_table.c["data"] self._test_index_criteria( - and_(name == 'r6', cast(col["a"], String) == type_coerce(5, JSON)), + and_(name == "r6", cast(col["a"], String) == type_coerce(5, JSON)), "r6", - test_literal=False + test_literal=False, ) def test_unicode_round_trip(self): @@ -961,17 +857,17 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): { "name": "r1", "data": { - util.u('réveillé'): util.u('réveillé'), - "data": {"k1": util.u('drôle')} - } - } + util.u("réveillé"): util.u("réveillé"), + "data": {"k1": util.u("drôle")}, + }, + }, ) eq_( conn.scalar(select([self.tables.data_table.c.data])), { - util.u('réveillé'): util.u('réveillé'), - "data": {"k1": util.u('drôle')} + util.u("réveillé"): util.u("réveillé"), + "data": {"k1": util.u("drôle")}, }, ) @@ -986,7 +882,7 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): s = Session(testing.db) - d1 = Data(name='d1', data=None, nulldata=None) + d1 = Data(name="d1", data=None, nulldata=None) s.add(d1) s.commit() @@ -995,24 +891,46 @@ class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): ) eq_( s.query( - cast(self.tables.data_table.c.data, String(convert_unicode="force")), - cast(self.tables.data_table.c.nulldata, String) - ).filter(self.tables.data_table.c.name == 'd1').first(), - ("null", None) + cast( + self.tables.data_table.c.data, + String(convert_unicode="force"), + ), + cast(self.tables.data_table.c.nulldata, String), + ) + .filter(self.tables.data_table.c.name == "d1") + .first(), + ("null", None), ) eq_( s.query( - cast(self.tables.data_table.c.data, String(convert_unicode="force")), - cast(self.tables.data_table.c.nulldata, String) - ).filter(self.tables.data_table.c.name == 'd2').first(), - ("null", None) - ) - - -__all__ = ('UnicodeVarcharTest', 'UnicodeTextTest', 'JSONTest', - 'DateTest', 'DateTimeTest', 'TextTest', - 'NumericTest', 'IntegerTest', - 'DateTimeHistoricTest', 'DateTimeCoercedToDateTimeTest', - 'TimeMicrosecondsTest', 'TimestampMicrosecondsTest', 'TimeTest', - 'DateTimeMicrosecondsTest', - 'DateHistoricTest', 'StringTest', 'BooleanTest') + cast( + self.tables.data_table.c.data, + String(convert_unicode="force"), + ), + cast(self.tables.data_table.c.nulldata, String), + ) + .filter(self.tables.data_table.c.name == "d2") + .first(), + ("null", None), + ) + + +__all__ = ( + "UnicodeVarcharTest", + "UnicodeTextTest", + "JSONTest", + "DateTest", + "DateTimeTest", + "TextTest", + "NumericTest", + "IntegerTest", + "DateTimeHistoricTest", + "DateTimeCoercedToDateTimeTest", + "TimeMicrosecondsTest", + "TimestampMicrosecondsTest", + "TimeTest", + "DateTimeMicrosecondsTest", + "DateHistoricTest", + "StringTest", + "BooleanTest", +) diff --git a/lib/sqlalchemy/testing/suite/test_update_delete.py b/lib/sqlalchemy/testing/suite/test_update_delete.py index e4c61e74a..b232c3a78 100644 --- a/lib/sqlalchemy/testing/suite/test_update_delete.py +++ b/lib/sqlalchemy/testing/suite/test_update_delete.py @@ -6,15 +6,17 @@ from ..schema import Table, Column class SimpleUpdateDeleteTest(fixtures.TablesTest): - run_deletes = 'each' + run_deletes = "each" __backend__ = True @classmethod def define_tables(cls, metadata): - Table('plain_pk', metadata, - Column('id', Integer, primary_key=True), - Column('data', String(50)) - ) + Table( + "plain_pk", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), + ) @classmethod def insert_data(cls): @@ -24,40 +26,29 @@ class SimpleUpdateDeleteTest(fixtures.TablesTest): {"id": 1, "data": "d1"}, {"id": 2, "data": "d2"}, {"id": 3, "data": "d3"}, - ] + ], ) def test_update(self): t = self.tables.plain_pk - r = config.db.execute( - t.update().where(t.c.id == 2), - data="d2_new" - ) + r = config.db.execute(t.update().where(t.c.id == 2), data="d2_new") assert not r.is_insert assert not r.returns_rows eq_( config.db.execute(t.select().order_by(t.c.id)).fetchall(), - [ - (1, "d1"), - (2, "d2_new"), - (3, "d3") - ] + [(1, "d1"), (2, "d2_new"), (3, "d3")], ) def test_delete(self): t = self.tables.plain_pk - r = config.db.execute( - t.delete().where(t.c.id == 2) - ) + r = config.db.execute(t.delete().where(t.c.id == 2)) assert not r.is_insert assert not r.returns_rows eq_( config.db.execute(t.select().order_by(t.c.id)).fetchall(), - [ - (1, "d1"), - (3, "d3") - ] + [(1, "d1"), (3, "d3")], ) -__all__ = ('SimpleUpdateDeleteTest', ) + +__all__ = ("SimpleUpdateDeleteTest",) |
