diff --git a/src/storops/core/migrate.py b/src/storops/core/migrate.py index 25c3a0f..7ca03e4 100644 --- a/src/storops/core/migrate.py +++ b/src/storops/core/migrate.py @@ -12,6 +12,7 @@ import json import os import shutil +import sys from datetime import datetime, timezone from pathlib import Path @@ -107,6 +108,23 @@ def execute(plan_file: str, *, confirm: bool = False, app_closed: bool = False) raw = json.loads(Path(plan_file).read_text(encoding="utf-8")) source, destination = raw["Source"], raw["Destination"] + # plan() already refuses dest==src, but the plan file is re-read from disk + # here and may no longer be the one plan() wrote. An empty source directory + # would slip past the non-empty-destination check below, the copy would + # trivially "verify" (0==0), and rmtree would then delete the directory as + # its own destination. normcase folds case on Windows only; darwin folds + # explicitly because default APFS volumes are case-insensitive too. + resolved_source = resolve_path(source) + resolved_destination = resolve_path(destination) + collide = os.path.normcase(resolved_source) == os.path.normcase(resolved_destination) + if sys.platform == "darwin": + collide = collide or resolved_source.lower() == resolved_destination.lower() + if collide: + raise UnsupportedOperationError( + f"StorOps: plan is invalid -- destination equals the source ('{source}'). " + "Re-run `storops migrate plan` with a distinct destination." + ) + if not os.path.isdir(source): raise StalePlanError(f"StorOps: source '{source}' no longer exists or is not a directory -- the plan is stale. Re-run `storops migrate plan`.") diff --git a/tests/unit/test_migrate_execute.py b/tests/unit/test_migrate_execute.py new file mode 100644 index 0000000..dc06144 --- /dev/null +++ b/tests/unit/test_migrate_execute.py @@ -0,0 +1,63 @@ +"""Regression tests for migrate.execute()'s destination==source guard. + +plan() refuses to generate a plan whose destination equals its source, but +execute() re-reads the plan JSON from disk, so a hand-edited or otherwise +tampered plan file can still pair a directory with itself. Without the guard, +an *empty* source directory slips past execute()'s "destination must be +empty" check, the copy degenerates to a no-op whose verification trivially +matches (0 files == 0 files), and shutil.rmtree() then deletes the directory +as its own destination. +""" +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path + +import pytest + +from storops.core.errors import UnsupportedOperationError +from storops.core.migrate import execute as migrate_execute + + +def _write_plan(tmp_path: Path, source: Path, destination: Path) -> Path: + plan_file = tmp_path / "storops-migrate-plan.json" + plan_file.write_text( + json.dumps({"Source": str(source), "Destination": str(destination)}), + encoding="utf-8", + ) + return plan_file + + +def test_execute_rejects_plan_whose_destination_equals_source(tmp_path): + src = tmp_path / "models" + src.mkdir() # empty on purpose: the non-empty-destination check must not be what saves us + plan_file = _write_plan(tmp_path, src, src) + + with pytest.raises(UnsupportedOperationError, match="destination equals the source"): + migrate_execute(str(plan_file), confirm=True) + + +def test_execute_rejects_destination_equal_after_path_normalization(tmp_path): + # Same directory spelled with a trailing separator -- resolve_path() + # normalizes both sides before comparing. + src = tmp_path / "models" + src.mkdir() + plan_file = _write_plan(tmp_path, src, Path(str(src) + os.sep)) + + with pytest.raises(UnsupportedOperationError, match="destination equals the source"): + migrate_execute(str(plan_file), confirm=True) + + +@pytest.mark.skipif( + sys.platform not in ("win32", "darwin"), + reason="only case-insensitive filesystems make the upper-cased path the same directory", +) +def test_execute_rejects_destination_equal_case_insensitively(tmp_path): + src = tmp_path / "models" + src.mkdir() + plan_file = _write_plan(tmp_path, src, Path(str(src).upper())) + + with pytest.raises(UnsupportedOperationError, match="destination equals the source"): + migrate_execute(str(plan_file), confirm=True)