Add location db adoption runbook
This commit is contained in:
+103
-2
@@ -10,8 +10,12 @@ from sqlalchemy.orm import sessionmaker
|
||||
|
||||
import app.db
|
||||
from app.main import create_app
|
||||
|
||||
LOCATION_BASELINE_REVISION = "20260419_01_location_baseline"
|
||||
from scripts.location_db_adopt import (
|
||||
EXPECTED_USER_VERSION,
|
||||
LOCATION_BASELINE_REVISION,
|
||||
LocationDatabaseAdoptionError,
|
||||
adopt_or_initialize_location_db,
|
||||
)
|
||||
|
||||
|
||||
def _make_alembic_config(database_url: str) -> Config:
|
||||
@@ -170,3 +174,100 @@ def test_legacy_style_location_db_can_be_stamped_and_adopted(
|
||||
assert row_count == 1
|
||||
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def test_location_db_adoption_initializes_new_db(tmp_path: Path) -> None:
|
||||
database_path = tmp_path / "new_location.db"
|
||||
result = adopt_or_initialize_location_db(f"sqlite:///{database_path}")
|
||||
|
||||
assert result == "initialized"
|
||||
assert database_path.exists()
|
||||
|
||||
conn = sqlite3.connect(database_path)
|
||||
try:
|
||||
revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0]
|
||||
location_table = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'location'"
|
||||
).fetchone()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
assert revision == LOCATION_BASELINE_REVISION
|
||||
assert location_table is not None
|
||||
|
||||
|
||||
def test_location_db_adoption_validates_and_stamps_legacy_db(tmp_path: Path) -> None:
|
||||
database_path = tmp_path / "legacy_location.db"
|
||||
conn = sqlite3.connect(database_path)
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE location (
|
||||
person TEXT NOT NULL,
|
||||
datetime TEXT NOT NULL,
|
||||
latitude REAL NOT NULL,
|
||||
longitude REAL NOT NULL,
|
||||
altitude REAL,
|
||||
PRIMARY KEY (person, datetime)
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
result = adopt_or_initialize_location_db(f"sqlite:///{database_path}")
|
||||
|
||||
assert result == "adopted"
|
||||
|
||||
conn = sqlite3.connect(database_path)
|
||||
try:
|
||||
revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
assert revision == LOCATION_BASELINE_REVISION
|
||||
|
||||
|
||||
def test_location_db_adoption_fails_closed_on_schema_mismatch(tmp_path: Path) -> None:
|
||||
database_path = tmp_path / "bad_schema.db"
|
||||
conn = sqlite3.connect(database_path)
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE location (
|
||||
person TEXT NOT NULL,
|
||||
datetime TEXT NOT NULL,
|
||||
latitude REAL NOT NULL,
|
||||
longitude REAL NOT NULL,
|
||||
PRIMARY KEY (person, datetime)
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
with pytest.raises(LocationDatabaseAdoptionError, match="schema does not match"):
|
||||
adopt_or_initialize_location_db(f"sqlite:///{database_path}")
|
||||
|
||||
|
||||
def test_location_db_adoption_fails_closed_on_user_version_mismatch(tmp_path: Path) -> None:
|
||||
database_path = tmp_path / "bad_user_version.db"
|
||||
conn = sqlite3.connect(database_path)
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE location (
|
||||
person TEXT NOT NULL,
|
||||
datetime TEXT NOT NULL,
|
||||
latitude REAL NOT NULL,
|
||||
longitude REAL NOT NULL,
|
||||
altitude REAL,
|
||||
PRIMARY KEY (person, datetime)
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute("PRAGMA user_version = 999")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
with pytest.raises(LocationDatabaseAdoptionError, match="Expected PRAGMA user_version"):
|
||||
adopt_or_initialize_location_db(f"sqlite:///{database_path}")
|
||||
|
||||
Reference in New Issue
Block a user