M1-T04: converge startup chain onto the single app DB

run_all_migrations() now adopts/initializes only the app DB and returns
{'app': ...}. app/main.py drops the location/poo readiness checks
(ensure_location_db_ready / ensure_poo_db_ready) and their imports;
ensure_runtime_dirs only provisions the app DB path; lifespan still
fail-closes on a missing/unmanaged app DB. Delete the retired
location/poo adopt scripts and the alembic_location / alembic_poo
chains. Update tests to single-DB expectations and drop the obsolete
location/poo adoption + readiness tests.

pytest 95 passed; ruff clean (pre-existing only); a fresh app DB
initialized via scripts.run_migrations contains location + poo_records.
This commit is contained in:
2026-06-12 16:50:05 +02:00
parent 3d3c2bcc57
commit 0d898e09f2
21 changed files with 12 additions and 1226 deletions
-149
View File
@@ -1,24 +1,8 @@
from datetime import datetime
from pathlib import Path
import sqlite3
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import text
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:
config = Config("alembic_location.ini")
config.set_main_option("sqlalchemy.url", database_url)
return config
def test_location_record_endpoint_writes_row(location_client) -> None:
client, engine = location_client
@@ -193,136 +177,3 @@ def test_location_record_endpoint_defaults_invalid_altitude_to_zero(location_cli
assert row.altitude == pytest.approx(0.0)
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_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="known migration revision"):
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)
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}")