Files
music-mirror/tests/test_music_mirror.py
T
Emma Thorpe 3f50577de6
Build and publish container / build (pull_request) Successful in 1m38s
fix: report renames honestly in a dry run
--dry-run described a FAT32 rename as an encode. process() skipped the rename
whenever dry_run was set, then found no file at the target and fell through to
the encode path, so a preview of enabling --fat32-safe announced a full
re-encode of every track whose name held a reserved character. The real run
moves those files in a moment. A preview that inverts the cost of the thing
being previewed is worse than no preview at all.

Prune compounded it. With nothing renamed, the pre-sanitisation files are still
on disk, and they were reported as orphans due for deletion -- so the same dry
run claimed the library would be re-encoded and the originals thrown away,
neither of which is true.

Renames are now their own outcome: reported as "would rename" in a dry run,
counted separately from encodes in the pass summary, and excluded from the
orphan list when a dry run leaves them in place.
2026-08-25 11:34:42 +01:00

616 lines
19 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