music-mirror/tests/test_music_mirror.py

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 audio_frames(path):
    """Return a hash of a file's audio frames, ignoring its tags."""
    return subprocess.run(
        ["ffmpeg", "-v", "error", "-i", str(path), "-map", "0:a", "-c", "copy", "-f", "md5", "-"],
        check=True,
        capture_output=True,
        text=True,
    ).stdout.strip()


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",
            # Not csv: a levelled file carries ReplayGain side data, which the
            # csv writer renders as a trailing empty field.
            "-of",
            "default=noprint_wrappers=1:nokey=1",
            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):
    """Compared by the audio frames rather than the whole file: the copy is
    tagged with its ReplayGain values afterwards, so the two differ in the
    container while carrying identical audio."""
    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 audio_frames(mirror / "b.mp3") == audio_frames(source / "b.mp3")


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


# Rockbox applies the offset a ReplayGain tag carries but never measures
# loudness itself, so an untagged mirror plays each album at whatever level it
# was mastered to. The tags have to be written here or nowhere.


def test_replaygain_tags_are_written(tmp_path, make_flac, probe_tag, require_rsgain):
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")

    run(source, mirror)

    track = mirror / "Album" / "a.mp3"
    assert probe_tag(track, "REPLAYGAIN_TRACK_GAIN").endswith("dB")
    assert probe_tag(track, "REPLAYGAIN_ALBUM_GAIN").endswith("dB")
    assert probe_tag(track, "REPLAYGAIN_TRACK_PEAK")


def test_the_album_shares_one_gain_and_the_tracks_keep_their_own(
    tmp_path, make_flac, probe_tag, require_rsgain
):
    """Album gain is what keeps a quiet track quiet within a record it belongs
    to. Track gain is written alongside it so the device can pick the other
    behaviour when shuffling."""
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "loud.flac")
    make_flac(source / "Album" / "quiet.flac", gain=-12)

    run(source, mirror)

    loud = mirror / "Album" / "loud.mp3"
    quiet = mirror / "Album" / "quiet.mp3"
    assert probe_tag(loud, "REPLAYGAIN_ALBUM_GAIN") == probe_tag(quiet, "REPLAYGAIN_ALBUM_GAIN")
    assert probe_tag(loud, "REPLAYGAIN_TRACK_GAIN") != probe_tag(quiet, "REPLAYGAIN_TRACK_GAIN")


def test_levelling_does_not_make_the_next_pass_re_encode(
    tmp_path, make_flac, caplog, require_rsgain
):
    """Writing tags rewrites the file, and staleness here is an mtime
    comparison. Without --preserve-mtimes every pass would re-encode the whole
    library and then level it again, forever."""
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")

    run(source, mirror)
    before = (mirror / "Album" / "a.mp3").stat().st_mtime

    with caplog.at_level("INFO"):
        run(source, mirror)

    assert "0 encoded" in caplog.text
    assert (mirror / "Album" / "a.mp3").stat().st_mtime == before


def test_an_already_levelled_album_is_not_measured_again(
    tmp_path, make_flac, caplog, require_rsgain
):
    """Measuring costs a decode of every track. A pass that changed nothing
    must not pay it."""
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")
    run(source, mirror)

    with caplog.at_level("INFO"):
        run(source, mirror)

    assert "0 levelled" in caplog.text


def test_a_new_track_relevels_the_album_around_it(
    tmp_path, make_flac, probe_tag, require_rsgain
):
    """Album gain is a property of the whole album, so a track arriving late
    makes the value stored on every one of its siblings wrong.

    -10 dB rather than something more dramatic: R128 gates quiet passages out
    of the measurement, and a track far enough below the rest of the record is
    excluded from it entirely."""
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")
    run(source, mirror)
    first = probe_tag(mirror / "Album" / "a.mp3", "REPLAYGAIN_ALBUM_GAIN")

    make_flac(source / "Album" / "b.flac", gain=-10)
    run(source, mirror)

    assert probe_tag(mirror / "Album" / "a.mp3", "REPLAYGAIN_ALBUM_GAIN") != first


def test_a_removed_track_relevels_the_album_behind_it(
    tmp_path, make_flac, caplog, require_rsgain
):
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")
    make_flac(source / "Album" / "b.flac", gain=-20)
    run(source, mirror)

    (source / "Album" / "b.flac").unlink()
    with caplog.at_level("INFO"):
        run(source, mirror)

    assert "1 levelled" in caplog.text


def test_embedded_cover_art_does_not_hide_the_tags(
    tmp_path, make_flac, make_cover, require_rsgain
):
    """The check walks ID3v2 frame headers and seeks over the bodies. Cover art
    sits between the text frames and the ones rsgain appends, so a check that
    only read the start of the tag would never reach them -- and would measure
    every album with a cover on every pass."""
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")
    make_cover(source / "Album" / "cover.jpg")

    run(source, mirror)

    assert music_mirror.has_replaygain(mirror / "Album" / "a.mp3")


def test_no_replaygain_leaves_the_tags_off(tmp_path, make_flac, probe_tag):
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")

    run(source, mirror, "--no-replaygain")

    assert not probe_tag(mirror / "Album" / "a.mp3", "REPLAYGAIN_ALBUM_GAIN")
    assert not music_mirror.has_replaygain(mirror / "Album" / "a.mp3")


def test_an_unlevelled_mirror_is_backfilled(tmp_path, make_flac, probe_tag, require_rsgain):
    """A mirror built before any of this existed has correct mtimes, so no pass
    would ever revisit those files on its own."""
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")
    run(source, mirror, "--no-replaygain")

    run(source, mirror)

    assert probe_tag(mirror / "Album" / "a.mp3", "REPLAYGAIN_ALBUM_GAIN").endswith("dB")


def test_a_missing_scanner_is_reported_and_not_fatal(tmp_path, make_flac, caplog, monkeypatch):
    """The mirror is still correct audio in the right place. It just plays at
    the level it was mastered to."""
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")
    monkeypatch.setattr(music_mirror.shutil, "which", lambda name: None)

    with caplog.at_level("INFO"):
        assert run(source, mirror) == 0

    assert "rsgain is not on PATH" in caplog.text
    assert (mirror / "Album" / "a.mp3").is_file()


def test_a_dry_run_measures_nothing(tmp_path, make_flac, caplog, require_rsgain):
    source = tmp_path / "src"
    mirror = tmp_path / "dst"
    make_flac(source / "Album" / "a.flac")

    with caplog.at_level("INFO"):
        run(source, mirror, "--dry-run")

    assert "would level 1 album" in caplog.text
    assert not mirror.joinpath("Album").exists()