Add batch duplicate review workflow
This commit is contained in:
@@ -5,6 +5,7 @@ import pytest
|
||||
from serato_doctor.repair import (
|
||||
BACKUP_FOLDER,
|
||||
apply_duplicate_repair,
|
||||
apply_duplicate_repair_batch,
|
||||
list_backups,
|
||||
plan_duplicate_repair,
|
||||
restore_backup,
|
||||
@@ -72,3 +73,29 @@ def test_backup_rotation_can_be_limited_or_unlimited(tmp_path):
|
||||
assert len(list(root.iterdir())) == 3
|
||||
rotate_backups(root, 2)
|
||||
assert {path.name for path in root.iterdir()} == {"002", "003"}
|
||||
|
||||
|
||||
def test_batch_repair_uses_one_backup_and_restores_every_group(tmp_path):
|
||||
serato, first_keeper, first_duplicate = library(tmp_path)
|
||||
second_keeper = tmp_path / "Music" / "Main" / "Other.mp3"
|
||||
second_duplicate = tmp_path / "Music" / "Old" / "Other.mp3"
|
||||
second_keeper.write_bytes(b"other-keeper")
|
||||
second_duplicate.write_bytes(b"other-duplicate")
|
||||
plans = (
|
||||
plan_duplicate_repair(
|
||||
first_keeper, (first_keeper, first_duplicate), serato
|
||||
),
|
||||
plan_duplicate_repair(
|
||||
second_keeper, (second_keeper, second_duplicate), serato
|
||||
),
|
||||
)
|
||||
|
||||
receipt = apply_duplicate_repair_batch(plans, serato, backup_limit=10)
|
||||
|
||||
assert first_duplicate.is_symlink()
|
||||
assert second_duplicate.is_symlink()
|
||||
assert len(list((serato / BACKUP_FOLDER).iterdir())) == 1
|
||||
assert len(receipt.replaced) == 2
|
||||
restore_backup(receipt.backup)
|
||||
assert first_duplicate.read_bytes() == b"duplicate"
|
||||
assert second_duplicate.read_bytes() == b"other-duplicate"
|
||||
|
||||
@@ -9,6 +9,7 @@ from serato_doctor.web import (
|
||||
_verified_audio,
|
||||
analyze_paths,
|
||||
duplicate_repair,
|
||||
duplicate_repair_batch,
|
||||
)
|
||||
|
||||
|
||||
@@ -113,3 +114,29 @@ def test_duplicate_details_include_preview_and_finder_controls(tmp_path):
|
||||
assert len(group["file_previews"]) == 2
|
||||
assert group["file_previews"][0]["audio_url"].startswith("/api/audio?")
|
||||
assert group["file_previews"][0]["reveal_token"]
|
||||
|
||||
|
||||
def test_batch_preview_combines_approved_groups_without_changes(tmp_path):
|
||||
serato = tmp_path / "_Serato_"
|
||||
serato.mkdir()
|
||||
music = tmp_path / "Music"
|
||||
choices = []
|
||||
for filename in ("First.mp3", "Second.mp3"):
|
||||
keeper = music / "A" / filename
|
||||
duplicate = music / "B" / filename
|
||||
keeper.parent.mkdir(parents=True, exist_ok=True)
|
||||
duplicate.parent.mkdir(parents=True, exist_ok=True)
|
||||
keeper.write_bytes(b"keeper")
|
||||
duplicate.write_bytes(b"duplicate")
|
||||
choices.append(
|
||||
{"keeper": str(keeper), "group_files": [str(keeper), str(duplicate)]}
|
||||
)
|
||||
|
||||
result = duplicate_repair_batch(
|
||||
serato, music, choices, backup_limit=10
|
||||
)
|
||||
|
||||
assert result["applied"] is False
|
||||
assert result["choice_count"] == 2
|
||||
assert len(result["replaced"]) == 2
|
||||
assert all(not Path(choice["group_files"][1]).is_symlink() for choice in choices)
|
||||
|
||||
Reference in New Issue
Block a user