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

homeassistant/components/recorder/core.py

@@ -1217,13 +1217,13 @@ def _end_session(self) -> None:
         """End the recorder session."""
         if self.event_session is None:
             return
-        try:
+        if self.run_history.active:
             self.run_history.end(self.event_session)
+        try:
             self._commit_event_session_or_retry()
             self.event_session.close()
         except Exception as err:  # pylint: disable=broad-except
             _LOGGER.exception("Error saving the event session during shutdown: %s", err)
-
         self.run_history.clear()
 
     def _shutdown(self) -> None:

homeassistant/components/recorder/run_history.py

@@ -72,6 +72,11 @@ def current(self) -> RecorderRuns:
             start=self.recording_start, created=dt_util.utcnow()
         )
 
+    @property
+    def active(self) -> bool:
+        """Return if a run is active."""
+        return self._current_run_info is not None
+
     def get(self, start: datetime) -> RecorderRuns | None:
         """Return the recorder run that started before or at start.
 
@@ -142,6 +147,5 @@ def clear(self) -> None:
 
         Must run in the recorder thread.
         """
-        assert self._current_run_info is not None
-        assert self._current_run_info.end is not None
-        self._current_run_info = None
+        if self._current_run_info:
+            self._current_run_info = None

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):
@@ -1624,7 +1631,6 @@ async def test_database_lock_timeout(recorder_mock, hass, recorder_db_url):
         return
 
     hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
-
     instance = get_instance(hass)
 
     class BlockQueue(recorder.tasks.RecorderTask):

tests/components/recorder/test_migrate.py

@@ -436,6 +436,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:
@@ -446,6 +447,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/components/recorder/test_run_history.py

@@ -5,8 +5,13 @@
 from homeassistant.components import recorder
 from homeassistant.components.recorder.db_schema import RecorderRuns
 from homeassistant.components.recorder.models import process_timestamp
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import recorder as recorder_helper
+from homeassistant.setup import async_setup_component
 from homeassistant.util import dt as dt_util
 
+from .common import async_wait_recording_done
+
 
 async def test_run_history(recorder_mock, hass):
     """Test the run history gives the correct run."""
@@ -47,12 +52,21 @@ async def test_run_history(recorder_mock, hass):
     )
 
 
-async def test_run_history_during_schema_migration(recorder_mock, hass):
-    """Test the run history during schema migration."""
+async def test_run_history_while_recorder_is_not_yet_started(
+    recorder_db_url: str, hass: HomeAssistant
+) -> None:
+    """Test the run history while recorder is not yet started.
+
+    This usually happens during schema migration because
+    we do not start right away.
+    """
+    recorder_helper.async_initialize_recorder(hass)
+    await async_setup_component(
+        hass, "recorder", {"recorder": {"db_url": recorder_db_url}}
+    )
     instance = recorder.get_instance(hass)
     run_history = instance.run_history
     assert run_history.current.start == run_history.recording_start
-    with instance.get_session() as session:
-        run_history.start(session)
+    await async_wait_recording_done(hass)
     assert run_history.current.start == run_history.recording_start
     assert run_history.current.created >= run_history.recording_start

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