mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-10-24 10:17:05 +02:00
migrate: pre-validate --stop-before targets across all sets before applying
When sqlite-utils migrate is invoked with multiple migration sets, an already-applied --stop-before target in a subsequent set would cause an error inside that set's apply() call only after earlier sets had already executed and committed their pending migrations. Pre-validate across all loaded migration sets upfront that no requested --stop-before targets have already been applied before entering the application loop, ensuring no pending migrations are applied when an error is raised. Fixes #870 Signed-off-by: Samran Asif <samranwebdev2000@gmail.com>
This commit is contained in:
parent
85b1be10c8
commit
8665adb695
2 changed files with 49 additions and 0 deletions
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue