From: Gord Thompson Date: Sun, 12 Apr 2020 13:03:25 +0000 (-0600) Subject: Clean up .execute calls in remaining suite tests X-Git-Tag: rel_1_4_0b1~388^2 X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=f564a15feca1dcead62a12b496940a863d5165c1;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git Clean up .execute calls in remaining suite tests Change-Id: Ib5c7f46067bcf5b162060476cc323bf671db101a --- diff --git a/lib/sqlalchemy/testing/suite/test_cte.py b/lib/sqlalchemy/testing/suite/test_cte.py index c7e6a266ca..fab4576067 100644 --- a/lib/sqlalchemy/testing/suite/test_cte.py +++ b/lib/sqlalchemy/testing/suite/test_cte.py @@ -37,16 +37,17 @@ class CTETest(fixtures.TablesTest): @classmethod def insert_data(cls): - config.db.execute( - cls.tables.some_table.insert(), - [ - {"id": 1, "data": "d1", "parent_id": None}, - {"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}, - ], - ) + with config.db.connect() as conn: + conn.execute( + cls.tables.some_table.insert(), + [ + {"id": 1, "data": "d1", "parent_id": None}, + {"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}, + ], + ) def test_select_nonrecursive_round_trip(self): some_table = self.tables.some_table diff --git a/lib/sqlalchemy/testing/suite/test_ddl.py b/lib/sqlalchemy/testing/suite/test_ddl.py index 81a55e18ae..1f49106fb6 100644 --- a/lib/sqlalchemy/testing/suite/test_ddl.py +++ b/lib/sqlalchemy/testing/suite/test_ddl.py @@ -67,25 +67,27 @@ class TableDDLTest(fixtures.TestBase): @requirements.comment_reflection @util.provide_metadata - def test_add_table_comment(self): + def test_add_table_comment(self, connection): table = self._simple_fixture() - table.create(config.db, checkfirst=False) + table.create(connection, checkfirst=False) table.comment = "a comment" - config.db.execute(schema.SetTableComment(table)) + connection.execute(schema.SetTableComment(table)) eq_( - inspect(config.db).get_table_comment("test_table"), + inspect(connection).get_table_comment("test_table"), {"text": "a comment"}, ) @requirements.comment_reflection @util.provide_metadata - def test_drop_table_comment(self): + def test_drop_table_comment(self, connection): table = self._simple_fixture() - table.create(config.db, checkfirst=False) + table.create(connection, checkfirst=False) table.comment = "a comment" - config.db.execute(schema.SetTableComment(table)) - config.db.execute(schema.DropTableComment(table)) - eq_(inspect(config.db).get_table_comment("test_table"), {"text": None}) + connection.execute(schema.SetTableComment(table)) + connection.execute(schema.DropTableComment(table)) + eq_( + inspect(connection).get_table_comment("test_table"), {"text": None} + ) __all__ = ("TableDDLTest",) diff --git a/lib/sqlalchemy/testing/suite/test_deprecations.py b/lib/sqlalchemy/testing/suite/test_deprecations.py index d0202a0a95..126d82fe97 100644 --- a/lib/sqlalchemy/testing/suite/test_deprecations.py +++ b/lib/sqlalchemy/testing/suite/test_deprecations.py @@ -24,20 +24,21 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): @classmethod def insert_data(cls): - config.db.execute( - cls.tables.some_table.insert(), - [ - {"id": 1, "x": 1, "y": 2}, - {"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) - - def test_plain_union(self): + with config.db.connect() as conn: + conn.execute( + cls.tables.some_table.insert(), + [ + {"id": 1, "x": 1, "y": 2}, + {"id": 2, "x": 2, "y": 3}, + {"id": 3, "x": 3, "y": 4}, + {"id": 4, "x": 4, "y": 5}, + ], + ) + + def _assert_result(self, conn, select, result, params=()): + eq_(conn.execute(select, params).fetchall(), result) + + def test_plain_union(self, connection): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2) s2 = select([table]).where(table.c.id == 3) @@ -47,7 +48,9 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): "The SelectBase.c and SelectBase.columns " "attributes are deprecated" ): - self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) + self._assert_result( + connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] + ) # note we've had to remove one use case entirely, which is this # one. the Select gets its FROMS from the WHERE clause and the @@ -56,7 +59,7 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): # ORDER BY without adding the SELECT into the FROM and breaking the # query. Users will have to adjust for this use case if they were doing # it before. - def _dont_test_select_from_plain_union(self): + def _dont_test_select_from_plain_union(self, connection): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2) s2 = select([table]).where(table.c.id == 3) @@ -66,11 +69,13 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): "The SelectBase.c and SelectBase.columns " "attributes are deprecated" ): - self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) + self._assert_result( + connection, 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): + def test_limit_offset_selectable_in_unions(self, connection): table = self.tables.some_table s1 = ( select([table]) @@ -90,10 +95,12 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): "The SelectBase.c and SelectBase.columns " "attributes are deprecated" ): - self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) + self._assert_result( + connection, 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): + def test_order_by_selectable_in_unions(self, connection): 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) @@ -103,9 +110,11 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): "The SelectBase.c and SelectBase.columns " "attributes are deprecated" ): - self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) + self._assert_result( + connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] + ) - def test_distinct_selectable_in_unions(self): + def test_distinct_selectable_in_unions(self, connection): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2).distinct() s2 = select([table]).where(table.c.id == 3).distinct() @@ -115,9 +124,11 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): "The SelectBase.c and SelectBase.columns " "attributes are deprecated" ): - self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) + self._assert_result( + connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] + ) - def test_limit_offset_aliased_selectable_in_unions(self): + def test_limit_offset_aliased_selectable_in_unions(self, connection): table = self.tables.some_table s1 = ( select([table]) @@ -141,4 +152,6 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest): "The SelectBase.c and SelectBase.columns " "attributes are deprecated" ): - self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) + self._assert_result( + connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] + ) diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py index 9dabdbd650..7719a3b3c3 100644 --- a/lib/sqlalchemy/testing/suite/test_types.py +++ b/lib/sqlalchemy/testing/suite/test_types.py @@ -519,10 +519,10 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): filter_=lambda n: n is not None and round(n, 5) or None, ) - def test_float_coerce_round_trip(self): + def test_float_coerce_round_trip(self, connection): expr = 15.7563 - val = testing.db.scalar(select([literal(expr)])) + val = connection.scalar(select([literal(expr)])) eq_(val, expr) # this does not work in MySQL, see #4036, however we choose not @@ -530,17 +530,17 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): @testing.requires.implicit_decimal_binds @testing.emits_warning(r".*does \*not\* support Decimal objects natively") - def test_decimal_coerce_round_trip(self): + def test_decimal_coerce_round_trip(self, connection): expr = decimal.Decimal("15.7563") - val = testing.db.scalar(select([literal(expr)])) + val = connection.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): + def test_decimal_coerce_round_trip_w_cast(self, connection): expr = decimal.Decimal("15.7563") - val = testing.db.scalar(select([cast(expr, Numeric(10, 4))])) + val = connection.scalar(select([cast(expr, Numeric(10, 4))])) eq_(val, expr) @testing.requires.precision_numerics_general diff --git a/lib/sqlalchemy/testing/suite/test_update_delete.py b/lib/sqlalchemy/testing/suite/test_update_delete.py index 97bdf0ad76..6003a09941 100644 --- a/lib/sqlalchemy/testing/suite/test_update_delete.py +++ b/lib/sqlalchemy/testing/suite/test_update_delete.py @@ -22,33 +22,34 @@ class SimpleUpdateDeleteTest(fixtures.TablesTest): @classmethod def insert_data(cls): - config.db.execute( - cls.tables.plain_pk.insert(), - [ - {"id": 1, "data": "d1"}, - {"id": 2, "data": "d2"}, - {"id": 3, "data": "d3"}, - ], - ) - - def test_update(self): + with config.db.connect() as conn: + conn.execute( + cls.tables.plain_pk.insert(), + [ + {"id": 1, "data": "d1"}, + {"id": 2, "data": "d2"}, + {"id": 3, "data": "d3"}, + ], + ) + + def test_update(self, connection): t = self.tables.plain_pk - r = config.db.execute(t.update().where(t.c.id == 2), data="d2_new") + r = connection.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(), + connection.execute(t.select().order_by(t.c.id)).fetchall(), [(1, "d1"), (2, "d2_new"), (3, "d3")], ) - def test_delete(self): + def test_delete(self, connection): t = self.tables.plain_pk - r = config.db.execute(t.delete().where(t.c.id == 2)) + r = connection.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(), + connection.execute(t.select().order_by(t.c.id)).fetchall(), [(1, "d1"), (3, "d3")], )