42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
|
|
from fastapi_oidc_op.store.sqlite.migrations import run_migrations
|
|
|
|
MIGRATIONS_DIR = (
|
|
Path(__file__).resolve().parent.parent.parent / "src" / "fastapi_oidc_op" / "store" / "sqlite" / "migrations"
|
|
)
|
|
|
|
|
|
async def test_run_migrations_applies_initial() -> None:
|
|
async with aiosqlite.connect(":memory:") as db:
|
|
await db.execute("PRAGMA foreign_keys=ON")
|
|
count = await run_migrations(db, MIGRATIONS_DIR)
|
|
assert count == 1
|
|
async with db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='users'") as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
async def test_run_migrations_skips_already_applied() -> None:
|
|
async with aiosqlite.connect(":memory:") as db:
|
|
await db.execute("PRAGMA foreign_keys=ON")
|
|
first_count = await run_migrations(db, MIGRATIONS_DIR)
|
|
second_count = await run_migrations(db, MIGRATIONS_DIR)
|
|
assert first_count == 1
|
|
assert second_count == 0
|
|
|
|
|
|
async def test_run_migrations_creates_all_tables() -> None:
|
|
async with aiosqlite.connect(":memory:") as db:
|
|
await db.execute("PRAGMA foreign_keys=ON")
|
|
await run_migrations(db, MIGRATIONS_DIR)
|
|
async with db.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") as cursor:
|
|
tables = [row[0] async for row in cursor]
|
|
assert "users" in tables
|
|
assert "user_groups" in tables
|
|
assert "webauthn_credentials" in tables
|
|
assert "password_credentials" in tables
|
|
assert "magic_links" in tables
|
|
assert "_migrations" in tables
|