37 lines
1.6 KiB
Python
37 lines
1.6 KiB
Python
import os
|
|
import sqlite3
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
def run_alembic(repo: Path, database: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["DODO_DATABASE_URL"] = f"sqlite+aiosqlite:///{database}"
|
|
return subprocess.run(
|
|
["uv", "run", "alembic", *args], cwd=repo, env=env, text=True, capture_output=True, check=False
|
|
)
|
|
|
|
|
|
def test_memos_migration_upgrade_and_downgrade(tmp_path: Path):
|
|
repo = Path(__file__).resolve().parents[1]
|
|
database = tmp_path / "migration.sqlite3"
|
|
assert run_alembic(repo, database, "upgrade", "0016_recurrence_trigger_modes").returncode == 0
|
|
|
|
upgraded = run_alembic(repo, database, "upgrade", "0017_memos")
|
|
assert upgraded.returncode == 0, upgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
columns = {row[1]: row for row in connection.execute("PRAGMA table_info(memos)")}
|
|
assert set(columns) == {"id", "user_id", "title", "content", "version", "created_at", "updated_at", "deleted_at"}
|
|
assert columns["title"][3] == 1
|
|
assert columns["content"][3] == 1
|
|
assert columns["version"][3] == 1
|
|
indexes = {row[1] for row in connection.execute("PRAGMA index_list(memos)")}
|
|
assert "ix_memos_user_deleted_updated" in indexes
|
|
|
|
downgraded = run_alembic(repo, database, "downgrade", "0016_recurrence_trigger_modes")
|
|
assert downgraded.returncode == 0, downgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
assert connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='memos'"
|
|
).fetchone() is None
|