47 lines
2.2 KiB
Python
47 lines
2.2 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_backup_migration_upgrade_downgrade_and_reupgrade(tmp_path):
|
|
repo = Path(__file__).resolve().parents[1]
|
|
database = tmp_path / "migration.sqlite3"
|
|
assert run_alembic(repo, database, "upgrade", "0018_task_completed_at").returncode == 0
|
|
upgraded = run_alembic(repo, database, "upgrade", "0019_backup_imports")
|
|
assert upgraded.returncode == 0, upgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
tables = {row[0] for row in connection.execute("select name from sqlite_master where type='table'")}
|
|
assert {"backup_imports", "backup_import_entities", "backup_preflights"} <= tables
|
|
downgraded = run_alembic(repo, database, "downgrade", "0018_task_completed_at")
|
|
assert downgraded.returncode == 0, downgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
tables = {row[0] for row in connection.execute("select name from sqlite_master where type='table'")}
|
|
assert "backup_imports" not in tables
|
|
assert "backup_import_entities" not in tables
|
|
assert "backup_preflights" not in tables
|
|
reupgraded = run_alembic(repo, database, "upgrade", "head")
|
|
assert reupgraded.returncode == 0, reupgraded.stderr
|
|
|
|
|
|
def test_fresh_upgrade_has_single_head_and_backup_tables(tmp_path):
|
|
repo = Path(__file__).resolve().parents[1]
|
|
heads = run_alembic(repo, tmp_path / "unused.sqlite3", "heads")
|
|
assert heads.returncode == 0, heads.stderr
|
|
assert heads.stdout.count("(head)") == 1
|
|
database = tmp_path / "fresh.sqlite3"
|
|
upgraded = run_alembic(repo, database, "upgrade", "head")
|
|
assert upgraded.returncode == 0, upgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
tables = {row[0] for row in connection.execute("select name from sqlite_master where type='table'")}
|
|
assert {"backup_imports", "backup_import_entities", "backup_preflights"} <= tables
|