Modernize some .execute calls in EnumSetTest

Change-Id: I9bbc3be4ec1797686ddf4a559f5b9e1679c62f58

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):