feat(db): WAL-safe seismo_relay.db snapshot helper
This commit is contained in:
@@ -0,0 +1,47 @@
|
||||
"""
|
||||
test_db_snapshot.py — unit tests for sfm/db_snapshot.py (snapshot/restore
|
||||
helpers behind the /db/snapshot, /db/waveforms/recent.zip, /db/restore routes).
|
||||
|
||||
Run:
|
||||
python -m pytest tests/test_db_snapshot.py -v
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import os
|
||||
import sys
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
|
||||
def _make_db(path: Path, rows: int = 3) -> None:
|
||||
conn = sqlite3.connect(str(path))
|
||||
conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
|
||||
conn.executemany("INSERT INTO t (v) VALUES (?)", [(f"row{i}",) for i in range(rows)])
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def _seed_event(db_path: Path, serial: str, fn: str, ts: str) -> None:
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.execute(
|
||||
"INSERT INTO events (id, serial, waveform_key, timestamp, "
|
||||
"blastware_filename, a5_pickle_filename, sidecar_filename, false_trigger) "
|
||||
"VALUES (?, ?, ?, ?, ?, ?, ?, 0)",
|
||||
(f"id-{fn}", serial, fn[:8], ts, fn, f"{fn}.a5.pkl", f"{fn}.sfm.json"),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_snapshot_db_produces_queryable_copy(tmp_path):
|
||||
from sfm.db_snapshot import snapshot_db
|
||||
src = tmp_path / "src.db"
|
||||
_make_db(src, rows=5)
|
||||
dest = tmp_path / "snap.db"
|
||||
out = snapshot_db(src, dest)
|
||||
assert out == dest and dest.exists()
|
||||
conn = sqlite3.connect(str(dest))
|
||||
n = conn.execute("SELECT COUNT(*) FROM t").fetchone()[0]
|
||||
conn.close()
|
||||
assert n == 5
|
||||
Reference in New Issue
Block a user