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 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_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