music-mirror/tests/test_music_mirror.py

import os
import subprocess
import time

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