Terminate stale MySQL connections at the end of test runs (#87794)

tests/components/recorder/test_init.py

@@ -103,10 +103,6 @@ async def test_shutdown_before_startup_finishes(
     tmp_path,
 ):
     """Test shutdown before recorder starts is clean."""
-    if recorder_db_url.startswith("mysql://"):
-        # Currently this test fails with MySQL
-        return
-
     if recorder_db_url == "sqlite://":
         # On-disk database because this test does not play nice with the
         # MutexPool

tests/conftest.py

@@ -1066,7 +1066,26 @@ def recorder_db_url(pytestconfig):
         assert not sqlalchemy_utils.database_exists(db_url)
         sqlalchemy_utils.create_database(db_url, encoding="utf8")
     yield db_url
-    if db_url.startswith("mysql://") or db_url.startswith("postgresql://"):
+    if db_url.startswith("mysql://"):
+        import sqlalchemy as sa
+
+        made_url = sa.make_url(db_url)
+        db = made_url.database
+        engine = sa.create_engine(db_url)
+        # Kill any open connections to the database before dropping it
+        # to ensure that InnoDB does not deadlock.
+        with engine.begin() as connection:
+            query = sa.text(
+                "select id FROM information_schema.processlist WHERE db=:db and id != CONNECTION_ID()"
+            )
+            for row in connection.execute(query, parameters={"db": db}).fetchall():
+                _LOGGER.warning(
+                    "Killing MySQL connection to temporary database %s", row.id
+                )
+                connection.execute(sa.text("KILL :id"), parameters={"id": row.id})
+        engine.dispose()
+        sqlalchemy_utils.drop_database(db_url)
+    elif db_url.startswith("postgresql://"):
         sqlalchemy_utils.drop_database(db_url)