Emma Thorpe
e9852e6c86
Build and publish container / build (pull_request) Successful in 15m16s
The umask handling added for group access cleared only the group bits and left owner and other to the environment. A container whose umask carries 0400 then produces mirror directories of mode 0300: writable and enterable, unreadable to the very run that created them, and unreadable to anything serving the share. Clear the owner read and execute bits from the umask as well. The `other` bits stay where the environment puts them, because whether the mirror is world-readable is a real policy question; being able to read a directory the process itself just created is not. Files were never exposed to this: mkstemp sets 0600 outright and copy2 takes the source file's mode, both ignoring the umask.
446 lines
13 KiB
Python
446 lines
13 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
|