38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
import os
|
|
import tempfile
|
|
import sqlite3
|
|
|
|
import backend.db as db_mod
|
|
|
|
|
|
|
|
def _setup_temp_db_with_m2_schema(tmp_dir: str) -> str:
|
|
db_path = os.path.join(tmp_dir, "test.db")
|
|
conn = sqlite3.connect(db_path)
|
|
# create old settings table
|
|
conn.execute("CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT, value_enc TEXT)")
|
|
conn.commit()
|
|
conn.close()
|
|
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 == []
|