82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from backend.database import Migration, MigrationError, MigrationRunner
|
|
from database import ReviewDatabase
|
|
|
|
|
|
class DatabaseMigrationTests(unittest.TestCase):
|
|
def test_fresh_database_records_the_adopted_schema_once(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
path = Path(root) / "review.db"
|
|
database = ReviewDatabase(path)
|
|
with database.connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT version, name FROM schema_migrations"
|
|
).fetchall()
|
|
self.assertEqual(
|
|
[(row["version"], row["name"]) for row in rows],
|
|
[
|
|
("0001", "adopt_legacy_schema"),
|
|
("0002", "create_job_runs"),
|
|
("0003", "extend_llm_audit"),
|
|
],
|
|
)
|
|
ReviewDatabase(path)
|
|
with database.connect() as connection:
|
|
count = connection.execute(
|
|
"SELECT COUNT(*) AS count FROM schema_migrations"
|
|
).fetchone()["count"]
|
|
self.assertEqual(count, 3)
|
|
|
|
def test_connection_factory_enables_required_pragmas(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = ReviewDatabase(Path(root) / "review.db")
|
|
with database.connect() as connection:
|
|
self.assertEqual(connection.execute("PRAGMA foreign_keys").fetchone()[0], 1)
|
|
self.assertEqual(connection.execute("PRAGMA journal_mode").fetchone()[0], "wal")
|
|
self.assertEqual(connection.execute("PRAGMA busy_timeout").fetchone()[0], 20000)
|
|
|
|
def test_failed_migration_rolls_back_and_is_not_recorded(self) -> None:
|
|
connection = sqlite3.connect(":memory:")
|
|
self.addCleanup(connection.close)
|
|
connection.row_factory = sqlite3.Row
|
|
|
|
def fail(conn: sqlite3.Connection) -> None:
|
|
conn.execute("CREATE TABLE should_rollback (id INTEGER)")
|
|
raise RuntimeError("stop")
|
|
|
|
migration = Migration("9000", "failure", fail, "failure:v1")
|
|
with self.assertRaises(MigrationError):
|
|
MigrationRunner().apply(connection, (migration,))
|
|
tables = {
|
|
row["name"]
|
|
for row in connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
)
|
|
}
|
|
self.assertNotIn("should_rollback", tables)
|
|
self.assertEqual(
|
|
connection.execute("SELECT COUNT(*) FROM schema_migrations").fetchone()[0],
|
|
0,
|
|
)
|
|
|
|
def test_applied_migration_checksum_is_immutable(self) -> None:
|
|
connection = sqlite3.connect(":memory:")
|
|
self.addCleanup(connection.close)
|
|
connection.row_factory = sqlite3.Row
|
|
first = Migration("9001", "example", lambda conn: None, "example:v1")
|
|
changed = Migration("9001", "example", lambda conn: None, "example:v2")
|
|
runner = MigrationRunner()
|
|
runner.apply(connection, (first,))
|
|
with self.assertRaises(MigrationError):
|
|
runner.apply(connection, (changed,))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|