Emma Thorpe
ece79515c0
Build and publish container / build (pull_request) Successful in 3m54s
The budget subtracted the prefix length plus two, on the assumption of a leading and a trailing slash. That is right for /Music and wrong for an empty prefix, where there is only one slash -- losing a character at the card root, which is exactly where the longest paths sit. Computed from the prefix as it will actually appear instead: /Music/ costs seven characters and gives a mirror-relative budget of 253, the root costs one and gives 259. Worth being exact about because the reverse error is worse. A checker comparing mirror-relative paths against the flat 260 passes everything between 253 and 260, and those are precisely the paths closest to the edge.
789 lines
25 KiB
Python
789 lines
25 KiB
Python
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import music_mirror
|
|
|
|
|
|
def run(source, mirror, *extra):
|
|
return music_mirror.main(["--source", str(source), "--mirror", str(mirror), *extra])
|
|
|
|
|
|
def test_parse_quality_accepts_vbr_and_cbr():
|
|
assert music_mirror.parse_quality("V0") == ["-q:a", "0"]
|
|
assert music_mirror.parse_quality("v2") == ["-q:a", "2"]
|
|
assert music_mirror.parse_quality("256") == ["-b:a", "256k"]
|
|
|
|
|
|
def test_parse_quality_rejects_nonsense():
|
|
with pytest.raises(ValueError):
|
|
music_mirror.parse_quality("best")
|
|
|
|
|
|
def test_parse_interval_units():
|
|
assert music_mirror.parse_interval("90") == 90
|
|
assert music_mirror.parse_interval("30m") == 1800
|
|
assert music_mirror.parse_interval("6h") == 21600
|
|
assert music_mirror.parse_interval("1d") == 86400
|
|
|
|
|
|
def test_encodes_and_preserves_layout_and_tags(tmp_path, make_flac, probe_tag):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Test Artist" / "Test Album" / "03 Song.flac")
|
|
|
|
assert run(source, mirror) == 0
|
|
|
|
output = mirror / "Test Artist" / "Test Album" / "03 Song.mp3"
|
|
assert output.is_file()
|
|
assert probe_tag(output, "title") == "Test Title"
|
|
assert probe_tag(output, "album") == "Test Album"
|
|
|
|
|
|
def test_output_is_mp3(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
run(source, mirror)
|
|
|
|
codec = subprocess.run(
|
|
[
|
|
"ffprobe",
|
|
"-v",
|
|
"error",
|
|
"-select_streams",
|
|
"a:0",
|
|
"-show_entries",
|
|
"stream=codec_name",
|
|
"-of",
|
|
"csv=p=0",
|
|
str(mirror / "a.mp3"),
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
assert codec == "mp3"
|
|
|
|
|
|
def test_second_pass_skips_unchanged_files(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
run(source, mirror)
|
|
first = (mirror / "a.mp3").stat().st_mtime_ns
|
|
|
|
run(source, mirror)
|
|
assert (mirror / "a.mp3").stat().st_mtime_ns == first
|
|
|
|
|
|
def test_changed_source_is_re_encoded(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
track = make_flac(source / "a.flac")
|
|
|
|
run(source, mirror)
|
|
before = (mirror / "a.mp3").stat().st_mtime
|
|
|
|
# A replaced file with a newer mtime is what a Lidarr quality upgrade
|
|
# looks like on disk.
|
|
later = time.time() + 120
|
|
os.utime(track, (later, later))
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "a.mp3").stat().st_mtime > before
|
|
|
|
|
|
def test_deleted_source_is_pruned(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "a.flac")
|
|
make_flac(source / "Album" / "b.flac")
|
|
|
|
run(source, mirror)
|
|
(source / "Album" / "b.flac").unlink()
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "Album" / "a.mp3").is_file()
|
|
assert not (mirror / "Album" / "b.mp3").exists()
|
|
|
|
|
|
def test_emptied_directory_is_removed(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Gone" / "a.flac")
|
|
|
|
run(source, mirror)
|
|
(source / "Gone" / "a.flac").unlink()
|
|
run(source, mirror)
|
|
|
|
assert not (mirror / "Gone").exists()
|
|
|
|
|
|
def test_no_prune_keeps_orphans(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
run(source, mirror)
|
|
(source / "a.flac").unlink()
|
|
run(source, mirror, "--no-prune")
|
|
|
|
assert (mirror / "a.mp3").is_file()
|
|
|
|
|
|
def test_existing_mp3_is_copied_not_re_encoded(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
flac = make_flac(source / "a.flac")
|
|
subprocess.run(
|
|
["ffmpeg", "-loglevel", "error", "-y", "-i", str(flac), str(source / "b.mp3")],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
flac.unlink()
|
|
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "b.mp3").read_bytes() == (source / "b.mp3").read_bytes()
|
|
|
|
|
|
def test_interrupted_copy_leaves_nothing_behind(tmp_path, make_flac, monkeypatch):
|
|
"""copy2 reproduces the source mtime, so a truncated copy left in the mirror
|
|
would be read as current by every later pass."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
flac = make_flac(source / "a.flac")
|
|
subprocess.run(
|
|
["ffmpeg", "-loglevel", "error", "-y", "-i", str(flac), str(source / "b.mp3")],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
flac.unlink()
|
|
|
|
def truncated(src, destination, **kwargs):
|
|
Path(destination).write_bytes(Path(src).read_bytes()[:64])
|
|
raise OSError("no space left on device")
|
|
|
|
monkeypatch.setattr(music_mirror.shutil, "copy2", truncated)
|
|
|
|
assert run(source, mirror) == 1
|
|
assert not (mirror / "b.mp3").exists()
|
|
assert list(mirror.rglob("*.part")) == []
|
|
|
|
|
|
def test_encoded_file_is_group_readable(tmp_path, make_flac, tight_umask):
|
|
"""mkstemp creates 0600 whatever the umask, so the bit has to be added."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "a.mp3").stat().st_mode & stat.S_IRGRP
|
|
|
|
|
|
def test_copied_file_is_group_readable(tmp_path, make_flac, tight_umask):
|
|
"""copy2 carries the source's mode across, and the source may be private."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
flac = make_flac(source / "a.flac")
|
|
subprocess.run(
|
|
["ffmpeg", "-loglevel", "error", "-y", "-i", str(flac), str(source / "b.mp3")],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
flac.unlink()
|
|
(source / "b.mp3").chmod(0o600)
|
|
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "b.mp3").stat().st_mode & stat.S_IRGRP
|
|
|
|
|
|
def test_mirror_directories_are_group_traversable(tmp_path, make_flac, tight_umask):
|
|
"""A readable file is unreachable if the group cannot enter its directory."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Artist" / "Album" / "a.flac")
|
|
|
|
run(source, mirror)
|
|
|
|
for directory in (mirror, mirror / "Artist", mirror / "Artist" / "Album"):
|
|
mode = directory.stat().st_mode
|
|
assert mode & stat.S_IRGRP, directory
|
|
assert mode & stat.S_IXGRP, directory
|
|
|
|
|
|
def test_mirror_directories_survive_an_owner_hostile_umask(
|
|
tmp_path, make_flac, owner_hostile_umask
|
|
):
|
|
"""A umask carrying 0400 otherwise builds a tree the run cannot read back."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Artist" / "Album" / "a.flac")
|
|
|
|
# Applied only now: the library already exists, and the umask under test is
|
|
# the one the container starts this run with.
|
|
owner_hostile_umask()
|
|
run(source, mirror)
|
|
|
|
for directory in (mirror, mirror / "Artist", mirror / "Artist" / "Album"):
|
|
mode = directory.stat().st_mode
|
|
assert mode & stat.S_IRUSR, directory
|
|
assert mode & stat.S_IXUSR, directory
|
|
assert mode & stat.S_IRGRP, directory
|
|
assert mode & stat.S_IXGRP, directory
|
|
|
|
|
|
def test_private_mirror_file_is_repaired_without_re_encoding(tmp_path, make_flac):
|
|
"""A mirror written by an older version has a correct mtime, so nothing
|
|
else in the pass would revisit it."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
run(source, mirror)
|
|
output = mirror / "a.mp3"
|
|
output.chmod(output.stat().st_mode & ~stat.S_IRGRP)
|
|
before = output.stat().st_mtime_ns
|
|
|
|
run(source, mirror)
|
|
|
|
assert output.stat().st_mode & stat.S_IRGRP
|
|
assert output.stat().st_mtime_ns == before
|
|
|
|
|
|
def test_dry_run_does_not_change_permissions(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
run(source, mirror)
|
|
output = mirror / "a.mp3"
|
|
output.chmod(output.stat().st_mode & ~stat.S_IRGRP)
|
|
|
|
run(source, mirror, "--dry-run")
|
|
|
|
assert not output.stat().st_mode & stat.S_IRGRP
|
|
|
|
|
|
def test_format_upgrade_replaces_rather_than_duplicating(tmp_path, make_flac):
|
|
"""Lidarr replacing an MP3 with a FLAC must not leave two mirror files."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
flac = make_flac(source / "Album" / "01 Song.flac")
|
|
subprocess.run(
|
|
["ffmpeg", "-loglevel", "error", "-y", "-i", str(flac), str(source / "Album" / "01 Song.mp3")],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
flac.unlink()
|
|
|
|
run(source, mirror)
|
|
assert sorted(p.name for p in (mirror / "Album").iterdir()) == ["01 Song.mp3"]
|
|
|
|
# The upgrade: the MP3 goes, a FLAC arrives at the same stem.
|
|
(source / "Album" / "01 Song.mp3").unlink()
|
|
make_flac(source / "Album" / "01 Song.flac", title="Upgraded")
|
|
run(source, mirror)
|
|
|
|
assert sorted(p.name for p in (mirror / "Album").iterdir()) == ["01 Song.mp3"]
|
|
|
|
|
|
def test_renamed_album_leaves_nothing_behind(tmp_path, make_flac):
|
|
"""A Lidarr rename is a delete plus an add; the old tree must not linger."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Artist" / "Album (2019)" / "01 Song.flac")
|
|
|
|
run(source, mirror)
|
|
(source / "Artist" / "Album (2019)").rename(source / "Artist" / "Album (2020)")
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "Artist" / "Album (2020)" / "01 Song.mp3").is_file()
|
|
assert not (mirror / "Artist" / "Album (2019)").exists()
|
|
|
|
|
|
def test_competing_sources_pick_the_lossless_one_and_stay_stable(tmp_path, make_flac, probe_tag):
|
|
"""Both a FLAC and an MP3 at one stem: the FLAC wins, and stays won."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac", title="From FLAC")
|
|
other = make_flac(tmp_path / "scratch" / "other.flac", title="From MP3")
|
|
subprocess.run(
|
|
["ffmpeg", "-loglevel", "error", "-y", "-i", str(other), str(source / "a.mp3")],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
# Age the loser well beyond the mtime tolerance, so "is it current?" gives
|
|
# a definite answer for it rather than one that depends on the clock.
|
|
older = time.time() - 3600
|
|
os.utime(source / "a.mp3", (older, older))
|
|
|
|
run(source, mirror)
|
|
assert probe_tag(mirror / "a.mp3", "title") == "From FLAC"
|
|
|
|
# The loser must not make the mirror look stale on the next pass, or every
|
|
# run would re-encode for ever.
|
|
first = (mirror / "a.mp3").stat().st_mtime_ns
|
|
run(source, mirror)
|
|
assert (mirror / "a.mp3").stat().st_mtime_ns == first
|
|
|
|
|
|
def test_uppercase_extension_is_not_pruned_and_re_encoded(tmp_path, make_flac):
|
|
"""A .FLAC source must not be treated as an orphan on the next pass."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
made = make_flac(source / "a.flac")
|
|
made.rename(source / "a.FLAC")
|
|
|
|
run(source, mirror)
|
|
first = (mirror / "a.mp3").stat().st_mtime_ns
|
|
|
|
run(source, mirror)
|
|
assert (mirror / "a.mp3").is_file()
|
|
assert (mirror / "a.mp3").stat().st_mtime_ns == first
|
|
|
|
|
|
def test_whole_library_deleted_empties_the_mirror(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "A" / "one.flac")
|
|
make_flac(source / "B" / "two.flac")
|
|
|
|
run(source, mirror)
|
|
shutil.rmtree(source / "A")
|
|
shutil.rmtree(source / "B")
|
|
run(source, mirror)
|
|
|
|
assert list(mirror.rglob("*.mp3")) == []
|
|
assert not (mirror / "A").exists()
|
|
assert not (mirror / "B").exists()
|
|
|
|
|
|
def test_dry_run_writes_nothing(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
assert run(source, mirror, "--dry-run") == 0
|
|
assert not (mirror / "a.mp3").exists()
|
|
|
|
|
|
def test_subdir_limits_the_pass_and_does_not_prune(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "One" / "a.flac")
|
|
make_flac(source / "Two" / "b.flac")
|
|
|
|
assert run(source, mirror, "--subdir", "One") == 0
|
|
|
|
assert (mirror / "One" / "a.mp3").is_file()
|
|
# Outside the requested directory: neither encoded nor treated as an orphan.
|
|
assert not (mirror / "Two").exists()
|
|
|
|
|
|
def test_external_cover_is_embedded(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "a.flac")
|
|
subprocess.run(
|
|
[
|
|
"ffmpeg",
|
|
"-loglevel",
|
|
"error",
|
|
"-y",
|
|
"-f",
|
|
"lavfi",
|
|
"-i",
|
|
"color=c=red:s=64x64:d=1",
|
|
"-frames:v",
|
|
"1",
|
|
str(source / "Album" / "cover.jpg"),
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
|
|
run(source, mirror)
|
|
|
|
streams = subprocess.run(
|
|
[
|
|
"ffprobe",
|
|
"-v",
|
|
"error",
|
|
"-select_streams",
|
|
"v",
|
|
"-show_entries",
|
|
"stream=codec_name",
|
|
"-of",
|
|
"csv=p=0",
|
|
str(mirror / "Album" / "a.mp3"),
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
assert streams # a picture stream is present
|
|
|
|
|
|
def test_mirror_inside_source_is_refused(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
make_flac(source / "a.flac")
|
|
assert run(source, source / "mp3") == 2
|
|
|
|
|
|
def test_missing_source_is_refused(tmp_path):
|
|
assert run(tmp_path / "nope", tmp_path / "dst") == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("name", "expected"),
|
|
[
|
|
("Kick Out the Epic Motherf**ker", "Kick Out the Epic Motherf__ker"),
|
|
("Where Are You?", "Where Are You_"),
|
|
("Song: Part 2", "Song_ Part 2"),
|
|
('"Heroes"', "_Heroes_"),
|
|
("trailing dot.", "trailing dot"),
|
|
("trailing space ", "trailing space"),
|
|
("...", "_"),
|
|
("Mötley Crüe", "Mötley Crüe"),
|
|
("perfectly ordinary", "perfectly ordinary"),
|
|
],
|
|
)
|
|
def test_fat32_safe_names(name, expected):
|
|
"""A name FAT32 will not take is a track that silently does not arrive."""
|
|
assert music_mirror.fat32_safe(name) == expected
|
|
|
|
|
|
def test_a_reserved_character_is_replaced_in_the_mirror_path(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Dada Life" / "Album" / "Kick Out the Epic Motherf**ker.flac")
|
|
|
|
run(source, mirror, "--fat32-safe")
|
|
|
|
assert (mirror / "Dada Life" / "Album" / "Kick Out the Epic Motherf__ker.mp3").is_file()
|
|
|
|
|
|
def test_without_the_flag_the_name_is_left_alone(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "Where Are You?.flac")
|
|
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "Album" / "Where Are You?.mp3").is_file()
|
|
|
|
|
|
def test_an_existing_mirror_is_renamed_not_re_encoded(tmp_path, make_flac):
|
|
"""Turning the flag on changes the path of every track holding a reserved
|
|
character. Re-encoding those would be hours of work to produce files that
|
|
already exist byte for byte."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "Where Are You?.flac")
|
|
|
|
run(source, mirror)
|
|
before = mirror / "Album" / "Where Are You?.mp3"
|
|
contents = before.read_bytes()
|
|
stamp = before.stat().st_mtime_ns
|
|
|
|
run(source, mirror, "--fat32-safe")
|
|
|
|
after = mirror / "Album" / "Where Are You_.mp3"
|
|
assert after.is_file()
|
|
assert not before.exists()
|
|
assert after.read_bytes() == contents, "it was re-encoded rather than moved"
|
|
assert after.stat().st_mtime_ns == stamp
|
|
|
|
|
|
def test_names_colliding_only_by_case_are_caught(tmp_path, make_flac):
|
|
"""Two files here, one file on FAT32. Better found now than as a silent
|
|
overwrite partway through the copy."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "Song.flac")
|
|
make_flac(source / "Album" / "SONG.flac")
|
|
|
|
run(source, mirror, "--fat32-safe")
|
|
|
|
written = sorted(p.name for p in (mirror / "Album").glob("*.mp3"))
|
|
assert len(written) == 1, written
|
|
|
|
|
|
def test_the_album_cover_is_copied_beside_the_tracks(tmp_path, make_flac, make_cover):
|
|
"""Rockbox looks for art on the filesystem; its search never touches the
|
|
picture embedded in the tag."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "a.flac")
|
|
make_cover(source / "Album" / "cover.jpg")
|
|
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "Album" / "cover.jpg").is_file()
|
|
|
|
|
|
def test_a_copied_cover_is_group_readable(tmp_path, make_flac, make_cover, tight_umask):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "a.flac")
|
|
make_cover(source / "Album" / "cover.jpg")
|
|
|
|
run(source, mirror)
|
|
|
|
assert (mirror / "Album" / "cover.jpg").stat().st_mode & stat.S_IRGRP
|
|
|
|
|
|
def test_a_cover_left_without_tracks_is_pruned(tmp_path, make_flac, make_cover):
|
|
"""Otherwise the directory never looks empty and never goes."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Gone" / "a.flac")
|
|
make_cover(source / "Gone" / "cover.jpg")
|
|
|
|
run(source, mirror)
|
|
assert (mirror / "Gone" / "cover.jpg").is_file()
|
|
|
|
shutil.rmtree(source / "Gone")
|
|
run(source, mirror)
|
|
|
|
assert not (mirror / "Gone").exists()
|
|
|
|
|
|
def test_a_dry_run_reports_a_rename_not_an_encode(tmp_path, make_flac, caplog):
|
|
"""The difference between moving a file and re-encoding it is the
|
|
difference between a minute and an afternoon, so a dry run must not
|
|
describe the first as the second."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "Where Are You?.flac")
|
|
run(source, mirror)
|
|
|
|
with caplog.at_level("INFO"):
|
|
run(source, mirror, "--fat32-safe", "--dry-run")
|
|
|
|
assert "would rename" in caplog.text
|
|
assert "would encode" not in caplog.text
|
|
|
|
|
|
def test_a_dry_run_does_not_call_the_old_paths_orphans(tmp_path, make_flac, caplog):
|
|
"""Nothing was renamed, so they are still there -- but they are the files a
|
|
real run would move, not files it would delete."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "Where Are You?.flac")
|
|
run(source, mirror)
|
|
|
|
with caplog.at_level("INFO"):
|
|
run(source, mirror, "--fat32-safe", "--dry-run")
|
|
|
|
assert "would remove orphan" not in caplog.text
|
|
|
|
|
|
def test_a_dry_run_moves_nothing(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "Where Are You?.flac")
|
|
run(source, mirror)
|
|
|
|
run(source, mirror, "--fat32-safe", "--dry-run")
|
|
|
|
assert (mirror / "Album" / "Where Are You?.mp3").is_file()
|
|
assert not (mirror / "Album" / "Where Are You_.mp3").exists()
|
|
|
|
|
|
def test_renames_are_counted_separately_from_encodes(tmp_path, make_flac, caplog):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / "Where Are You?.flac")
|
|
run(source, mirror)
|
|
|
|
with caplog.at_level("INFO"):
|
|
run(source, mirror, "--fat32-safe")
|
|
|
|
assert "1 renamed" in caplog.text
|
|
assert "0 encoded" in caplog.text
|
|
|
|
|
|
def test_a_long_path_is_shortened_from_the_deepest_component(tmp_path, make_flac):
|
|
"""The track name carries the least navigational value and the artist the
|
|
most, so the filename is sacrificed before the album."""
|
|
artist = "A" * 60
|
|
album = "B" * 60
|
|
title = "C" * 150
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / artist / album / f"{title}.flac")
|
|
|
|
run(source, mirror, "--fat32-safe", "--max-path", "160", "--device-prefix", "/Music")
|
|
|
|
written = list(mirror.rglob("*.mp3"))
|
|
assert len(written) == 1
|
|
relative = written[0].relative_to(mirror)
|
|
assert relative.parts[0] == artist, "the artist directory should be untouched"
|
|
assert relative.parts[1] == album, "the album directory should be untouched"
|
|
assert len(str(relative)) <= 160 - len("Music") - 2
|
|
|
|
|
|
def test_shortening_is_stable_across_passes(tmp_path, make_flac):
|
|
"""An unstable name would rename every file on every pass, for ever."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / ("D" * 80) / ("E" * 80) / f"{'F' * 120}.flac")
|
|
|
|
run(source, mirror, "--fat32-safe", "--max-path", "180")
|
|
first = sorted(str(p.relative_to(mirror)) for p in mirror.rglob("*.mp3"))
|
|
stamp = next(mirror.rglob("*.mp3")).stat().st_mtime_ns
|
|
|
|
run(source, mirror, "--fat32-safe", "--max-path", "180")
|
|
|
|
assert sorted(str(p.relative_to(mirror)) for p in mirror.rglob("*.mp3")) == first
|
|
assert next(mirror.rglob("*.mp3")).stat().st_mtime_ns == stamp
|
|
|
|
|
|
def test_two_long_names_do_not_collide_after_shortening(tmp_path, make_flac):
|
|
"""They share a prefix and cut to the same string; the hash is what keeps
|
|
them apart."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
shared = "G" * 140
|
|
make_flac(source / "Album" / f"{shared}one.flac")
|
|
make_flac(source / "Album" / f"{shared}two.flac")
|
|
|
|
run(source, mirror, "--fat32-safe", "--max-path", "120")
|
|
|
|
assert len(list(mirror.rglob("*.mp3"))) == 2
|
|
|
|
|
|
def test_a_sanitised_mirror_is_renamed_rather_than_re_encoded_when_shortening(
|
|
tmp_path, make_flac
|
|
):
|
|
"""The previous naming is sanitised-but-not-shortened, not the original."""
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Album" / f"Where Are You? {'H' * 140}.flac")
|
|
|
|
run(source, mirror, "--fat32-safe", "--max-path", "400")
|
|
before = next(mirror.rglob("*.mp3"))
|
|
contents = before.read_bytes()
|
|
|
|
run(source, mirror, "--fat32-safe", "--max-path", "120")
|
|
|
|
after = next(mirror.rglob("*.mp3"))
|
|
assert after != before
|
|
assert after.read_bytes() == contents, "it was re-encoded rather than moved"
|
|
|
|
|
|
def test_shortening_only_applies_when_over_budget(tmp_path, make_flac):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "Artist" / "Album" / "Short Name.flac")
|
|
|
|
run(source, mirror, "--fat32-safe")
|
|
|
|
assert (mirror / "Artist" / "Album" / "Short Name.mp3").is_file()
|
|
|
|
|
|
def test_a_path_that_cannot_be_made_to_fit_is_reported(tmp_path, make_flac, caplog):
|
|
"""Too deeply nested to shorten without making every component unreadable."""
|
|
deep = Path(*["I" * 20 for _ in range(10)])
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / deep / "track.flac")
|
|
|
|
with caplog.at_level("WARNING"):
|
|
run(source, mirror, "--fat32-safe", "--max-path", "80")
|
|
|
|
assert "too" in caplog.text and "nested" in caplog.text
|
|
|
|
|
|
# Lidarr writes "Artist - Album - NN - Title.mp3" inside a directory already
|
|
# named for that artist and album, so a long album title appears three times in
|
|
# one path. These are real paths from a real library.
|
|
GIZZARD_ALBUM = (
|
|
"PetroDragonic Apocalypse; or, Dawn of Eternal Night - An Annihilation of"
|
|
" Planet Earth and the Beginning of Merciless Damnation"
|
|
)
|
|
GIZZARD_TRACKS = (
|
|
"01 - Motor Spirit",
|
|
"02 - Supercell",
|
|
"03 - Converge",
|
|
"04 - Witchcraft",
|
|
"05 - Gila Monster",
|
|
"06 - Dragon",
|
|
"07 - Flamethrower",
|
|
)
|
|
|
|
|
|
def gizzard_path(track):
|
|
return Path(
|
|
"King Gizzard & the Lizard Wizard",
|
|
f"{GIZZARD_ALBUM} (2023)",
|
|
f"King Gizzard & the Lizard Wizard - {GIZZARD_ALBUM} - {track}.mp3",
|
|
)
|
|
|
|
|
|
def test_shortening_keeps_the_part_that_tells_tracks_apart():
|
|
"""Cutting from the end discards the track number and title, which is all
|
|
that distinguishes one track on the record from another."""
|
|
for track in GIZZARD_TRACKS:
|
|
fitted = music_mirror.fit_path(gizzard_path(track), 253)
|
|
assert len(str(fitted)) <= 253
|
|
assert fitted.name.endswith(f"{track}.mp3"), fitted.name
|
|
|
|
|
|
def test_every_track_on_a_long_album_keeps_a_distinct_name():
|
|
fitted = {music_mirror.fit_path(gizzard_path(t), 253).name for t in GIZZARD_TRACKS}
|
|
|
|
assert len(fitted) == len(GIZZARD_TRACKS)
|
|
|
|
|
|
def test_a_long_title_keeps_both_ends():
|
|
"""The Beatles' 'The Long One' is one track whose title is the long part."""
|
|
path = Path(
|
|
"The Beatles",
|
|
"Abbey Road (1969)",
|
|
"Digital Media 03",
|
|
"The Beatles - Abbey Road - 09 - The Long One - You Never Give Me Your Money"
|
|
" + Sun King + Mean Mr Mustard + Her Majesty + Polythene Pam + She Came In"
|
|
" Through the Bathroom Window+ Golden Slumbers + Carry That Weight + The End.mp3",
|
|
)
|
|
|
|
fitted = music_mirror.fit_path(path, 253)
|
|
|
|
assert len(str(fitted)) <= 253
|
|
assert fitted.name.startswith("The Beatles - Abbey Road - 09 - The Long One")
|
|
assert fitted.name.endswith("The End.mp3")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("prefix", "expected"),
|
|
[("/Music", 253), ("Music", 253), ("/Music/", 253), ("", 259), ("/", 259)],
|
|
)
|
|
def test_the_device_prefix_is_costed_exactly(prefix, expected):
|
|
"""A mirror-relative path of 253 characters becomes 260 on the device once
|
|
/Music/ is in front of it, which is the whole of the limit. Approximating
|
|
the prefix loses a character at the root, where the longest paths are."""
|
|
assert 260 - len(music_mirror.device_prefix_length(prefix)) == expected
|
|
|
|
|
|
def test_the_budget_is_reported_so_it_can_be_checked(tmp_path, make_flac, caplog):
|
|
source = tmp_path / "src"
|
|
mirror = tmp_path / "dst"
|
|
make_flac(source / "a.flac")
|
|
|
|
with caplog.at_level("INFO"):
|
|
run(source, mirror, "--fat32-safe")
|
|
|
|
assert "limited to 253 characters" in caplog.text
|