101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
"""Tests for app.db — database schema initialisation."""
|
|
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
|
|
from app.db import init_db
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_settings_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``settings`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='settings';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_sessions_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``sessions`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='sessions';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_blocklist_sources_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``blocklist_sources`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='blocklist_sources';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_creates_import_log_table(tmp_path: Path) -> None:
|
|
"""``init_db`` must create the ``import_log`` table."""
|
|
db_path = str(tmp_path / "test.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await init_db(db)
|
|
async with db.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='import_log';"
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
assert row is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_is_idempotent(tmp_path: Path) -> None:
|
|
"""Calling ``init_db`` twice on the same database must not raise."""
|
|
db_path = str(tmp_path / "test.db")
|
|
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
|