fix(db): schema_version drift detection + rebuild for pre-M3 volumes

- Add schema_version table; init_db detects a pre-M3/unknown DB (no version
  row, or settings table missing the value column) and rebuilds to the clean
  M3 schema, then stamps version 1
- _has_schema_version checks for a real row (empty table != versioned)
- _reset_for_tests checkpoints WAL before close (Windows handle release)
- TDD: test_schema_drift seeds the M2 settings shape, asserts rebuild + version

Fixes live 500 'no such column: value' on the UAT host (stale M2 slab_db volume).
Gates: check_requirements + check_coverage exit 0; pytest 14 passed.
This commit is contained in:
2026-09-19 16:05:40 -07:00
parent 60bc2fb9d0
commit a0032ddd2d
2 changed files with 49 additions and 25 deletions
+40 -22
View File
@@ -1,37 +1,55 @@
import os
import tempfile
import shutil
import sqlite3
import tempfile
import pytest
import backend.db as db_mod
@pytest.fixture
def drift_db_dir():
# Persistent temp dir; on Windows the global WAL connection can outlive the
# test frame, so we avoid TemporaryDirectory's immediate rmtree race and clean
# up best-effort after the handle is released.
tmp = tempfile.mkdtemp(prefix="slab_drift_")
yield tmp
db_mod._reset_for_tests()
shutil.rmtree(tmp, ignore_errors=True)
def _setup_temp_db_with_m2_schema(tmp_dir: str) -> str:
db_path = os.path.join(tmp_dir, "test.db")
# M2-era settings shape: NO `value` column (this is the drift signature).
# Use a raw connection (DELETE journal) exactly as an old M2 db would be on disk.
conn = sqlite3.connect(db_path)
# create old settings table
conn.execute("CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT, value_enc TEXT)")
conn.execute("CREATE TABLE settings (key TEXT PRIMARY KEY, value_enc BLOB, updated_at TEXT)")
conn.execute("INSERT INTO settings (key, value_enc, updated_at) VALUES ('woo_wp_username', X'00', '2026-01-01')")
conn.commit()
conn.close()
# Force any journal sidecars to merge before the WAL-mode factory opens the file.
for suffix in ("-journal", "-wal", "-shm"):
try:
os.remove(db_path + suffix)
except OSError:
pass
return db_path
def test_drift_detection_and_rebuild(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
db_path = _setup_temp_db_with_m2_schema(tmp)
monkeypatch.setenv("SLAB_DB_PATH", db_path)
# init_db should rebuild
conn = db_mod.init_db()
# check schema_version exists
cur = conn.execute("SELECT version FROM schema_version").fetchone()
assert cur is not None
assert cur[0] == 1
# check settings has columns value and value_enc
info = conn.execute("PRAGMA table_info(settings)").fetchall()
cols = {row[1] for row in info}
assert "value" in cols
assert "value_enc" in cols
# also ensure old settings table dropped and recreated
# no other data
rows = conn.execute("SELECT key FROM settings").fetchall()
assert rows == []
def test_drift_detection_and_rebuild(monkeypatch, drift_db_dir):
db_path = _setup_temp_db_with_m2_schema(drift_db_dir)
monkeypatch.setenv("SLAB_DB_PATH", db_path)
# init_db must detect pre-M3 drift and rebuild to the M3 settings shape
conn = db_mod.init_db()
cur = conn.execute("SELECT version FROM schema_version").fetchone()
assert cur is not None
assert cur[0] == 1
info = conn.execute("PRAGMA table_info(settings)").fetchall()
cols = {row[1] for row in info}
assert "value" in cols
assert "value_enc" in cols
# M2 rows are disposable (M3 supersedes M2 on this dev volume)
rows = conn.execute("SELECT key FROM settings").fetchall()
assert rows == []