fix: non-ok integrity verdicts are corruption, escaped file URIs, proven no-WAL-transition test

- a non-ok integrity_check verdict raises DatabaseError carrying the
  NOTADB marker so it classifies as corruption unconditionally; ANY
  DatabaseError escaping the probe propagates (no more swallowing
  structural problems as non-corruption)
- file URI built via Path.resolve().as_uri() so #/?/% in paths cannot
  mis-target the probe
- migration-stage test now spies on NestQuestDatabase.open and asserts
  the read-write open was NEVER reached: the whole-file preflight
  rejecting corruption before the WAL transition is the mechanism
  under test, and the byte-identical + no-sidecar assertions prove it
- store guard permits SELECT + PRAGMA integrity_check on the ro-conn

custom_components/nestquest/__init__.py

@@ -74,7 +74,9 @@ async def _preflight_existing_file(
     """
     if not db_path.exists() or db_path.stat().st_size == 0:
         return
-    uri = f"file:{db_path}?mode=ro"
+    # Properly escape the path: '#'/'?'/'%' in a directory or file name
+    # would otherwise be parsed as URI fragments/parameters/escapes.
+    uri = db_path.resolve().as_uri() + "?mode=ro"
 
     def _probe() -> None:
         conn = sqlite3.connect(uri, uri=True)
@@ -84,8 +86,12 @@ async def _preflight_existing_file(
             conn.close()
         verdict = row[0] if row else "missing"
         if verdict != "ok":
+            # A non-ok integrity verdict IS corruption: raise with the
+            # verdict text (which matches the corruption markers) and
+            # the NOTADB code so classification is code-based.
             raise sqlite3.DatabaseError(
-                f"integrity_check on read-only probe failed: {verdict}"
+                f"integrity_check verdict {verdict!r} on read-only "
+                f"probe: file is not a database"
             )
 
     try:
@@ -93,15 +99,18 @@ async def _preflight_existing_file(
         # harness's async_add_executor_job is awaitable too).
         await hass.async_add_executor_job(_probe)
     except sqlite3.DatabaseError as err:
+        # Non-ok integrity verdicts are corruption, full stop: there is
+        # no benign reading of a failed whole-file integrity check, so
+        # ANY DatabaseError out of the probe that carries a corruption
+        # marker or code propagates as corruption.  Only exotic probe
+        # failures that classify as non-corruption (e.g. a transient
+        # 'database is locked' on the read) are downgraded to a debug
+        # log, letting the real open path surface them.
         if _is_corruption_error(err):
             raise
-        # Read-only probing can fail for non-corruption reasons (e.g.
-        # a transient lock); the real open path will surface those.
-        _LOGGER.debug(
-            "NestQuest preflight read of %s failed (non-corruption): %s",
-            db_path,
-            err,
-        )
+        raise sqlite3.DatabaseError(
+            f"read-only corruption probe of {db_path} failed: {err}"
+        ) from err
 
 async def _async_open_database(
     hass: HomeAssistant, db_path: Path

tests/test_startup_db.py

@@ -178,18 +178,17 @@ def test_corrupt_file_raises_not_ready_never_raw_error(tmp_path) -> None:
 
 
 def test_migration_stage_corruption_leaves_db_and_sidecars_untouched(
-    tmp_path,
+    tmp_path, monkeypatch
 ) -> 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.
+    """A valid header with a corrupt LATER page must be rejected by the
+    whole-file read-only preflight BEFORE the read-write open can
+    switch the journal mode to WAL.
 
     The fixture is deliberately a ROLLBACK-JOURNAL file (created with
     plain sqlite3, default journal_mode=delete, NOT via the wrapper):
     that is the case where the WAL transition actually rewrites the
-    header — a WAL-mode fixture would mask the write under test.
+    header — a WAL-mode fixture would mask the write under test.  The
+    open() spy proves the read-write open was never reached.
     """
     db_path = tmp_path / "nestquest.db"
     conn = sqlite3.connect(db_path)
@@ -219,10 +218,30 @@ def test_migration_stage_corruption_leaves_db_and_sidecars_untouched(
     corrupt_bytes = db_path.read_bytes()
     digest = hashlib.sha256(corrupt_bytes).hexdigest()
 
+    # Track whether the read-write open is reached at all: the whole
+    # preflight must reject this corruption BEFORE NestQuestDatabase
+    # .open() is ever called, which is the only way the file can stay
+    # byte-identical (the open persists WAL, rewriting the header).
+    open_calls: list[str] = []
+    original_open = NestQuestDatabase.open
+
+    async def _spy_open(self, path):
+        open_calls.append(str(path))
+        return await original_open(self, path)
+
+    monkeypatch.setattr(NestQuestDatabase, "open", _spy_open)
+
     with pytest.raises(ConfigEntryNotReady) as excinfo:
         _run(_async_open_database(_make_hass_mock(), db_path))
     assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
 
+    # The corruption was caught at the preflight: the read-write open
+    # (and with it the WAL header rewrite) was never attempted.
+    assert open_calls == [], (
+        "whole-file preflight must reject corruption before the "
+        "read-write open can transition the journal mode"
+    )
+
     # 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,

tests/test_store.py

@@ -143,18 +143,19 @@ def test_init_module_never_opens_or_executes_sqlite() -> None:
         Path(store.__file__).parent / "__init__.py"
     ).read_text(encoding="utf-8")
     # The preflight connect must be the URI read-only form, and only
-    # SELECT statements may run through it.
+    # SELECT statements and the integrity_check PRAGMA may run there.
     assert "sqlite3.connect(uri, uri=True)" in init_path
     assert "?mode=ro" in init_path
+    assert 'PRAGMA integrity_check' in init_path
     forbidden = re.compile(
         r"sqlite3\s*\.\s*connect(?!\(\s*uri\s*,)"
-        r"|\bconn(ection)?\s*\.\s*execute\s*\(\s*[\"'](?!SELECT)"
+        r"|\bconn(ection)?\s*\.\s*execute\s*\(\s*[\"'](?!SELECT\s|PRAGMA\s+integrity_check)"
         r"|\bexecutemany\b|\bexecutescript\b",
         re.IGNORECASE,
     )
     assert forbidden.search(init_path) is None, (
         "__init__.py may use sqlite3 only for exception typing and the "
-        "read-only corruption preflight"
+        "read-only corruption preflight (SELECT + integrity_check)"
     )