59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
|
|
from app.db import open_db
|
|
|
|
|
|
async def test_open_db_applies_hardening_pragmas(tmp_path: Path) -> None:
|
|
database_path = str(tmp_path / "bangui_test.db")
|
|
db = await open_db(database_path)
|
|
try:
|
|
async with db.execute("PRAGMA journal_mode;") as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None and row[0].lower() == "wal"
|
|
|
|
async with db.execute("PRAGMA foreign_keys;") as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None and row[0] == 1
|
|
|
|
async with db.execute("PRAGMA busy_timeout;") as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None and row[0] == 5000
|
|
finally:
|
|
await db.close()
|
|
|
|
|
|
async def test_open_db_respects_busy_timeout_for_concurrent_writes(tmp_path: Path) -> None:
|
|
database_path = str(tmp_path / "bangui_lock.db")
|
|
connection_a = await open_db(database_path)
|
|
try:
|
|
await connection_a.execute(
|
|
"CREATE TABLE IF NOT EXISTS test_lock (id INTEGER PRIMARY KEY, value TEXT);"
|
|
)
|
|
await connection_a.commit()
|
|
|
|
await connection_a.execute("BEGIN EXCLUSIVE;")
|
|
|
|
async def write_after_lock() -> None:
|
|
connection_b = await open_db(database_path)
|
|
try:
|
|
await connection_b.execute(
|
|
"INSERT INTO test_lock (value) VALUES ('locked');"
|
|
)
|
|
await connection_b.commit()
|
|
finally:
|
|
await connection_b.close()
|
|
|
|
task = asyncio.create_task(write_after_lock())
|
|
await asyncio.sleep(0.1)
|
|
await connection_a.commit()
|
|
await task
|
|
|
|
async with connection_a.execute("SELECT value FROM test_lock;") as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None and row[0] == "locked"
|
|
finally:
|
|
await connection_a.close()
|