28 lines
1.1 KiB
Python
28 lines
1.1 KiB
Python
import sqlite3
|
|
from sfm.database import SeismoDb
|
|
|
|
_SHAPE_COLS = {"shape_crest_factor", "shape_near_peak_count",
|
|
"shape_sample_count", "shape_axis"}
|
|
|
|
def _cols(db):
|
|
with sqlite3.connect(db.db_path) as c:
|
|
return {r[1] for r in c.execute("PRAGMA table_info(events)")}
|
|
|
|
def test_fresh_db_has_shape_columns(tmp_path):
|
|
db = SeismoDb(tmp_path / "s.db")
|
|
assert _SHAPE_COLS <= _cols(db)
|
|
|
|
def test_existing_db_migrates_shape_columns(tmp_path):
|
|
p = tmp_path / "s.db"
|
|
db = SeismoDb(p)
|
|
with sqlite3.connect(p) as c: # simulate an older DB missing the columns
|
|
for col in _SHAPE_COLS:
|
|
c.execute(f"ALTER TABLE events DROP COLUMN {col}")
|
|
# sanity: dropped. Read via the already-constructed `db` (a raw PRAGMA
|
|
# read against db.db_path) rather than a fresh SeismoDb(p) — the latter
|
|
# would immediately re-trigger _migrate's self-healing ADD COLUMN loop
|
|
# and mask the gap we're trying to confirm.
|
|
assert not (_SHAPE_COLS <= _cols(db))
|
|
SeismoDb(p) # re-open triggers _migrate
|
|
assert _SHAPE_COLS <= _cols(SeismoDb(p))
|