fix: read-only corruption preflight, early-return safety path, errorcode classification, sidecar snapshot test

- _preflight_existing_file probes an EXISTING file with a mode=ro URI
  connection before the read-write open, so a corrupt file is detected
  BEFORE WAL mode persists or sidecars appear; the preflight is inside
  the same ConfigEntryNotReady classification as open()/migrations
- corruption classification now prefers sqlite_errorcode (SQLITE_CORRUPT/
  SQLITE_NOTADB primary-byte match, locale-independent, Python 3.11+)
  with message fallback including 'database disk image is malformed'
- the early-return path (existing runtime record) routes through
  _async_open_database and reopens when db is None OR not connected
- new migration-stage corruption test: valid header + damaged page
  4096+, asserts main file byte-identical AND no -wal/-shm sidecars
  left behind; vacuous test replaced with pytest.raises form
This commit is contained in:
Joshua King
2026-09-14 08:37:07 -04:00
parent c5cebc9f49
commit 0281b51676
3 changed files with 122 additions and 28 deletions
+39 -9
View File
@@ -166,19 +166,49 @@ def test_corrupt_file_raises_not_ready_and_leaves_file_untouched(
), "the error log must name the corrupt path"
def test_corrupt_file_does_not_raise_database_error_directly(tmp_path) -> None:
def test_corrupt_file_raises_not_ready_never_raw_error(tmp_path) -> None:
"""The corrupt case must surface as ConfigEntryNotReady, not crash
setup with a raw sqlite3.DatabaseError."""
setup with a raw sqlite3.DatabaseError — and it must ALWAYS raise
(returning normally would mean corruption was silently accepted)."""
db_path = tmp_path / "nestquest.db"
_truncated_database(db_path)
try:
with pytest.raises(ConfigEntryNotReady) as excinfo:
_run(_async_open_database(_make_hass_mock(), db_path))
except ConfigEntryNotReady:
pass
except sqlite3.DatabaseError as err:
raise AssertionError(
f"corruption leaked as raw {type(err).__name__}: {err}"
) from err
assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
def test_migration_stage_corruption_leaves_db_and_sidecars_untouched(
tmp_path,
) -> None:
"""A valid header with a corrupt LATER page is detected during
migrations, AFTER the read-write open has switched the file to WAL.
The leave-untouched guarantee must still hold byte-for-byte for
the main file AND no WAL/SHM sidecars may be left behind for the
owner to clean up.
"""
db_path = tmp_path / "nestquest.db"
_valid_database(db_path)
data = bytearray(db_path.read_bytes())
# Corrupt a page well past the 100-byte header (offset 4096, the
# second page region): the header itself stays valid, so the
# read-write open succeeds and only migrations discover damage.
assert len(data) > 4200
for offset in range(4096, 4128):
data[offset] = 0xFF
db_path.write_bytes(bytes(data))
corrupt_bytes = db_path.read_bytes()
digest = hashlib.sha256(corrupt_bytes).hexdigest()
with pytest.raises(ConfigEntryNotReady) as excinfo:
_run(_async_open_database(_make_hass_mock(), db_path))
assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
# The main file is byte-identical to the corrupt state we handed in.
assert hashlib.sha256(db_path.read_bytes()).hexdigest() == digest
# No WAL/SHM sidecars left behind (WAL mode was never persisted,
# or they were cleaned up on close).
assert not (tmp_path / "nestquest.db-wal").exists()
assert not (tmp_path / "nestquest.db-shm").exists()
def test_open_failure_closes_connection_when_opened(
+16 -8
View File
@@ -130,8 +130,11 @@ def test_no_sqlite3_import_outside_db_module() -> None:
def test_init_module_never_opens_or_executes_sqlite() -> None:
"""The __init__.py sqlite3 exemption is exception-typing only: no
sqlite3.connect, no .execute, no cursor usage may appear there.
"""The __init__.py sqlite3 usage is limited to (a) exception typing
for corruption classification and (b) the ONE read-only preflight
connection (mode=ro URI, SELECT-only) that detects corruption in an
existing file before the real read-write open. No read-write
connect, no cursor writes, no non-SELECT statements may appear.
"""
import re
from pathlib import Path
@@ -139,14 +142,19 @@ def test_init_module_never_opens_or_executes_sqlite() -> None:
init_path = (
Path(store.__file__).parent / "__init__.py"
).read_text(encoding="utf-8")
usage_pattern = re.compile(
r"sqlite3\s*\.\s*(connect|Cursor|Row|register|complete_statement)"
r"|\bconn(ection)?\s*\.\s*execute\b",
# The preflight connect must be the URI read-only form, and only
# SELECT statements may run through it.
assert "sqlite3.connect(uri, uri=True)" in init_path
assert "?mode=ro" in init_path
forbidden = re.compile(
r"sqlite3\s*\.\s*connect(?!\(\s*uri\s*,)"
r"|\bconn(ection)?\s*\.\s*execute\s*\(\s*[\"'](?!SELECT)"
r"|\bexecutemany\b|\bexecutescript\b",
re.IGNORECASE,
)
assert usage_pattern.search(init_path) is None, (
"__init__.py must use sqlite3 exception types only, never open "
"connections or execute SQL"
assert forbidden.search(init_path) is None, (
"__init__.py may use sqlite3 only for exception typing and the "
"read-only corruption preflight"
)