129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
|
"""A ReplayGain re-level rewrites a track's tags without changing its size or
|
||
|
|
its mtime, which is precisely the pair rsync's quick check compares. These
|
||
|
|
cover the list that is fed back to rsync to copy those tracks anyway."""
|
||
|
|
|
||
|
|
import io
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
|
||
|
|
|
||
|
|
import touched_albums # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
def album(root, name, *tracks):
|
||
|
|
directory = root / name
|
||
|
|
directory.mkdir(parents=True, exist_ok=True)
|
||
|
|
for track in tracks:
|
||
|
|
(directory / track).write_bytes(b"x")
|
||
|
|
return directory
|
||
|
|
|
||
|
|
|
||
|
|
def listing(mirror, *lines):
|
||
|
|
return touched_albums.remaining(mirror, *touched_albums.touched(list(lines)))
|
||
|
|
|
||
|
|
|
||
|
|
def test_the_siblings_of_a_new_track_are_listed(tmp_path):
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3", "03.mp3")
|
||
|
|
|
||
|
|
assert listing(tmp_path, "4096 Artist/Album/03.mp3") == [
|
||
|
|
"Artist/Album/01.mp3",
|
||
|
|
"Artist/Album/02.mp3",
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_a_track_rsync_just_copied_is_not_copied_twice(tmp_path):
|
||
|
|
"""A quality upgrade replaces every track on the record. Listing them again
|
||
|
|
would send the album across twice."""
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
||
|
|
|
||
|
|
assert listing(tmp_path, "4096 Artist/Album/01.mp3", "4096 Artist/Album/02.mp3") == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_a_deletion_relevels_what_is_left(tmp_path):
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
||
|
|
|
||
|
|
assert listing(tmp_path, "deleting Artist/Album/03.mp3") == [
|
||
|
|
"Artist/Album/01.mp3",
|
||
|
|
"Artist/Album/02.mp3",
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_an_album_deleted_outright_lists_nothing(tmp_path):
|
||
|
|
assert listing(tmp_path, "deleting Artist/Gone/01.mp3") == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_untouched_albums_are_left_alone(tmp_path):
|
||
|
|
album(tmp_path, "Artist/Changed", "01.mp3", "02.mp3")
|
||
|
|
album(tmp_path, "Artist/Quiet", "01.mp3", "02.mp3")
|
||
|
|
|
||
|
|
assert listing(tmp_path, "4096 Artist/Changed/01.mp3") == ["Artist/Changed/02.mp3"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_a_replaced_cover_is_not_an_album_change(tmp_path):
|
||
|
|
"""Only a track can change an album's gains, and covers are replaced often
|
||
|
|
enough that treating one as a re-level would copy records for nothing."""
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
||
|
|
|
||
|
|
assert listing(tmp_path, "17408 Artist/Album/cover.jpg") == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_directories_are_not_mistaken_for_tracks(tmp_path):
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3")
|
||
|
|
|
||
|
|
assert listing(tmp_path, "4096 Artist/Album/", "deleting Artist/Old/") == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_rsync_talking_to_the_operator_is_not_a_path(tmp_path):
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
||
|
|
|
||
|
|
assert (
|
||
|
|
listing(
|
||
|
|
tmp_path,
|
||
|
|
"sending incremental file list",
|
||
|
|
"",
|
||
|
|
"sent 1,234 bytes received 56 bytes 2,580.00 bytes/sec",
|
||
|
|
"total size is 7,890 speedup is 6.12",
|
||
|
|
)
|
||
|
|
== []
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_a_track_at_the_mirror_root_does_not_pull_in_the_whole_tree(tmp_path):
|
||
|
|
"""Nothing writes a mirror this way, but the directory of a root-level file
|
||
|
|
is the root, and recursing from there would be the whole library."""
|
||
|
|
(tmp_path / "loose.mp3").write_bytes(b"x")
|
||
|
|
(tmp_path / "other.mp3").write_bytes(b"x")
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3")
|
||
|
|
|
||
|
|
assert listing(tmp_path, "4096 loose.mp3") == ["other.mp3"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_the_paths_are_written_one_per_line(tmp_path):
|
||
|
|
"""They are fed straight back to rsync as --files-from."""
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
||
|
|
out = io.StringIO()
|
||
|
|
|
||
|
|
touched_albums.main(
|
||
|
|
["--mirror", str(tmp_path)],
|
||
|
|
stream=["4096 Artist/Album/01.mp3"],
|
||
|
|
out=out,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert out.getvalue() == "Artist/Album/02.mp3\n"
|
||
|
|
|
||
|
|
|
||
|
|
def test_it_runs_as_a_script(tmp_path):
|
||
|
|
album(tmp_path, "Artist/Album", "01.mp3", "02.mp3")
|
||
|
|
|
||
|
|
completed = subprocess.run(
|
||
|
|
[sys.executable, touched_albums.__file__, "--mirror", str(tmp_path)],
|
||
|
|
input="4096 Artist/Album/01.mp3\n",
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert completed.returncode == 0
|
||
|
|
assert completed.stdout == "Artist/Album/02.mp3\n"
|