Merge 8665adb695 into 9c66bafa52

sqlite_utils/cli.py

@@ -3598,6 +3598,20 @@ def migrate(db_path, migrations, stop_before, list_, verbose):
                     ", ".join(unknown)
                 )
             )
+        for migration_set in migration_sets:
+            matches = _stop_before_for_migration_set(stop_before, migration_set.name)
+            if hasattr(migration_set, "applied"):
+                applied_names = {m.name for m in migration_set.applied(db)}
+                already_applied = set(matches).intersection(applied_names)
+                if already_applied:
+                    raise click.ClickException(
+                        "Cannot stop before migration{} {} in set '{}' - already "
+                        "been applied".format(
+                            "s" if len(already_applied) > 1 else "",
+                            ", ".join(sorted(already_applied)),
+                            migration_set.name,
+                        )
+                    )
     for migration_set in migration_sets:
         matches = _stop_before_for_migration_set(stop_before, migration_set.name)
         if isinstance(migration_set, sqlite_utils.Migrations):

tests/test_cli_migrate.py

@@ -506,3 +506,38 @@ def test_list_with_legacy_class_is_read_only(tmpdir):
     db2 = sqlite_utils.Database(db_path)
     assert "_sqlite_migrations" not in db2.table_names()
     db2.close()
+
+
+def test_stop_before_applied_migration_multiple_sets_does_not_apply_earlier(tmpdir):
+    root = pathlib.Path(tmpdir)
+    for name in ("a", "b"):
+        (root / f"{name}.py").write_text(
+            "from sqlite_utils import Migrations\n"
+            f'migrations = Migrations("{name}")\n'
+            "@migrations()\n"
+            "def first(db):\n"
+            f'    db.table("{name}").insert({{"value": 1}})\n',
+            "utf-8",
+        )
+    db_path = str(root / "test.db")
+    runner = CliRunner()
+    # Apply set b first
+    seed = runner.invoke(sqlite_utils.cli.cli, ["migrate", db_path, str(root / "b.py")])
+    assert seed.exit_code == 0
+    # Now migrate both a and b with --stop-before b:first (which is already applied).
+    # This must error and must NOT apply set a.
+    result = runner.invoke(
+        sqlite_utils.cli.cli,
+        [
+            "migrate",
+            db_path,
+            str(root / "a.py"),
+            str(root / "b.py"),
+            "--stop-before",
+            "b:first",
+        ],
+    )
+    assert result.exit_code != 0
+    assert "already been applied" in result.output
+    db = sqlite_utils.Database(db_path)
+    assert not db.table("a").exists()