52 lines
2.9 KiB
Python
52 lines
2.9 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_task_completed_at_migration_upgrade_and_downgrade(tmp_path: Path):
|
|
repo = Path(__file__).resolve().parents[1]
|
|
database = tmp_path / "migration.sqlite3"
|
|
assert run_alembic(repo, database, "upgrade", "0017_memos").returncode == 0
|
|
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"INSERT INTO users (id, username, password_hash, timezone, created_at) VALUES (?, ?, ?, ?, ?)",
|
|
("00000000-0000-0000-0000-000000000001", "owner", "hash", "Asia/Shanghai", "2026-09-15 00:00:00"),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO task_lists (id, user_id, folder_id, name, is_inbox, position, created_at, deleted_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
|
("00000000-0000-0000-0000-000000000002", "00000000-0000-0000-0000-000000000001", None, "收集箱", 1, 0, "2026-09-15 00:00:00", None),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO tasks (id, user_id, list_id, parent_id, title, description, priority, completed, due_at, due_has_time, version, position, created_at, updated_at, deleted_at, external_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
("00000000-0000-0000-0000-000000000003", "00000000-0000-0000-0000-000000000001", "00000000-0000-0000-0000-000000000002", None, "已完成任务", "", 0, 1, None, 0, 1, 0, "2026-09-14 00:00:00", "2026-09-15 06:00:00", None, None),
|
|
)
|
|
connection.commit()
|
|
|
|
upgraded = run_alembic(repo, database, "upgrade", "0018_task_completed_at")
|
|
assert upgraded.returncode == 0, upgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
columns = {row[1]: row for row in connection.execute("PRAGMA table_info(tasks)")}
|
|
assert "completed_at" in columns
|
|
indexes = {row[1] for row in connection.execute("PRAGMA index_list(tasks)")}
|
|
assert "ix_tasks_completed_at" in indexes
|
|
completed_at = connection.execute("SELECT completed_at FROM tasks WHERE id = ?", ("00000000-0000-0000-0000-000000000003",)).fetchone()
|
|
assert completed_at == ("2026-09-15 06:00:00",)
|
|
sqlite_boolean = connection.execute("SELECT completed IS TRUE FROM tasks WHERE id = ?", ("00000000-0000-0000-0000-000000000003",)).fetchone()
|
|
assert sqlite_boolean == (1,)
|
|
|
|
downgraded = run_alembic(repo, database, "downgrade", "0017_memos")
|
|
assert downgraded.returncode == 0, downgraded.stderr
|
|
with sqlite3.connect(database) as connection:
|
|
columns = {row[1]: row for row in connection.execute("PRAGMA table_info(tasks)")}
|
|
assert "completed_at" not in columns
|