Add ability to disable the sqlite3 quick_check (#39479)

tests/components/recorder/test_util.py

@@ -69,29 +69,77 @@ def test_recorder_bad_execute(hass_recorder):
     assert e_mock.call_count == 2
 
 
-def test_validate_or_move_away_sqlite_database(hass, tmpdir, caplog):
-    """Ensure a malformed sqlite database is moved away."""
+def test_validate_or_move_away_sqlite_database_with_integrity_check(
+    hass, tmpdir, caplog
+):
+    """Ensure a malformed sqlite database is moved away.
+
+    A quick_check is run here
+    """
+
+    db_integrity_check = True
 
     test_dir = tmpdir.mkdir("test_validate_or_move_away_sqlite_database")
     test_db_file = f"{test_dir}/broken.db"
     dburl = f"{SQLITE_URL_PREFIX}{test_db_file}"
 
-    util.validate_sqlite_database(test_db_file) is True
+    util.validate_sqlite_database(test_db_file, db_integrity_check) is True
 
     assert os.path.exists(test_db_file) is True
-    assert util.validate_or_move_away_sqlite_database(dburl) is False
+    assert (
+        util.validate_or_move_away_sqlite_database(dburl, db_integrity_check) is False
+    )
 
     _corrupt_db_file(test_db_file)
 
-    assert util.validate_sqlite_database(dburl) is False
+    assert util.validate_sqlite_database(dburl, db_integrity_check) is False
 
-    assert util.validate_or_move_away_sqlite_database(dburl) is False
+    assert (
+        util.validate_or_move_away_sqlite_database(dburl, db_integrity_check) is False
+    )
 
     assert "corrupt or malformed" in caplog.text
 
-    assert util.validate_sqlite_database(dburl) is False
+    assert util.validate_sqlite_database(dburl, db_integrity_check) is False
 
-    assert util.validate_or_move_away_sqlite_database(dburl) is True
+    assert util.validate_or_move_away_sqlite_database(dburl, db_integrity_check) is True
+
+
+def test_validate_or_move_away_sqlite_database_without_integrity_check(
+    hass, tmpdir, caplog
+):
+    """Ensure a malformed sqlite database is moved away.
+
+    The quick_check is skipped, but we can still find
+    corruption if the whole database is unreadable
+    """
+
+    db_integrity_check = False
+
+    test_dir = tmpdir.mkdir("test_validate_or_move_away_sqlite_database")
+    test_db_file = f"{test_dir}/broken.db"
+    dburl = f"{SQLITE_URL_PREFIX}{test_db_file}"
+
+    util.validate_sqlite_database(test_db_file, db_integrity_check) is True
+
+    assert os.path.exists(test_db_file) is True
+    assert (
+        util.validate_or_move_away_sqlite_database(dburl, db_integrity_check) is False
+    )
+
+    _corrupt_db_file(test_db_file)
+
+    assert util.validate_sqlite_database(dburl, db_integrity_check) is False
+
+    assert (
+        util.validate_or_move_away_sqlite_database(dburl, db_integrity_check) is False
+    )
+
+    assert "corrupt or malformed" in caplog.text
+
+    assert util.validate_sqlite_database(dburl, db_integrity_check) is False
+
+    assert util.validate_or_move_away_sqlite_database(dburl, db_integrity_check) is True
 
 
 def test_last_run_was_recently_clean(hass_recorder):
@@ -134,25 +182,32 @@ def test_combined_checks(hass_recorder):
     """Run Checks on the open database."""
     hass = hass_recorder()
 
+    db_integrity_check = False
+
     cursor = hass.data[DATA_INSTANCE].engine.raw_connection().cursor()
 
-    assert util.run_checks_on_open_db("fake_db_path", cursor) is None
+    assert (
+        util.run_checks_on_open_db("fake_db_path", cursor, db_integrity_check) 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
+        assert (
+            util.run_checks_on_open_db("fake_db_path", cursor, db_integrity_check)
+            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)
+        util.run_checks_on_open_db("fake_db_path", cursor, db_integrity_check)
 
     cursor.execute("DROP TABLE events;")
 
     with pytest.raises(sqlite3.DatabaseError):
-        util.run_checks_on_open_db("fake_db_path", cursor)
+        util.run_checks_on_open_db("fake_db_path", cursor, db_integrity_check)
 
 
 def _corrupt_db_file(test_db_file):