refactor(update): fold the auto-restore sequence into a shared helper
Addresses review feedback on the regression test. The test previously parsed the update_cmd.py AST to assert that each auto-restore call site cleared the destination's sidecars before copying. That bound the fix to source text rather than behaviour, and would break on unrelated refactors. Extract _restore_state_db_from_snapshot(state_path, snap_state), which performs the clear -> copy -> verify sequence as one unit and returns whether the restored file passes its integrity check. Both auto-restore paths now call it, so the ordering is guaranteed by construction instead of by inspection, and the two byte-identical blocks collapse to a single call each. The regression test now exercises that helper directly against a database that still owns a hot WAL: removing the clear from inside the helper fails it with 201 rows where 400 were expected, so the guard remains bound to behaviour. Also covers the two failure modes the callers already handle: a snapshot that does not survive the copy returns False, and a missing snapshot raises OSError.
This commit is contained in:
briandevans
+27
-20
@@ -1516,6 +1516,27 @@ def _write_gateway_update_exit_code(ok: bool) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def _restore_state_db_from_snapshot(state_path: Path, snap_state: Path) -> bool:
|
||||
"""Replace *state_path* with the snapshot image at *snap_state*.
|
||||
|
||||
Shared by both post-update auto-restore paths (the ZIP update and the git
|
||||
pull). The destination's stale sidecars are cleared before the copy, so the
|
||||
restored image cannot be silently overwritten by the corrupt database's WAL
|
||||
replay — see :func:`_clear_stale_sqlite_sidecars`.
|
||||
|
||||
Returns ``True`` when the restored file passes an integrity check. Raises
|
||||
``OSError`` if the copy itself fails, which callers already report.
|
||||
"""
|
||||
from hermes_cli.backup import verify_sqlite_integrity
|
||||
|
||||
_clear_stale_sqlite_sidecars(state_path)
|
||||
shutil.copy2(snap_state, state_path)
|
||||
restored = verify_sqlite_integrity(
|
||||
state_path, check_header=True, run_pragma=True
|
||||
)
|
||||
return bool(restored.get("valid"))
|
||||
|
||||
|
||||
def _update_via_zip(args, *, had_desktop_app_before_update: bool = False) -> bool:
|
||||
"""Update Hermes Agent by downloading a ZIP archive.
|
||||
|
||||
@@ -1897,16 +1918,9 @@ def _update_via_zip(args, *, had_desktop_app_before_update: bool = False) -> boo
|
||||
)
|
||||
if _snap_ok.get("valid"):
|
||||
try:
|
||||
import shutil as _shutil
|
||||
|
||||
_clear_stale_sqlite_sidecars(_state_path)
|
||||
_shutil.copy2(_snap_state, _state_path)
|
||||
_restored_ok = verify_sqlite_integrity(
|
||||
_state_path,
|
||||
check_header=True,
|
||||
run_pragma=True,
|
||||
)
|
||||
if _restored_ok.get("valid"):
|
||||
if _restore_state_db_from_snapshot(
|
||||
_state_path, _snap_state
|
||||
):
|
||||
print(
|
||||
" ✓ Auto-restored from snapshot "
|
||||
f"{_snap_dir.name}"
|
||||
@@ -7104,16 +7118,9 @@ def _cmd_update_impl(args, gateway_mode: bool):
|
||||
)
|
||||
if _snap_ok.get("valid"):
|
||||
try:
|
||||
import shutil as _shutil
|
||||
|
||||
_clear_stale_sqlite_sidecars(_state_path)
|
||||
_shutil.copy2(_snap_state, _state_path)
|
||||
_restored_ok = verify_sqlite_integrity(
|
||||
_state_path,
|
||||
check_header=True,
|
||||
run_pragma=True,
|
||||
)
|
||||
if _restored_ok.get("valid"):
|
||||
if _restore_state_db_from_snapshot(
|
||||
_state_path, _snap_state
|
||||
):
|
||||
print(
|
||||
" ✓ Auto-restored from pre-update "
|
||||
f"snapshot ({_pre_snap_id})"
|
||||
|
||||
@@ -18,14 +18,16 @@ These tests exercise REAL SQLite files, in WAL mode, with a genuinely hot
|
||||
sidecar.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import shutil
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.update_cmd import _clear_stale_sqlite_sidecars
|
||||
from hermes_cli.update_cmd import (
|
||||
_clear_stale_sqlite_sidecars,
|
||||
_restore_state_db_from_snapshot,
|
||||
)
|
||||
|
||||
OLD_ROWS = 201
|
||||
SNAPSHOT_ROWS = 400
|
||||
@@ -174,61 +176,38 @@ def test_clear_removes_every_sidecar_suffix_and_spares_the_database(tmp_path):
|
||||
assert db_path.read_bytes() == b"main-db"
|
||||
|
||||
|
||||
def test_both_auto_restore_call_sites_clear_sidecars_first():
|
||||
"""Bind the fix to the production call sites, not just the helper.
|
||||
def test_restore_helper_serves_snapshot_rows_over_a_hot_wal(
|
||||
live_db_with_hot_wal, snapshot_db
|
||||
):
|
||||
"""The shared restore helper is what both update paths call.
|
||||
|
||||
Both auto-restore blocks (the ZIP-update path and the git-pull path) copy
|
||||
the snapshot with ``_shutil.copy2(_snap_state, _state_path)``. Each one must
|
||||
be immediately preceded by the sidecar clear, or the helper is dead code.
|
||||
It must clear, copy and verify as one unit: after it returns, the database
|
||||
has to hold the SNAPSHOT's rows even though the destination still owned a
|
||||
hot WAL from the corrupt database.
|
||||
"""
|
||||
source = Path(__file__).resolve().parents[2] / "hermes_cli" / "update_cmd.py"
|
||||
tree = ast.parse(source.read_text(encoding="utf-8"))
|
||||
assert _restore_state_db_from_snapshot(live_db_with_hot_wal, snapshot_db) is True
|
||||
|
||||
guarded = 0
|
||||
for node in ast.walk(tree):
|
||||
body = getattr(node, "body", None)
|
||||
if not isinstance(body, list):
|
||||
continue
|
||||
for previous, statement in zip(body, body[1:]):
|
||||
if not _is_snapshot_copy(statement):
|
||||
continue
|
||||
assert _is_sidecar_clear(previous), (
|
||||
"auto-restore at line "
|
||||
f"{statement.lineno} copies the snapshot without clearing the "
|
||||
"destination's stale SQLite sidecars first"
|
||||
)
|
||||
guarded += 1
|
||||
|
||||
assert guarded == 2, f"expected 2 auto-restore call sites, found {guarded}"
|
||||
assert _row_count(live_db_with_hot_wal) == SNAPSHOT_ROWS
|
||||
for suffix in ("-wal", "-shm", "-journal"):
|
||||
assert not _sidecar(live_db_with_hot_wal, suffix).exists()
|
||||
|
||||
|
||||
def _is_snapshot_copy(statement) -> bool:
|
||||
call = _expression_call(statement)
|
||||
if call is None:
|
||||
return False
|
||||
func = call.func
|
||||
return (
|
||||
isinstance(func, ast.Attribute)
|
||||
and func.attr == "copy2"
|
||||
and isinstance(func.value, ast.Name)
|
||||
and func.value.id == "_shutil"
|
||||
and bool(call.args)
|
||||
and isinstance(call.args[0], ast.Name)
|
||||
and call.args[0].id == "_snap_state"
|
||||
)
|
||||
def test_restore_helper_reports_failure_when_the_restored_copy_is_corrupt(tmp_path):
|
||||
"""A snapshot that does not survive the copy must return False, so the
|
||||
caller prints the failure branch instead of claiming success."""
|
||||
state_path = tmp_path / "state.db"
|
||||
state_path.write_bytes(b"whatever")
|
||||
bad_snapshot = tmp_path / "bad-snapshot.db"
|
||||
bad_snapshot.write_bytes(b"\x00" * 4096)
|
||||
|
||||
assert _restore_state_db_from_snapshot(state_path, bad_snapshot) is False
|
||||
|
||||
|
||||
def _is_sidecar_clear(statement) -> bool:
|
||||
call = _expression_call(statement)
|
||||
if call is None:
|
||||
return False
|
||||
return (
|
||||
isinstance(call.func, ast.Name)
|
||||
and call.func.id == "_clear_stale_sqlite_sidecars"
|
||||
)
|
||||
def test_restore_helper_propagates_copy_errors(tmp_path):
|
||||
"""A missing snapshot raises OSError, which both call sites already catch
|
||||
and report as 'Auto-restore file copy failed'."""
|
||||
state_path = tmp_path / "state.db"
|
||||
state_path.write_bytes(b"whatever")
|
||||
|
||||
|
||||
def _expression_call(statement):
|
||||
if isinstance(statement, ast.Expr) and isinstance(statement.value, ast.Call):
|
||||
return statement.value
|
||||
return None
|
||||
with pytest.raises(OSError):
|
||||
_restore_state_db_from_snapshot(state_path, tmp_path / "does-not-exist.db")
|
||||
|
||||
Reference in New Issue
Block a user