tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(db_migrate): extract _apply_migrations_until helper; add edge-case tests

Browse Source 
Issue 1 (refactor): run() and migrate_to_version() duplicated the inner
loop body — every change to the migration loop had to be applied in both
places. Pull the loop into _apply_migrations_until(engine, target_version):
when target_version is None, apply all pending; otherwise stop when the
file's version exceeds target. run() and migrate_to_version() are now
thin wrappers around the helper.

Issue 2 (tests): migrate_to_version() had no direct tests for its
edge-case behavior. Add 4 tests:
- idempotent: calling migrate_to_version(engine, N) twice == once
- lower-than-current: migrate_to_version(engine, 3) after N=13 is a no-op
- zero on fresh DB: migrate_to_version(engine, 0) is a no-op
- beyond-max: migrate_to_version(engine, 999) applies every migration

Tests use a fresh sa.Engine (not db.engine()) so they exercise the
target-version behavior from a clean user_version=0 starting point —
the conftest autouse fixture has already brought the module engine to
the latest version, which would mask the no-op branches.
This commit is contained in:
Grok
2026-06-21 19:43:23 -06:00
parent ced8d20aed
commit c055787bd4
2 changed files with 88 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/db_migrate.py
+21 -37
View File
@@ -47,36 +47,43 @@ def _strip_comments(sql: str) -> str:
return "\n".join(lines)
def run(engine: sa.Engine) -> None:
"""Apply any pending migrations. Idempotent."""
migrations_dir = MIGRATIONS_DIR
if not migrations_dir.exists():
return # no migrations directory yet — fresh checkout, no-op
migration_files = sorted(migrations_dir.glob("*.sql"))
if not migration_files:
return
def _apply_migrations_until(engine: sa.Engine, target_version: int | None) -> None:
"""Apply migrations up to ``target_version`` (inclusive).
If ``target_version`` is ``None``, apply every pending migration. If a
migration's version is greater than ``target_version``, stop iterating.
Idempotent: migrations whose version is <= the current ``PRAGMA user_version``
are skipped.
"""
with engine.begin() as conn:
current = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
for path in migration_files:
for path in sorted(MIGRATIONS_DIR.glob("*.sql")):
sql = path.read_text()
version = _migration_version(sql, path.name)
if version <= current:
continue
if target_version is not None and version > target_version:
break
statements_sql = _strip_comments(sql)
statements = [s.strip() for s in statements_sql.split(";") if s.strip()]
with engine.begin() as conn:
for stmt in statements:
conn.exec_driver_sql(stmt)
conn.exec_driver_sql(f"PRAGMA user_version = {version}")
current = version
def run(engine: sa.Engine) -> None:
"""Apply any pending migrations. Idempotent."""
if not MIGRATIONS_DIR.exists():
return # no migrations directory yet — fresh checkout, no-op
if not list(MIGRATIONS_DIR.glob("*.sql")):
return
_apply_migrations_until(engine, None)
def migrate_to_version(engine: sa.Engine, target_version: int) -> None:
"""Apply migrations up to and including ``target_version``. Idempotent.
@@ -92,29 +99,6 @@ def migrate_to_version(engine: sa.Engine, target_version: int) -> None:
Calling with ``target_version`` <= the current version is a no-op.
"""
migrations_dir = MIGRATIONS_DIR
if not migrations_dir.exists():
if not MIGRATIONS_DIR.exists():
return
migration_files = sorted(migrations_dir.glob("*.sql"))
if not migration_files:
return
with engine.begin() as conn:
current = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
for path in migration_files:
sql = path.read_text()
version = _migration_version(sql, path.name)
if version <= current:
continue
if version > target_version:
break
statements_sql = _strip_comments(sql)
statements = [s.strip() for s in statements_sql.split(";") if s.strip()]
with engine.begin() as conn:
for stmt in statements:
conn.exec_driver_sql(stmt)
conn.exec_driver_sql(f"PRAGMA user_version = {version}")
_apply_migrations_until(engine, target_version)
backend/tests/test_db_migrate.py
+67 -1
View File
@@ -587,4 +587,70 @@ def test_migration_0014_preserves_joint_fk_rows(tmp_path, monkeypatch):
assert rows == [("CLM-J", "B-JOIN", "CLP-J", "B-JOIN-R")], (
"matches batch_id + remittance_batch_id must be populated by 0014; "
"got " + repr(rows)
)
)
# =============================================================================
# migrate_to_version: edge-case behavior
# =============================================================================
# These tests use a fresh ``sa.Engine`` (not ``db.engine()``) because the
# autouse conftest fixture has already initialized the module engine at the
# latest version. We need a clean engine that starts at ``user_version=0``
# so that ``migrate_to_version(N)`` actually exercises the apply / no-op /
# beyond-max branches.
def test_migrate_to_version_idempotent(tmp_path, monkeypatch):
"""Calling migrate_to_version(engine, N) twice is equivalent to calling once."""
from cyclone.db_migrate import migrate_to_version
engine = _fresh_engine(tmp_path)
migrate_to_version(engine, 5)
with engine.begin() as conn:
v1 = conn.exec_driver_sql("PRAGMA user_version").scalar()
migrate_to_version(engine, 5)
with engine.begin() as conn:
v2 = conn.exec_driver_sql("PRAGMA user_version").scalar()
assert v1 == v2 == 5
def test_migrate_to_version_lower_than_current_is_noop(tmp_path, monkeypatch):
"""migrate_to_version(engine, 3) after migrate_to_version(engine, 13) is a no-op."""
from cyclone.db_migrate import migrate_to_version
engine = _fresh_engine(tmp_path)
migrate_to_version(engine, 13)
with engine.begin() as conn:
v_after_13 = conn.exec_driver_sql("PRAGMA user_version").scalar()
migrate_to_version(engine, 3) # lower, no-op
with engine.begin() as conn:
v_after_3 = conn.exec_driver_sql("PRAGMA user_version").scalar()
assert v_after_13 == v_after_3 == 13
def test_migrate_to_version_zero_on_fresh_db(tmp_path, monkeypatch):
"""migrate_to_version(engine, 0) on a fresh DB is a no-op."""
from cyclone.db_migrate import migrate_to_version
engine = _fresh_engine(tmp_path)
migrate_to_version(engine, 0)
with engine.begin() as conn:
v = conn.exec_driver_sql("PRAGMA user_version").scalar()
assert v == 0
def test_migrate_to_version_beyond_max_applies_all(tmp_path, monkeypatch):
"""migrate_to_version(engine, 999) applies every migration (equivalent to run)."""
import re
from cyclone.db_migrate import MIGRATIONS_DIR, migrate_to_version
engine = _fresh_engine(tmp_path)
migrate_to_version(engine, 999)
with engine.begin() as conn:
v = conn.exec_driver_sql("PRAGMA user_version").scalar()
# Should equal the highest version present in the migrations directory.
max_version = max(
int(re.match(r"^(\d+)", p.name).group(1))
for p in MIGRATIONS_DIR.glob("*.sql")
)
assert v == max_version