AdrianoDev
49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
from pathlib import Path
|
|
|
|
from mcp_common.storage import Database, run_migrations
|
|
|
|
|
|
def test_database_creates_wal(tmp_path: Path):
|
|
db_path = tmp_path / "test.db"
|
|
db = Database(db_path)
|
|
db.connect()
|
|
# WAL mode attivo
|
|
mode = db.conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
assert mode.lower() == "wal"
|
|
db.close()
|
|
|
|
|
|
def test_database_migrations_run_once(tmp_path: Path):
|
|
db_path = tmp_path / "test.db"
|
|
db = Database(db_path)
|
|
db.connect()
|
|
migrations = {
|
|
1: "CREATE TABLE foo (id INTEGER PRIMARY KEY, name TEXT);",
|
|
2: "ALTER TABLE foo ADD COLUMN value INTEGER DEFAULT 0;",
|
|
}
|
|
run_migrations(db.conn, migrations)
|
|
# Second run: should be no-op
|
|
run_migrations(db.conn, migrations)
|
|
cols = [r[1] for r in db.conn.execute("PRAGMA table_info(foo)").fetchall()]
|
|
assert "name" in cols
|
|
assert "value" in cols
|
|
version = db.conn.execute("SELECT MAX(version) FROM _schema_version").fetchone()[0]
|
|
assert version == 2
|
|
db.close()
|
|
|
|
|
|
def test_database_partial_migration(tmp_path: Path):
|
|
db_path = tmp_path / "test.db"
|
|
db = Database(db_path)
|
|
db.connect()
|
|
migrations_v1 = {1: "CREATE TABLE foo (id INTEGER);"}
|
|
run_migrations(db.conn, migrations_v1)
|
|
migrations_v2 = {**migrations_v1, 2: "CREATE TABLE bar (id INTEGER);"}
|
|
run_migrations(db.conn, migrations_v2)
|
|
tables = {r[0] for r in db.conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall()}
|
|
assert "foo" in tables
|
|
assert "bar" in tables
|
|
db.close()
|