Ty
38eebbc8bb
Live host had stale M2 slabs table (no deleted_at/server_rev) -> GET /slabs 500. Drift check now verifies sentinel (table,column) pairs across settings/slabs/ slab_photos/sync_log; on drift drops all app-owned M3 tables and rebuilds, stamps schema_version. Drift test now seeds M2 settings + slabs and asserts both rebuilt. pytest 23 passed; gates exit 0.
63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
import os
|
|
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)
|
|
# M2-era settings: NO `value` column (drift signature).
|
|
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')")
|
|
# M2-era slabs: NO deleted_at / server_rev (drift signature).
|
|
conn.execute("CREATE TABLE slabs (id TEXT PRIMARY KEY, sku TEXT, status TEXT)")
|
|
conn.execute("INSERT INTO slabs (id, sku, status) VALUES ('old', 'OLD-1', 'calibrated')")
|
|
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, 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
|
|
# slabs must also be rebuilt to the M3 shape (deleted_at present)
|
|
slab_cols = {row[1] for row in conn.execute("PRAGMA table_info(slabs)").fetchall()}
|
|
assert "deleted_at" in slab_cols and "server_rev" in slab_cols
|
|
# M2 rows are disposable (M3 supersedes M2 on this dev volume)
|
|
assert conn.execute("SELECT key FROM settings").fetchall() == []
|
|
assert conn.execute("SELECT id FROM slabs").fetchall() == []
|