test_db.py (3217B)
1 """T008 — Tests for DB connection and migration runner.""" 2 3 import asyncio 4 from pathlib import Path 5 6 import pytest 7 8 from app.db import Database 9 10 11 @pytest.fixture 12 async def db(temp_db_path: Path) -> Database: 13 """Yield a connected database, close after test.""" 14 database = Database(str(temp_db_path)) 15 await database.connect() 16 yield database 17 await database.close() 18 19 20 class TestConnection: 21 """SQLite connection basics.""" 22 23 async def test_connects_to_sqlite(self, db: Database) -> None: 24 """Connection is established without error.""" 25 assert db.connection is not None 26 27 async def test_wal_mode_enabled(self, db: Database) -> None: 28 """WAL journal mode is set for concurrent reads.""" 29 result = await db.execute_fetchall("PRAGMA journal_mode") 30 assert result[0][0] == "wal" 31 32 async def test_close_releases_connection(self, temp_db_path: Path) -> None: 33 """After close(), the connection object is gone.""" 34 database = Database(str(temp_db_path)) 35 await database.connect() 36 await database.close() 37 assert database.connection is None 38 39 40 class TestExecute: 41 """Direct execute method.""" 42 43 async def test_execute_inserts_data(self, db: Database) -> None: 44 """execute() inserts rows without error.""" 45 await db.execute( 46 "CREATE TABLE test_t (id INTEGER PRIMARY KEY, val TEXT)" 47 ) 48 await db.execute("INSERT INTO test_t (val) VALUES (?)", ("hello",)) 49 rows = await db.fetch_all("SELECT val FROM test_t") 50 assert rows[0][0] == "hello" 51 52 53 class TestRequireConnection: 54 """_require_connection raises when not connected.""" 55 56 def test_raises_when_not_connected(self) -> None: 57 """RuntimeError raised if connect() was not called.""" 58 from pathlib import Path 59 db = Database(str(Path("/tmp/noexist.db"))) 60 import pytest 61 with pytest.raises(RuntimeError, match="not connected"): 62 db._require_connection() 63 64 65 class TestMigrations: 66 """Migration runner behaviour.""" 67 68 async def test_runs_migrations_creates_tables(self, db: Database) -> None: 69 """Applying 0001_init.sql creates the expected tables.""" 70 await db.run_migrations(Path(__file__).resolve().parent.parent / "app" / "migrations") 71 tables = await db.fetch_all( 72 "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" 73 ) 74 table_names = {row[0] for row in tables} 75 assert "topics" in table_names 76 assert "topic_sessions" in table_names 77 assert "attachments" in table_names 78 assert "branches" in table_names 79 assert "auth_config" in table_names 80 81 async def test_migration_is_idempotent(self, db: Database) -> None: 82 """Running migrations twice does not raise.""" 83 migrations_dir = Path(__file__).resolve().parent.parent / "app" / "migrations" 84 await db.run_migrations(migrations_dir) 85 await db.run_migrations(migrations_dir) 86 tables = await db.fetch_all( 87 "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" 88 ) 89 table_names = {row[0] for row in tables} 90 assert "topics" in table_names