82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
"""
|
|
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
|
|
|
|
|
|
def test_recent_waveforms_zip_bundles_all_per_event_files(tmp_path):
|
|
import zipfile
|
|
from sfm.database import SeismoDb
|
|
from sfm.waveform_store import WaveformStore
|
|
from sfm.db_snapshot import build_recent_waveforms_zip
|
|
|
|
db = SeismoDb(tmp_path / "seismo_relay.db") # creates schema on init
|
|
store = WaveformStore(tmp_path / "waveforms")
|
|
|
|
def add(serial, fn, ts):
|
|
d = store.root / serial
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
for suffix in ("", ".h5", ".a5.pkl", ".sfm.json"):
|
|
(d / f"{fn}{suffix}").write_bytes(b"data")
|
|
_seed_event(db.db_path, serial, fn, ts)
|
|
|
|
add("BE111", "AAAA1111.OE0H", "2026-06-01T00:00:00")
|
|
add("BE111", "BBBB2222.OE0H", "2026-06-02T00:00:00") # newest
|
|
|
|
dest = tmp_path / "wf.zip"
|
|
stats = build_recent_waveforms_zip(db, store, n=1, dest_path=dest)
|
|
|
|
assert stats["events"] == 1
|
|
assert stats["files"] == 4
|
|
with zipfile.ZipFile(dest) as zf:
|
|
names = set(zf.namelist())
|
|
assert names == {
|
|
"BE111/BBBB2222.OE0H",
|
|
"BE111/BBBB2222.OE0H.h5",
|
|
"BE111/BBBB2222.OE0H.a5.pkl",
|
|
"BE111/BBBB2222.OE0H.sfm.json",
|
|
}
|