57 lines
2.1 KiB
Python
57 lines
2.1 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_recurrence_trigger_migration_downgrade_and_reupgrade_with_data(tmp_path: Path):
|
|
repo = Path(__file__).resolve().parents[1]
|
|
database = tmp_path / "migration.sqlite3"
|
|
assert run_alembic(repo, database, "upgrade", "0015_purge_operations").returncode == 0
|
|
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"INSERT INTO recurrence_templates "
|
|
"(id, user_id, task_id, rrule, starts_at, ends_at, created_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(
|
|
"00000000000000000000000000000001",
|
|
"00000000000000000000000000000002",
|
|
"00000000000000000000000000000003",
|
|
"FREQ=DAILY",
|
|
"2026-03-08 07:30:00",
|
|
None,
|
|
"2026-03-01 00:00:00",
|
|
),
|
|
)
|
|
|
|
upgraded = run_alembic(repo, database, "upgrade", "0016_recurrence_trigger_modes")
|
|
assert upgraded.returncode == 0, upgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"UPDATE recurrence_templates SET trigger_mode='after_completion', "
|
|
"after_completion_days=1, rrule=NULL"
|
|
)
|
|
|
|
downgraded = run_alembic(repo, database, "downgrade", "0015_purge_operations")
|
|
assert downgraded.returncode == 0, downgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
row = connection.execute("SELECT rrule FROM recurrence_templates").fetchone()
|
|
assert row == ("FREQ=DAILY;INTERVAL=1",)
|
|
|
|
reupgraded = run_alembic(repo, database, "upgrade", "0016_recurrence_trigger_modes")
|
|
assert reupgraded.returncode == 0, reupgraded.stderr
|