From: Gord Thompson Date: Sat, 5 Sep 2020 18:22:43 +0000 (-0600) Subject: Modernize some .execute calls in EnumSetTest X-Git-Tag: rel_1_4_0b1~124^2 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=5b6dfc0c38bf1f01da4b8;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git Modernize some .execute calls in EnumSetTest Change-Id: I9bbc3be4ec1797686ddf4a559f5b9e1679c62f58 --- diff --git a/test/dialect/mysql/test_types.py b/test/dialect/mysql/test_types.py index bc5619cdbe..60c4f0ec06 100644 --- a/test/dialect/mysql/test_types.py +++ b/test/dialect/mysql/test_types.py @@ -475,7 +475,7 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): # fixed in mysql-connector as of 2.0.1, # see http://bugs.mysql.com/bug.php?id=73266 @testing.provide_metadata - def test_precision_float_roundtrip(self): + def test_precision_float_roundtrip(self, connection): t = Table( "t", self.metadata, @@ -488,24 +488,20 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): mysql.DOUBLE(decimal_return_scale=12, asdecimal=True), ), ) - with testing.db.connect() as conn: - t.create(conn) - conn.execute( - t.insert(), - scale_value=45.768392065789, - unscale_value=45.768392065789, - ) - result = conn.scalar(select(t.c.scale_value)) - eq_(result, decimal.Decimal("45.768392065789")) + t.create(connection) + connection.execute( + t.insert(), + dict(scale_value=45.768392065789, unscale_value=45.768392065789,), + ) + result = connection.scalar(select(t.c.scale_value)) + eq_(result, decimal.Decimal("45.768392065789")) - result = conn.scalar(select(t.c.unscale_value)) - eq_(result, decimal.Decimal("45.768392065789")) + result = connection.scalar(select(t.c.unscale_value)) + eq_(result, decimal.Decimal("45.768392065789")) @testing.only_if("mysql") - @testing.fails_on("mysql+mysqlconnector", "different unicode behavior") - @testing.exclude("mysql", "<", (5, 0, 5), "a 5.0+ feature") @testing.provide_metadata - def test_charset_collate_table(self): + def test_charset_collate_table(self, connection): t = Table( "foo", self.metadata, @@ -514,9 +510,8 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): mysql_default_charset="utf8", mysql_collate="utf8_bin", ) - t.create() - m2 = MetaData(testing.db) - t2 = Table("foo", m2, autoload=True) + t.create(connection) + t2 = Table("foo", MetaData(), autoload_with=connection) eq_(t2.kwargs["mysql_collate"], "utf8_bin") eq_(t2.kwargs["mysql_default charset"], "utf8") @@ -524,8 +519,8 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): # in order to test the condition here, need to use # MySQLdb 1.2.3 and also need to pass either use_unicode=1 # or charset=utf8 to the URL. - t.insert().execute(id=1, data=u("some text")) - assert isinstance(testing.db.scalar(select(t.c.data)), util.text_type) + connection.execute(t.insert(), dict(id=1, data=u("some text"))) + assert isinstance(connection.scalar(select(t.c.data)), util.text_type) @testing.metadata_fixture(ddl="class") def bit_table(self, metadata): @@ -556,17 +551,14 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): ([0, 0, 0, 0, 0, 0, 0, l], None), argnames="store, expected", ) - def test_bit_50_roundtrip(self, bit_table, store, expected): + def test_bit_50_roundtrip(self, connection, bit_table, store, expected): - meta2 = MetaData(testing.db) - reflected = Table("mysql_bits", meta2, autoload=True) + reflected = Table("mysql_bits", MetaData(), autoload_with=connection) - with testing.db.connect() as conn: - expected = expected or store - conn.execute(reflected.insert(store)) - row = conn.execute(reflected.select()).first() - eq_(list(row), expected) - conn.execute(reflected.delete()) + expected = expected or store + connection.execute(reflected.insert().values(store)) + row = connection.execute(reflected.select()).first() + eq_(list(row), expected) @testing.combinations( (([0] * 8), None), @@ -579,16 +571,15 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): ([0, 0, 0, 0, 0, 0, 0, l], None), argnames="store, expected", ) - def test_bit_50_roundtrip_reflected(self, bit_table, store, expected): - meta2 = MetaData() - bit_table = Table("mysql_bits", meta2, autoload_with=testing.db) + def test_bit_50_roundtrip_reflected( + self, connection, bit_table, store, expected + ): + bit_table = Table("mysql_bits", MetaData(), autoload_with=connection) - with testing.db.connect() as conn: - expected = expected or store - conn.execute(bit_table.insert(store)) - row = conn.execute(bit_table.select()).first() - eq_(list(row), expected) - conn.execute(bit_table.delete()) + expected = expected or store + connection.execute(bit_table.insert().values(store)) + row = connection.execute(bit_table.select()).first() + eq_(list(row), expected) @testing.metadata_fixture(ddl="class") def boolean_table(self, metadata): @@ -611,18 +602,18 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): ([False, False, 0, 0, 0], [False, False, 0, 0, 0]), argnames="store, expected", ) - def test_boolean_roundtrip(self, boolean_table, store, expected): + def test_boolean_roundtrip( + self, connection, boolean_table, store, expected + ): table = boolean_table - with testing.db.connect() as conn: - expected = expected or store - conn.execute(table.insert(store)) - row = conn.execute(table.select()).first() - eq_(list(row), expected) - for i, val in enumerate(expected): - if isinstance(val, bool): - self.assert_(val is row[i]) - conn.execute(table.delete()) + expected = expected or store + connection.execute(table.insert().values(store)) + row = connection.execute(table.select()).first() + eq_(list(row), expected) + for i, val in enumerate(expected): + if isinstance(val, bool): + self.assert_(val is row[i]) @testing.combinations( ([None, None, None, None, None], None), @@ -632,34 +623,32 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): ([False, False, 0, 0, 0], [False, False, False, False, 0]), argnames="store, expected", ) - def test_boolean_roundtrip_reflected(self, boolean_table, store, expected): - meta2 = MetaData(testing.db) - table = Table("mysql_bool", meta2, autoload=True) + def test_boolean_roundtrip_reflected( + self, connection, boolean_table, store, expected + ): + table = Table("mysql_bool", MetaData(), autoload_with=connection) eq_(colspec(table.c.b3), "b3 TINYINT(1)") eq_regex(colspec(table.c.b4), r"b4 TINYINT(?:\(1\))? UNSIGNED") - meta2 = MetaData(testing.db) table = Table( "mysql_bool", - meta2, + MetaData(), Column("b1", BOOLEAN), Column("b2", Boolean), Column("b3", BOOLEAN), Column("b4", BOOLEAN), - autoload=True, + autoload_with=connection, ) eq_(colspec(table.c.b3), "b3 BOOL") eq_(colspec(table.c.b4), "b4 BOOL") - with testing.db.connect() as conn: - expected = expected or store - conn.execute(table.insert(store)) - row = conn.execute(table.select()).first() - eq_(list(row), expected) - for i, val in enumerate(expected): - if isinstance(val, bool): - self.assert_(val is row[i]) - conn.execute(table.delete()) + expected = expected or store + connection.execute(table.insert().values(store)) + row = connection.execute(table.select()).first() + eq_(list(row), expected) + for i, val in enumerate(expected): + if isinstance(val, bool): + self.assert_(val is row[i]) class MyTime(TypeDecorator): impl = TIMESTAMP @@ -711,22 +700,20 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): [(now, now), (None, now), (None, now)], ) - @testing.fails_on("mysql+oursql", "TODO: probable OurSQL bug") @testing.provide_metadata - def test_time_roundtrip(self): + def test_time_roundtrip(self, connection): t = Table("mysql_time", self.metadata, Column("t1", mysql.TIME())) - with testing.db.connect() as conn: - t.create(conn) + t.create(connection) - conn.execute(t.insert().values(t1=datetime.time(8, 37, 35))) - eq_( - conn.execute(select(t.c.t1)).scalar(), - datetime.time(8, 37, 35), - ) + connection.execute(t.insert().values(t1=datetime.time(8, 37, 35))) + eq_( + connection.execute(select(t.c.t1)).scalar(), + datetime.time(8, 37, 35), + ) @testing.provide_metadata - def test_year(self): + def test_year(self, connection): """Exercise YEAR.""" year_table = Table( @@ -740,17 +727,17 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults): for col in year_table.c: self.assert_(repr(col)) - year_table.create() - reflected = Table("mysql_year", MetaData(testing.db), autoload=True) + year_table.create(connection) + reflected = Table("mysql_year", MetaData(), autoload_with=connection) for table in year_table, reflected: - with testing.db.connect() as conn: - conn.execute(table.insert(["1950", "50", None, 1950])) - row = conn.execute(table.select()).first() - eq_(list(row), [1950, 2050, None, 1950]) - conn.execute(table.delete()) - self.assert_(colspec(table.c.y1).startswith("y1 YEAR")) - eq_regex(colspec(table.c.y5), r"y5 YEAR(?:\(4\))?") + connection.execute( + table.insert().values(["1950", "50", None, 1950]) + ) + row = connection.execute(table.select()).first() + eq_(list(row), [1950, 2050, None, 1950]) + self.assert_(colspec(table.c.y1).startswith("y1 YEAR")) + eq_regex(colspec(table.c.y5), r"y5 YEAR(?:\(4\))?") class JSONTest(fixtures.TestBase): @@ -760,31 +747,30 @@ class JSONTest(fixtures.TestBase): @testing.provide_metadata @testing.requires.reflects_json_type - def test_reflection(self): + def test_reflection(self, connection): Table("mysql_json", self.metadata, Column("foo", mysql.JSON)) - self.metadata.create_all() + self.metadata.create_all(connection) - reflected = Table("mysql_json", MetaData(), autoload_with=testing.db) + reflected = Table("mysql_json", MetaData(), autoload_with=connection) is_(reflected.c.foo.type._type_affinity, sqltypes.JSON) assert isinstance(reflected.c.foo.type, mysql.JSON) @testing.provide_metadata - def test_rudimental_round_trip(self): + def test_rudimental_round_trip(self, connection): # note that test_suite has many more JSON round trip tests # using the backend-agnostic JSON type mysql_json = Table( "mysql_json", self.metadata, Column("foo", mysql.JSON) ) - self.metadata.create_all() + self.metadata.create_all(connection) value = {"json": {"foo": "bar"}, "recs": ["one", "two"]} - with testing.db.connect() as conn: - conn.execute(mysql_json.insert(), foo=value) + connection.execute(mysql_json.insert(), dict(foo=value)) - eq_(conn.scalar(select(mysql_json.c.foo)), value) + eq_(connection.scalar(select(mysql_json.c.foo)), value) class EnumSetTest( @@ -816,7 +802,7 @@ class EnumSetTest( return [str(v.value) for v in some_enum.__members__.values()] @testing.provide_metadata - def test_enum(self): + def test_enum(self, connection): """Exercise the ENUM type.""" e1 = mysql.ENUM("a", "b") @@ -864,60 +850,61 @@ class EnumSetTest( colspec(enum_table.c.e8), "e8 ENUM('one','two','three','AMember','BMember')", ) - enum_table.create() - - assert_raises( - exc.DBAPIError, - enum_table.insert().execute, - e1=None, - e2=None, - e3=None, - e4=None, - ) + enum_table.create(connection) assert enum_table.c.e2generic.type.validate_strings + assert_raises( exc.StatementError, - enum_table.insert().execute, - e1="c", - e2="c", - e2generic="c", - e3="c", - e4="c", - e5="c", - e5generic="c", - e6="c", - e7="c", - e8="c", + connection.execute, + enum_table.insert(), + dict( + e1="c", + e2="c", + e2generic="c", + e3="c", + e4="c", + e5="c", + e5generic="c", + e6="c", + e7="c", + e8="c", + ), ) - enum_table.insert().execute() - enum_table.insert().execute( - e1="a", - e2="a", - e2generic="a", - e3="a", - e4="a", - e5="a", - e5generic="a", - e6="'a'", - e7="a", - e8="AMember", + connection.execute(enum_table.insert()) + connection.execute( + enum_table.insert(), + dict( + e1="a", + e2="a", + e2generic="a", + e3="a", + e4="a", + e5="a", + e5generic="a", + e6="'a'", + e7="a", + e8="AMember", + ), ) - enum_table.insert().execute( - e1="b", - e2="b", - e2generic="b", - e3="b", - e4="b", - e5="b", - e5generic="b", - e6="b", - e7="b", - e8="BMember", + connection.execute( + enum_table.insert(), + dict( + e1="b", + e2="b", + e2generic="b", + e3="b", + e4="b", + e5="b", + e5generic="b", + e6="b", + e7="b", + e8="BMember", + ), ) - res = enum_table.select().execute().fetchall() + res = connection.execute(enum_table.select()).fetchall() expected = [ (None, "a", "a", None, "a", None, None, None, None, None), @@ -976,54 +963,39 @@ class EnumSetTest( eq_(colspec(set_table.c.e5), "e5 SET('a','b')") @testing.provide_metadata - def test_no_null(self): + def test_no_null(self, connection): set_table = self._set_fixture_one() - set_table.create() + set_table.create(connection) assert_raises( exc.DBAPIError, - set_table.insert().execute, - e1=None, - e2=None, - e3=None, - e4=None, - ) - - @testing.only_on("+oursql") - @testing.provide_metadata - def test_oursql_error_one(self): - set_table = self._set_fixture_one() - set_table.create() - assert_raises( - exc.StatementError, - set_table.insert().execute, - e1="c", - e2="c", - e3="c", - e4="c", + connection.execute, + set_table.insert(), + dict(e1=None, e2=None, e3=None, e4=None), ) @testing.requires.mysql_non_strict @testing.provide_metadata - def test_empty_set_no_empty_string(self): + def test_empty_set_no_empty_string(self, connection): t = Table( "t", self.metadata, Column("id", Integer), Column("data", mysql.SET("a", "b")), ) - t.create() - with testing.db.begin() as conn: - conn.execute( - t.insert(), + t.create(connection) + connection.execute( + t.insert(), + [ {"id": 1, "data": set()}, {"id": 2, "data": set([""])}, {"id": 3, "data": set(["a", ""])}, {"id": 4, "data": set(["b"])}, - ) - eq_( - conn.execute(t.select().order_by(t.c.id)).fetchall(), - [(1, set()), (2, set()), (3, set(["a"])), (4, set(["b"]))], - ) + ], + ) + eq_( + connection.execute(t.select().order_by(t.c.id)).fetchall(), + [(1, set()), (2, set()), (3, set(["a"])), (4, set(["b"]))], + ) def test_bitwise_required_for_empty(self): assert_raises_message( @@ -1037,68 +1009,61 @@ class EnumSetTest( ) @testing.provide_metadata - def test_empty_set_empty_string(self): + def test_empty_set_empty_string(self, connection): t = Table( "t", self.metadata, Column("id", Integer), Column("data", mysql.SET("a", "b", "", retrieve_as_bitwise=True)), ) - t.create() - with testing.db.begin() as conn: - conn.execute( - t.insert(), + t.create(connection) + connection.execute( + t.insert(), + [ {"id": 1, "data": set()}, {"id": 2, "data": set([""])}, {"id": 3, "data": set(["a", ""])}, {"id": 4, "data": set(["b"])}, - ) - eq_( - conn.execute(t.select().order_by(t.c.id)).fetchall(), - [ - (1, set()), - (2, set([""])), - (3, set(["a", ""])), - (4, set(["b"])), - ], - ) + ], + ) + eq_( + connection.execute(t.select().order_by(t.c.id)).fetchall(), + [ + (1, set()), + (2, set([""])), + (3, set(["a", ""])), + (4, set(["b"])), + ], + ) @testing.provide_metadata - def test_string_roundtrip(self): + def test_string_roundtrip(self, connection): set_table = self._set_fixture_one() - set_table.create() - with testing.db.begin() as conn: - conn.execute( - set_table.insert(), - dict(e1="a", e2="a", e3="a", e4="'a'", e5="a,b"), - ) - conn.execute( - set_table.insert(), - dict(e1="b", e2="b", e3="b", e4="b", e5="a,b"), - ) + set_table.create(connection) + connection.execute( + set_table.insert(), + dict(e1="a", e2="a", e3="a", e4="'a'", e5="a,b"), + ) + connection.execute( + set_table.insert(), dict(e1="b", e2="b", e3="b", e4="b", e5="a,b"), + ) - expected = [ - ( - set(["a"]), - set(["a"]), - set(["a"]), - set(["'a'"]), - set(["a", "b"]), - ), - ( - set(["b"]), - set(["b"]), - set(["b"]), - set(["b"]), - set(["a", "b"]), - ), - ] - res = conn.execute(set_table.select()).fetchall() + expected = [ + ( + set(["a"]), + set(["a"]), + set(["a"]), + set(["'a'"]), + set(["a", "b"]), + ), + (set(["b"]), set(["b"]), set(["b"]), set(["b"]), set(["a", "b"]),), + ] + res = connection.execute(set_table.select()).fetchall() - eq_(res, expected) + eq_(res, expected) @testing.provide_metadata - def test_unicode_roundtrip(self): + def test_unicode_roundtrip(self, connection): set_table = Table( "t", self.metadata, @@ -1106,38 +1071,36 @@ class EnumSetTest( Column("data", mysql.SET(u("réveillé"), u("drôle"), u("S’il"))), ) - set_table.create() - with testing.db.begin() as conn: - conn.execute( - set_table.insert(), {"data": set([u("réveillé"), u("drôle")])} - ) + set_table.create(connection) + connection.execute( + set_table.insert(), {"data": set([u("réveillé"), u("drôle")])} + ) - row = conn.execute(set_table.select()).first() + row = connection.execute(set_table.select()).first() - eq_(row, (1, set([u("réveillé"), u("drôle")]))) + eq_(row, (1, set([u("réveillé"), u("drôle")]))) @testing.provide_metadata - def test_int_roundtrip(self): + def test_int_roundtrip(self, connection): set_table = self._set_fixture_one() - set_table.create() - with testing.db.begin() as conn: - conn.execute( - set_table.insert(), dict(e1=1, e2=2, e3=3, e4=3, e5=0) - ) - res = conn.execute(set_table.select()).first() - eq_( - res, - ( - set(["a"]), - set(["b"]), - set(["a", "b"]), - set(["'a'", "b"]), - set([]), - ), - ) + set_table.create(connection) + connection.execute( + set_table.insert(), dict(e1=1, e2=2, e3=3, e4=3, e5=0) + ) + res = connection.execute(set_table.select()).first() + eq_( + res, + ( + set(["a"]), + set(["b"]), + set(["a", "b"]), + set(["'a'", "b"]), + set([]), + ), + ) @testing.provide_metadata - def test_set_roundtrip_plus_reflection(self): + def test_set_roundtrip_plus_reflection(self, connection): set_table = Table( "mysql_set", self.metadata, @@ -1149,16 +1112,16 @@ class EnumSetTest( eq_(colspec(set_table.c.s1), "s1 SET('dq','sq')") eq_(colspec(set_table.c.s2), "s2 SET('a')") eq_(colspec(set_table.c.s3), "s3 SET('5','7','9')") - set_table.create() - reflected = Table("mysql_set", MetaData(testing.db), autoload=True) + set_table.create(connection) + reflected = Table("mysql_set", MetaData(), autoload_with=connection) for table in set_table, reflected: def roundtrip(store, expected=None): expected = expected or store - table.insert(store).execute() - row = table.select().execute().first() + connection.execute(table.insert().values(store)) + row = connection.execute(table.select()).first() eq_(row, tuple(expected)) - table.delete().execute() + connection.execute(table.delete()) roundtrip([None, None, None], [None] * 3) roundtrip(["", "", ""], [set([])] * 3) @@ -1166,25 +1129,26 @@ class EnumSetTest( roundtrip(["dq", "a", "5"], [set(["dq"]), set(["a"]), set(["5"])]) roundtrip([1, 1, 1], [set(["dq"]), set(["a"]), set(["5"])]) roundtrip([set(["dq", "sq"]), None, set(["9", "5", "7"])]) - set_table.insert().execute( - {"s3": set(["5"])}, - {"s3": set(["5", "7"])}, - {"s3": set(["5", "7", "9"])}, - {"s3": set(["7", "9"])}, + connection.execute( + set_table.insert(), + [ + {"s3": set(["5"])}, + {"s3": set(["5", "7"])}, + {"s3": set(["5", "7", "9"])}, + {"s3": set(["7", "9"])}, + ], ) - rows = ( - select( - [set_table.c.s3], set_table.c.s3.in_([set(["5"]), ["5", "7"]]) + rows = connection.execute( + select(set_table.c.s3).where( + set_table.c.s3.in_([set(["5"]), ["5", "7"]]) ) - .execute() - .fetchall() - ) - found = set([frozenset(row[0]) for row in rows]) - eq_(found, set([frozenset(["5"]), frozenset(["5", "7"])])) + ).fetchall() + + eq_(list(rows), [({"5"},), ({"7", "5"},)]) @testing.provide_metadata - def test_unicode_enum(self): + def test_unicode_enum(self, connection): metadata = self.metadata t1 = Table( "table", @@ -1193,12 +1157,18 @@ class EnumSetTest( Column("value", Enum(u("réveillé"), u("drôle"), u("S’il"))), Column("value2", mysql.ENUM(u("réveillé"), u("drôle"), u("S’il"))), ) - metadata.create_all() - t1.insert().execute(value=u("drôle"), value2=u("drôle")) - t1.insert().execute(value=u("réveillé"), value2=u("réveillé")) - t1.insert().execute(value=u("S’il"), value2=u("S’il")) + metadata.create_all(connection) + + connection.execute( + t1.insert(), + [ + dict(value=u("drôle"), value2=u("drôle")), + dict(value=u("réveillé"), value2=u("réveillé")), + dict(value=u("S’il"), value2=u("S’il")), + ], + ) eq_( - t1.select().order_by(t1.c.id).execute().fetchall(), + connection.execute(t1.select().order_by(t1.c.id)).fetchall(), [ (1, u("drôle"), u("drôle")), (2, u("réveillé"), u("réveillé")), @@ -1208,8 +1178,7 @@ class EnumSetTest( # test reflection of the enum labels - m2 = MetaData(testing.db) - t2 = Table("table", m2, autoload=True) + t2 = Table("table", MetaData(), autoload_with=connection) # TODO: what's wrong with the last element ? is there # latin-1 stuff forcing its way in ? @@ -1245,8 +1214,7 @@ class EnumSetTest( ) @testing.provide_metadata - @testing.exclude("mysql", "<", (4,), "3.23 can't handle an ENUM of ''") - def test_enum_parse(self): + def test_enum_parse(self, connection): enum_table = Table( "mysql_enum", @@ -1263,8 +1231,8 @@ class EnumSetTest( for col in enum_table.c: self.assert_(repr(col)) - enum_table.create() - reflected = Table("mysql_enum", MetaData(testing.db), autoload=True) + enum_table.create(connection) + reflected = Table("mysql_enum", MetaData(), autoload_with=connection) for t in enum_table, reflected: eq_(t.c.e1.type.enums, ["a"]) eq_(t.c.e2.type.enums, [""]) @@ -1275,8 +1243,7 @@ class EnumSetTest( eq_(t.c.e7.type.enums, ["", "'a'", "b'b", "'"]) @testing.provide_metadata - @testing.exclude("mysql", "<", (5,)) - def test_set_parse(self): + def test_set_parse(self, connection): set_table = Table( "mysql_set", self.metadata, @@ -1295,10 +1262,10 @@ class EnumSetTest( for col in set_table.c: self.assert_(repr(col)) - set_table.create() + set_table.create(connection) # don't want any warnings on reflection - reflected = Table("mysql_set", MetaData(testing.db), autoload=True) + reflected = Table("mysql_set", MetaData(), autoload_with=connection) for t in set_table, reflected: eq_(t.c.e1.type.values, ("a",)) eq_(t.c.e2.type.values, ("",)) @@ -1310,7 +1277,7 @@ class EnumSetTest( @testing.requires.mysql_non_strict @testing.provide_metadata - def test_broken_enum_returns_blanks(self): + def test_broken_enum_returns_blanks(self, connection): t = Table( "enum_missing", self.metadata, @@ -1318,22 +1285,24 @@ class EnumSetTest( Column("e1", sqltypes.Enum("one", "two", "three")), Column("e2", mysql.ENUM("one", "two", "three")), ) - t.create() + t.create(connection) - with testing.db.connect() as conn: - conn.execute( - t.insert(), {"e1": "nonexistent", "e2": "nonexistent"} - ) - conn.execute(t.insert(), {"e1": "", "e2": ""}) - conn.execute(t.insert(), {"e1": "two", "e2": "two"}) - conn.execute(t.insert(), {"e1": None, "e2": None}) + connection.execute( + t.insert(), + [ + {"e1": "nonexistent", "e2": "nonexistent"}, + {"e1": "", "e2": ""}, + {"e1": "two", "e2": "two"}, + {"e1": None, "e2": None}, + ], + ) - eq_( - conn.execute( - select(t.c.e1, t.c.e2).order_by(t.c.id) - ).fetchall(), - [("", ""), ("", ""), ("two", "two"), (None, None)], - ) + eq_( + connection.execute( + select(t.c.e1, t.c.e2).order_by(t.c.id) + ).fetchall(), + [("", ""), ("", ""), ("two", "two"), (None, None)], + ) def colspec(c):