2026-08-21 15:30:58 +01:00
|
|
|
import os
|
|
|
|
|
import shutil
|
2026-08-24 11:26:13 +01:00
|
|
|
import stat
|
2026-08-21 15:30:58 +01:00
|
|
|
import subprocess
|
|
|
|
|
import time
|
2026-08-24 11:33:23 +01:00
|
|
|
from pathlib import Path
|
2026-08-21 15:30:58 +01:00
|
|
|
|
|
|
|
|
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()
|
|
|
|
|
|
|
|
|
|
|
2026-08-24 11:33:23 +01:00
|
|
|
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")) == []
|
|
|
|
|
|
|
|
|
|
|
2026-08-24 11:26:13 +01:00
|
|
|
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
|
|
|
|
|
|
|
|
|
|
|
2026-08-24 13:21:07 +01:00
|
|
|
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
|
|
|
|
|
|
|
|
|
|
|
2026-08-24 11:26:13 +01:00
|
|
|
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
|
|
|
|
|
|
|
|
|
|
|
2026-08-21 15:30:58 +01:00
|
|
|
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
|
2026-08-25 10:29:55 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
@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()
|
2026-08-25 11:26:24 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
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
|
2026-08-25 11:45:27 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
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
|
2026-08-25 11:50:42 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
# 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")
|
2026-08-25 11:52:03 +01:00
|
|
|
|
|
|
|
|
|
|
|
|
|
@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
|