340 lines
12 KiB
Python
340 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import gyxx_flow.migration.data as migration_data
|
|
from gyxx_flow.core.artifacts import sha256_file
|
|
from gyxx_flow.migration.data import (
|
|
DataLayer,
|
|
HistoricalDataMigrationExecutor,
|
|
HistoricalDataMigrationPlanner,
|
|
MigrationConflictError,
|
|
MigrationExecutionError,
|
|
MigrationSource,
|
|
MigrationValidationError,
|
|
)
|
|
|
|
|
|
def _source_tree(root: Path) -> dict[str, bytes]:
|
|
files = {
|
|
"2025/01/orders.csv": b"order_id,total\n1,19.90\n",
|
|
"2025/01/detail.json": b'{"id": 1, "items": 2}\n',
|
|
}
|
|
for relative, content in files.items():
|
|
path = root / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(content)
|
|
return files
|
|
|
|
|
|
def _planner(data_root: Path, *, free_bytes: int = 10_000_000) -> HistoricalDataMigrationPlanner:
|
|
return HistoricalDataMigrationPlanner(
|
|
data_root=data_root,
|
|
free_space_provider=lambda _path: free_bytes,
|
|
)
|
|
|
|
|
|
def _source(root: Path, *, layer: DataLayer = DataLayer.RAW) -> MigrationSource:
|
|
return MigrationSource(
|
|
source_id="legacy-orders",
|
|
module="product_commerce",
|
|
layer=layer,
|
|
root=root,
|
|
)
|
|
|
|
|
|
def test_plan_is_machine_readable_and_routes_files_by_module_and_layer(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
contents = _source_tree(source_root)
|
|
manifest_path = tmp_path / "new" / "plans" / "migration.json"
|
|
|
|
plan = _planner(tmp_path / "new").create_plan(
|
|
[_source(source_root)], manifest_path=manifest_path
|
|
)
|
|
|
|
assert plan.source_summary.file_count == 2
|
|
assert plan.source_summary.total_bytes == sum(map(len, contents.values()))
|
|
assert {entry.relative_path for entry in plan.files} == set(contents)
|
|
assert {
|
|
entry.target_relative_path for entry in plan.files
|
|
} == {
|
|
f"data/raw/product_commerce/legacy/legacy-orders/{relative}"
|
|
for relative in contents
|
|
}
|
|
payload = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
assert payload["schema_version"] == 1
|
|
assert payload["source_summary"]["file_count"] == 2
|
|
assert all(len(entry["sha256"]) == 64 for entry in payload["files"])
|
|
assert payload["precheck"]["status"] == "ready"
|
|
|
|
|
|
@pytest.mark.parametrize("layer", list(DataLayer))
|
|
def test_each_supported_layer_has_an_explicit_target_directory(
|
|
tmp_path: Path, layer: DataLayer
|
|
) -> None:
|
|
source_root = tmp_path / f"legacy-{layer.value}"
|
|
(source_root / "one.txt").parent.mkdir(parents=True)
|
|
(source_root / "one.txt").write_text("one", encoding="utf-8")
|
|
|
|
plan = _planner(tmp_path / "new").create_plan([_source(source_root, layer=layer)])
|
|
|
|
assert plan.files[0].target_relative_path.startswith(
|
|
f"data/{layer.value}/product_commerce/legacy/legacy-orders/"
|
|
)
|
|
|
|
|
|
def test_default_execution_is_plan_only_and_never_creates_data_files(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
data_root = tmp_path / "new"
|
|
plan = _planner(data_root).create_plan([_source(source_root)])
|
|
|
|
report = HistoricalDataMigrationExecutor().execute(plan)
|
|
|
|
assert report.status == "plan_only"
|
|
assert report.applied is False
|
|
assert report.destination_summary.file_count == 0
|
|
assert not (data_root / "data").exists()
|
|
|
|
|
|
def test_apply_copies_without_changing_or_removing_source_and_reconciles(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
contents = _source_tree(source_root)
|
|
data_root = tmp_path / "new"
|
|
report_path = data_root / "evidence" / "migration-report.json"
|
|
before = {
|
|
relative: (
|
|
sha256_file(source_root / relative),
|
|
(source_root / relative).stat().st_mtime_ns,
|
|
)
|
|
for relative in contents
|
|
}
|
|
plan = _planner(data_root).create_plan([_source(source_root)])
|
|
|
|
report = HistoricalDataMigrationExecutor().execute(
|
|
plan, apply=True, report_path=report_path
|
|
)
|
|
|
|
assert report.status == "reconciled"
|
|
assert report.applied is True
|
|
assert report.source_summary == report.destination_summary
|
|
assert report.copied_count == 2
|
|
assert report.mismatches == ()
|
|
for relative, expected_content in contents.items():
|
|
source_path = source_root / relative
|
|
destination = (
|
|
data_root
|
|
/ "data/raw/product_commerce/legacy/legacy-orders"
|
|
/ relative
|
|
)
|
|
assert source_path.exists()
|
|
assert source_path.read_bytes() == expected_content
|
|
assert (sha256_file(source_path), source_path.stat().st_mtime_ns) == before[relative]
|
|
assert destination.read_bytes() == expected_content
|
|
assert sha256_file(destination) == before[relative][0]
|
|
payload = json.loads(report_path.read_text(encoding="utf-8"))
|
|
assert payload["status"] == "reconciled"
|
|
assert payload["destination_summary"] == payload["source_summary"]
|
|
|
|
|
|
def test_progress_reporting_performs_full_destination_reconciliation_only_once(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
plan = _planner(tmp_path / "new").create_plan([_source(source_root)])
|
|
original = migration_data._summarize_destinations
|
|
original_write = migration_data._write_report
|
|
calls = 0
|
|
writes = 0
|
|
|
|
def counted(entries, data_root):
|
|
nonlocal calls
|
|
calls += 1
|
|
return original(entries, data_root)
|
|
|
|
monkeypatch.setattr(migration_data, "_summarize_destinations", counted)
|
|
|
|
def counted_write(path, report):
|
|
nonlocal writes
|
|
writes += 1
|
|
return original_write(path, report)
|
|
|
|
monkeypatch.setattr(migration_data, "_write_report", counted_write)
|
|
|
|
report = HistoricalDataMigrationExecutor().execute(plan, apply=True)
|
|
|
|
assert report.status == "reconciled"
|
|
assert calls == 1
|
|
assert writes == 1
|
|
|
|
|
|
def test_repeated_apply_is_idempotent_and_skips_matching_files(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
data_root = tmp_path / "new"
|
|
plan = _planner(data_root).create_plan([_source(source_root)])
|
|
executor = HistoricalDataMigrationExecutor()
|
|
first = executor.execute(plan, apply=True)
|
|
destination = data_root / plan.files[0].target_relative_path
|
|
first_mtime = destination.stat().st_mtime_ns
|
|
|
|
second = executor.execute(plan, apply=True)
|
|
|
|
assert first.copied_count == 2
|
|
assert second.copied_count == 0
|
|
assert second.skipped_count == 2
|
|
assert destination.stat().st_mtime_ns == first_mtime
|
|
assert second.status == "reconciled"
|
|
|
|
|
|
def test_precheck_rejects_conflicting_target_before_copying_anything(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
data_root = tmp_path / "new"
|
|
conflict = (
|
|
data_root
|
|
/ "data/raw/product_commerce/legacy/legacy-orders/2025/01/orders.csv"
|
|
)
|
|
conflict.parent.mkdir(parents=True)
|
|
conflict.write_text("different", encoding="utf-8")
|
|
|
|
with pytest.raises(MigrationConflictError, match="conflict"):
|
|
_planner(data_root).create_plan([_source(source_root)])
|
|
|
|
assert not (
|
|
data_root
|
|
/ "data/raw/product_commerce/legacy/legacy-orders/2025/01/detail.json"
|
|
).exists()
|
|
|
|
|
|
def test_precheck_rejects_unplanned_files_in_dedicated_target_scope(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
data_root = tmp_path / "new"
|
|
unexpected = (
|
|
data_root
|
|
/ "data/raw/product_commerce/legacy/legacy-orders/unexpected.txt"
|
|
)
|
|
unexpected.parent.mkdir(parents=True)
|
|
unexpected.write_text("not in source", encoding="utf-8")
|
|
|
|
with pytest.raises(MigrationConflictError, match="unplanned target"):
|
|
_planner(data_root).create_plan([_source(source_root)])
|
|
|
|
|
|
def test_precheck_rejects_insufficient_disk_space(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
|
|
with pytest.raises(MigrationValidationError, match="disk space"):
|
|
_planner(tmp_path / "new", free_bytes=1).create_plan([_source(source_root)])
|
|
|
|
|
|
def test_source_and_destination_must_not_overlap(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
|
|
with pytest.raises(MigrationValidationError, match="overlap"):
|
|
_planner(source_root).create_plan([_source(source_root)])
|
|
|
|
|
|
def test_source_change_after_planning_stops_before_overwriting_destination(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
data_root = tmp_path / "new"
|
|
plan = _planner(data_root).create_plan([_source(source_root)])
|
|
changed = source_root / "2025/01/orders.csv"
|
|
changed.write_text("changed after planning", encoding="utf-8")
|
|
|
|
with pytest.raises(MigrationExecutionError, match="source changed"):
|
|
HistoricalDataMigrationExecutor().execute(plan, apply=True)
|
|
|
|
target = data_root / next(
|
|
entry.target_relative_path
|
|
for entry in plan.files
|
|
if entry.relative_path == "2025/01/orders.csv"
|
|
)
|
|
assert not target.exists()
|
|
|
|
|
|
def test_apply_persists_progress_and_can_resume_after_interruption(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
data_root = tmp_path / "new"
|
|
report_path = data_root / "evidence" / "progress.json"
|
|
plan = _planner(data_root).create_plan([_source(source_root)])
|
|
executor = HistoricalDataMigrationExecutor()
|
|
original_copy = executor._copy_file_atomic
|
|
calls = 0
|
|
|
|
def fail_on_second(
|
|
source: Path, destination: Path, expected_size: int, expected_sha256: str
|
|
) -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 2:
|
|
raise OSError("simulated interruption")
|
|
original_copy(source, destination, expected_size, expected_sha256)
|
|
|
|
monkeypatch.setattr(executor, "_copy_file_atomic", fail_on_second)
|
|
with pytest.raises(MigrationExecutionError, match="simulated interruption"):
|
|
executor.execute(plan, apply=True, report_path=report_path)
|
|
|
|
interrupted = json.loads(report_path.read_text(encoding="utf-8"))
|
|
assert interrupted["status"] == "interrupted"
|
|
assert interrupted["copied_count"] == 1
|
|
|
|
resumed = HistoricalDataMigrationExecutor().execute(
|
|
plan, apply=True, report_path=report_path
|
|
)
|
|
assert resumed.status == "reconciled"
|
|
assert resumed.copied_count == 1
|
|
assert resumed.skipped_count == 1
|
|
|
|
|
|
def test_corrupt_temporary_copy_is_never_committed_to_destination(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
(source_root / "one.txt").parent.mkdir(parents=True)
|
|
(source_root / "one.txt").write_text("original", encoding="utf-8")
|
|
data_root = tmp_path / "new"
|
|
plan = _planner(data_root).create_plan([_source(source_root)])
|
|
|
|
def corrupt_copy(_source: Path, temporary: Path) -> None:
|
|
temporary.write_text("corrupt", encoding="utf-8")
|
|
|
|
monkeypatch.setattr("gyxx_flow.migration.data.shutil.copy2", corrupt_copy)
|
|
with pytest.raises(MigrationExecutionError, match="temporary copy verification"):
|
|
HistoricalDataMigrationExecutor().execute(plan, apply=True)
|
|
|
|
destination = data_root / plan.files[0].target_relative_path
|
|
assert not destination.exists()
|
|
|
|
|
|
def test_unsafe_ids_and_symlink_escape_are_rejected(tmp_path: Path) -> None:
|
|
source_root = tmp_path / "legacy"
|
|
_source_tree(source_root)
|
|
|
|
with pytest.raises(MigrationValidationError, match="module"):
|
|
MigrationSource("source", "../escape", DataLayer.RAW, source_root)
|
|
|
|
outside = tmp_path / "outside.txt"
|
|
outside.write_text("outside", encoding="utf-8")
|
|
link = source_root / "escape.txt"
|
|
try:
|
|
link.symlink_to(outside)
|
|
except OSError as exc:
|
|
pytest.skip(f"symlinks are unavailable: {exc}")
|
|
|
|
with pytest.raises(MigrationValidationError, match="symbolic link"):
|
|
_planner(tmp_path / "new").create_plan([_source(source_root)])
|