Ensure MySQL tests cleanup connections and raise an exception if they do not

tests/components/recorder/test_init.py

@@ -103,7 +103,6 @@ async def test_shutdown_before_startup_finishes(
     tmp_path,
 ):
     """Test shutdown before recorder starts is clean."""
-
     if recorder_db_url == "sqlite://":
         # On-disk database because this test does not play nice with the
         # MutexPool
@@ -1529,6 +1528,10 @@ async def test_database_lock_and_unlock(
     tmp_path,
 ):
     """Test writing events during lock getting written after unlocking."""
+    if recorder_db_url.startswith(("mysql://", "postgresql://")):
+        # Database locking is only used for SQLite
+        return
+
     if recorder_db_url == "sqlite://":
         # Use file DB, in memory DB cannot do write locks.
         recorder_db_url = "sqlite:///" + str(tmp_path / "pytest.db")
@@ -1574,6 +1577,10 @@ async def test_database_lock_and_overflow(
     tmp_path,
 ):
     """Test writing events during lock leading to overflow the queue causes the database to unlock."""
+    if recorder_db_url.startswith(("mysql://", "postgresql://")):
+        # Database locking is only used for SQLite
+        return
+
     # Use file DB, in memory DB cannot do write locks.
     if recorder_db_url == "sqlite://":
         # Use file DB, in memory DB cannot do write locks.

tests/conftest.py

@@ -1066,7 +1066,26 @@ def _ha_orm_quote(mixed, ident):
         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)