45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
|
|
from porchlight.store.sqlite.db import open_db
|
|
|
|
MIGRATIONS_DIR = (
|
|
Path(__file__).resolve().parent.parent.parent / "src" / "porchlight" / "store" / "sqlite" / "migrations"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def migrations_dir() -> Path:
|
|
return MIGRATIONS_DIR
|
|
|
|
|
|
async def test_open_db_returns_connection(tmp_path: Path, migrations_dir: Path) -> None:
|
|
db_path = str(tmp_path / "test.db")
|
|
async with open_db(db_path, migrations_dir) as db:
|
|
assert isinstance(db, aiosqlite.Connection)
|
|
|
|
|
|
async def test_open_db_runs_migrations(tmp_path: Path, migrations_dir: Path) -> None:
|
|
db_path = str(tmp_path / "test.db")
|
|
async with (
|
|
open_db(db_path, migrations_dir) as db,
|
|
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_open_db_sets_wal_mode(tmp_path: Path, migrations_dir: Path) -> None:
|
|
db_path = str(tmp_path / "test.db")
|
|
async with open_db(db_path, migrations_dir) as db, db.execute("PRAGMA journal_mode") as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row[0] == "wal"
|
|
|
|
|
|
async def test_open_db_creates_parent_dirs(tmp_path: Path, migrations_dir: Path) -> None:
|
|
db_path = str(tmp_path / "sub" / "dir" / "test.db")
|
|
async with open_db(db_path, migrations_dir) as db:
|
|
assert isinstance(db, aiosqlite.Connection)
|
|
assert (tmp_path / "sub" / "dir" / "test.db").exists()
|