Add runtime DB schema migration and version tracking
This commit is contained in:
@@ -67,3 +67,34 @@ async def test_init_db_is_idempotent(tmp_path: Path) -> None:
|
||||
async with aiosqlite.connect(db_path) as db:
|
||||
await init_db(db)
|
||||
await init_db(db) # Second call must be a no-op.
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_init_db_records_schema_version(tmp_path: Path) -> None:
|
||||
"""``init_db`` must record the current schema version."""
|
||||
db_path = str(tmp_path / "test.db")
|
||||
async with aiosqlite.connect(db_path) as db:
|
||||
await init_db(db)
|
||||
async with db.execute(
|
||||
"SELECT version FROM schema_migrations ORDER BY version DESC LIMIT 1;"
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row[0] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_init_db_migrates_legacy_database_without_schema_version(tmp_path: Path) -> None:
|
||||
"""Legacy databases without a schema marker must be migrated on startup."""
|
||||
db_path = str(tmp_path / "test.db")
|
||||
async with aiosqlite.connect(db_path) as db:
|
||||
await init_db(db)
|
||||
await db.execute("DROP TABLE schema_migrations;")
|
||||
await db.commit()
|
||||
await init_db(db)
|
||||
async with db.execute(
|
||||
"SELECT version FROM schema_migrations ORDER BY version DESC LIMIT 1;"
|
||||
) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row[0] == 1
|
||||
|
||||
Reference in New Issue
Block a user