Skip the sqlite quick_check on clean restarts (#38972)

tests/components/recorder/test_util.py

@@ -1,10 +1,13 @@
 """Test util methods."""
+from datetime import timedelta
 import os
+import sqlite3
 
 import pytest
 
 from homeassistant.components.recorder import util
 from homeassistant.components.recorder.const import DATA_INSTANCE, SQLITE_URL_PREFIX
+from homeassistant.util import dt as dt_util
 
 from tests.async_mock import MagicMock, patch
 from tests.common import get_test_home_assistant, init_recorder_component
@@ -74,17 +77,81 @@ def test_validate_or_move_away_sqlite_database(hass, tmpdir, caplog):
     util.validate_sqlite_database(test_db_file) is True
 
     assert os.path.exists(test_db_file) is True
-    assert util.validate_or_move_away_sqlite_database(dburl) is True
+    assert util.validate_or_move_away_sqlite_database(dburl) is False
 
     _corrupt_db_file(test_db_file)
 
+    assert util.validate_sqlite_database(dburl) is False
+
     assert util.validate_or_move_away_sqlite_database(dburl) is False
 
     assert "corrupt or malformed" in caplog.text
 
+    assert util.validate_sqlite_database(dburl) is False
+
     assert util.validate_or_move_away_sqlite_database(dburl) is True
 
 
+def test_last_run_was_recently_clean(hass_recorder):
+    """Test we can check if the last recorder run was recently clean."""
+    hass = hass_recorder()
+
+    cursor = hass.data[DATA_INSTANCE].engine.raw_connection().cursor()
+
+    assert util.last_run_was_recently_clean(cursor) is False
+
+    hass.data[DATA_INSTANCE]._close_run()
+
+    assert util.last_run_was_recently_clean(cursor) is True
+
+    thirty_min_future_time = dt_util.utcnow() + timedelta(minutes=30)
+
+    with patch(
+        "homeassistant.components.recorder.dt_util.utcnow",
+        return_value=thirty_min_future_time,
+    ):
+        assert util.last_run_was_recently_clean(cursor) is False
+
+
+def test_basic_sanity_check(hass_recorder):
+    """Test the basic sanity checks with a missing table."""
+    hass = hass_recorder()
+
+    cursor = hass.data[DATA_INSTANCE].engine.raw_connection().cursor()
+
+    assert util.basic_sanity_check(cursor) is True
+
+    cursor.execute("DROP TABLE states;")
+
+    with pytest.raises(sqlite3.DatabaseError):
+        util.basic_sanity_check(cursor)
+
+
+def test_combined_checks(hass_recorder):
+    """Run Checks on the open database."""
+    hass = hass_recorder()
+
+    cursor = hass.data[DATA_INSTANCE].engine.raw_connection().cursor()
+
+    assert util.run_checks_on_open_db("fake_db_path", cursor) is None
+
+    # We are patching recorder.util here in order
+    # to avoid creating the full database on disk
+    with patch("homeassistant.components.recorder.util.last_run_was_recently_clean"):
+        assert util.run_checks_on_open_db("fake_db_path", cursor) is None
+
+    with patch(
+        "homeassistant.components.recorder.util.last_run_was_recently_clean",
+        side_effect=sqlite3.DatabaseError,
+    ), pytest.raises(sqlite3.DatabaseError):
+        util.run_checks_on_open_db("fake_db_path", cursor)
+
+    cursor.execute("DROP TABLE events;")
+
+    with pytest.raises(sqlite3.DatabaseError):
+        util.run_checks_on_open_db("fake_db_path", cursor)
+
+
 def _corrupt_db_file(test_db_file):
     """Corrupt an sqlite3 database file."""
     f = open(test_db_file, "a")