Harden location db startup validation
This commit is contained in:
+44
-8
@@ -25,11 +25,8 @@ def _make_alembic_config(database_url: str) -> Config:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def location_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||||
database_path = tmp_path / "location_test.db"
|
||||
database_url = f"sqlite:///{database_path}"
|
||||
|
||||
command.upgrade(_make_alembic_config(database_url), "head")
|
||||
def location_client(ready_location_database, monkeypatch: pytest.MonkeyPatch):
|
||||
database_url = ready_location_database["location_url"]
|
||||
|
||||
engine = create_engine(database_url, connect_args={"check_same_thread": False})
|
||||
session_local = sessionmaker(bind=engine, autoflush=False, autocommit=False)
|
||||
@@ -122,9 +119,11 @@ def test_location_record_endpoint_keeps_legacy_lenient_number_parsing(location_c
|
||||
|
||||
|
||||
def test_legacy_style_location_db_can_be_stamped_and_adopted(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
test_database_urls, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
database_path = tmp_path / "legacy_location.db"
|
||||
database_path = test_database_urls["location_path"]
|
||||
database_url = test_database_urls["location_url"]
|
||||
|
||||
conn = sqlite3.connect(database_path)
|
||||
conn.execute(
|
||||
"""
|
||||
@@ -142,7 +141,6 @@ def test_legacy_style_location_db_can_be_stamped_and_adopted(
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
database_url = f"sqlite:///{database_path}"
|
||||
command.stamp(_make_alembic_config(database_url), LOCATION_BASELINE_REVISION)
|
||||
|
||||
engine = create_engine(database_url, connect_args={"check_same_thread": False})
|
||||
@@ -228,6 +226,44 @@ def test_location_db_adoption_validates_and_stamps_legacy_db(tmp_path: Path) ->
|
||||
assert revision == LOCATION_BASELINE_REVISION
|
||||
|
||||
|
||||
def test_location_db_adoption_accepts_already_managed_matching_revision(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
database_path = tmp_path / "managed_location.db"
|
||||
command.upgrade(_make_alembic_config(f"sqlite:///{database_path}"), "head")
|
||||
|
||||
result = adopt_or_initialize_location_db(f"sqlite:///{database_path}")
|
||||
|
||||
assert result == "already_managed"
|
||||
|
||||
|
||||
def test_location_db_adoption_fails_closed_on_alembic_revision_mismatch(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
database_path = tmp_path / "wrong_revision.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("CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)")
|
||||
conn.execute("INSERT INTO alembic_version (version_num) VALUES ('wrong_revision')")
|
||||
conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
with pytest.raises(LocationDatabaseAdoptionError, match="revision does not match"):
|
||||
adopt_or_initialize_location_db(f"sqlite:///{database_path}")
|
||||
|
||||
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user