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

tests/components/recorder/test_init.py

@@ -130,6 +130,10 @@ async def test_shutdown_before_startup_finishes(
     assert run_info.run_id == 1
     assert run_info.start is not None
     assert run_info.end is not None
+    # We patched out engine to prevent the close from happening
+    # so we need to manually close the session
+    session.close()
+    await hass.async_add_executor_job(instance._shutdown)
 
 
 async def test_canceled_before_startup_finishes(
@@ -152,6 +156,9 @@ async def test_canceled_before_startup_finishes(
         "Recorder startup was externally canceled before it could complete"
         in caplog.text
     )
+    # We patched out engine to prevent the close from happening
+    # so we need to manually close the session
+    await hass.async_add_executor_job(instance._shutdown)
 
 
 async def test_shutdown_closes_connections(recorder_mock, hass):

tests/components/recorder/test_migrate.py

@@ -440,6 +440,7 @@ def test_forgiving_add_column(recorder_db_url: str) -> None:
         migration._add_columns(
             instance.get_session, "hello", ["context_id CHARACTER(36)"]
         )
+    engine.dispose()
 
 
 def test_forgiving_add_index(recorder_db_url: str) -> None:
@@ -450,6 +451,7 @@ def test_forgiving_add_index(recorder_db_url: str) -> None:
         instance = Mock()
         instance.get_session = Mock(return_value=session)
         migration._create_index(instance.get_session, "states", "ix_states_context_id")
+    engine.dispose()
 
 
 @pytest.mark.parametrize(

tests/conftest.py

@@ -1077,17 +1077,17 @@ def _ha_orm_quote(mixed, ident):
         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
+        # Check for 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
+            rows = connection.execute(query, parameters={"db": db}).fetchall()
+            if rows:
+                raise RuntimeError(
+                    f"Unable to drop database {db} because it is in use by {rows}"
                 )
-                connection.execute(sa.text("KILL :id"), parameters={"id": row.id})
         engine.dispose()
         sqlalchemy_utils.drop_database(db_url)
     elif db_url.startswith("postgresql://"):