Stricter versioning: always store migrate version record, file must be within versions included in migrations. Log only for migrations that made changes. Allow deleting older migration functions when no longer required.

tests/test_kanta_integration.py

@@ -17,6 +17,7 @@ from .support import (
     change_actions,
     fixed_change,
     make_kanta,
+    make_migrations_module,
     read_changes,
     seed_single_change,
 )
@@ -473,6 +474,43 @@ async def test_msgspec_normalization_logs_migration(tmp_path, format_config):
     assert "migrate:msgspec" in change_actions(path, format_config)
 
 
+@pytest.mark.asyncio
+async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(
+        path, fixed_change("init", {"counter": 0, "users": {}}), format_config
+    )
+
+    def migrate_v1(d, kanta):
+        """No-op migration that only bumps the schema version."""
+        pass
+
+    mod = make_migrations_module("empty_migration_mod", "migrate_v1", migrate_v1)
+
+    try:
+        kanta = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta.open()
+        assert kanta.version == 1
+        await kanta.flush()
+        await kanta.close()
+
+        records = read_changes(path, format_config)
+        migration_records = [r for r in records if r.a.startswith("migrate")]
+        assert len(migration_records) == 1
+        assert migration_records[0].v == 1
+        assert migration_records[0].diff == {}
+
+        kanta2 = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta2.open()
+        assert kanta2.version == 1
+        await kanta2.close()
+
+        records2 = read_changes(path, format_config)
+        assert len([r for r in records2 if r.a.startswith("migrate")]) == 1
+    finally:
+        sys.modules.pop("empty_migration_mod", None)
+
+
 @pytest.mark.asyncio
 async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
     path = tmp_path / "test.db"

tests/test_migrations.py

@@ -1,5 +1,9 @@
+import logging
 from types import ModuleType, SimpleNamespace
 
+import pytest
+
+from kanta.exceptions import DatabaseError
 from kanta.migrations import Migrations
 
 
@@ -89,3 +93,112 @@ def test_migration_can_omit_kanta_argument():
     new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
     assert new_ver == 1
     assert state["x"] == 1
+
+
+def test_version_too_new():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v2 is newer than the highest supported version v1",
+    ):
+        reg.apply({}, current_version=2, kanta=kanta, silent=True)
+
+
+def test_version_too_old():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v1 is older than the minimum supported version v2",
+    ):
+        reg.apply({}, current_version=1, kanta=kanta, silent=True)
+
+
+def test_missing_middle_migration_is_skipped():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v3(d):
+        d["y"] = 3
+
+    state = {"x": 1}
+    new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 1
+    assert state["y"] == 3
+
+
+def test_old_migrations_deleted_current_supported():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    state = {"x": 2}
+    new_ver = reg.apply(state, current_version=2, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 3
+
+
+def test_migration_log_only_when_changed(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """Set x."""
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v2(d):
+        """No-op."""
+        pass
+
+    @reg.register
+    def migrate_v3(d):
+        """Set y."""
+        d["y"] = 3
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(messages) == 2
+    assert "migrate_v1" in messages[0]
+    assert "Set x" in messages[0]
+    assert "migrate_v3" in messages[1]
+    assert "Set y" in messages[1]
+
+
+def test_no_op_migration_produces_no_log(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """No-op."""
+        pass
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages