Fix logic for purge of recorder runs (#130378)

* Fix logic for purge of recorder runs

* Make test more explicit

* Explicitly don't remove unclosed recorder runs in purge

homeassistant/components/recorder/queries.py

@@ -608,7 +608,8 @@ def delete_recorder_runs_rows(
     """Delete recorder_runs rows."""
     return lambda_stmt(
         lambda: delete(RecorderRuns)
-        .filter(RecorderRuns.start < purge_before)
+        .filter(RecorderRuns.end.is_not(None))
+        .filter(RecorderRuns.end < purge_before)
         .filter(RecorderRuns.run_id != current_run_id)
         .execution_options(synchronize_session=False)
     )

tests/components/recorder/test_purge.py

@@ -352,6 +352,8 @@ async def test_purge_old_recorder_runs(
     with session_scope(hass=hass) as session:
         recorder_runs = session.query(RecorderRuns)
         assert recorder_runs.count() == 7
+        # Make sure we have a run that is not closed
+        assert sum(run.end is None for run in recorder_runs) == 1
 
     purge_before = dt_util.utcnow()
 
@@ -376,7 +378,9 @@ async def test_purge_old_recorder_runs(
 
     with session_scope(hass=hass) as session:
         recorder_runs = session.query(RecorderRuns)
-        assert recorder_runs.count() == 1
+        assert recorder_runs.count() == 3
+        # Make sure we did not purge the unclosed run
+        assert sum(run.end is None for run in recorder_runs) == 1
 
 
 async def test_purge_old_statistics_runs(

tests/components/recorder/test_purge_v32_schema.py

@@ -347,7 +347,7 @@ async def test_purge_old_recorder_runs(
 
     with session_scope(hass=hass) as session:
         recorder_runs = session.query(RecorderRuns)
-        assert recorder_runs.count() == 1
+        assert recorder_runs.count() == 3
 
 
 async def test_purge_old_statistics_runs(