music-curator/tests/test_find_missing_tracks.py

import os
import plistlib
import sys
import unicodedata
import urllib.parse
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))

import find_missing_tracks  # noqa: E402


def write_library(path, tracks):
    """Write a Library.xml in the shape Music exports."""
    path.write_bytes(plistlib.dumps({"Tracks": {str(i): t for i, t in enumerate(tracks)}}))
    return path


def track(root, name, filename=None):
    """A local track entry. Omit filename for a streaming entry with no file."""
    entry = {"Name": name, "Artist": "An Artist", "Album": "An Album"}
    if filename is not None:
        entry["Location"] = "file://" + urllib.parse.quote(str(root / filename))
    return entry


def missing_from(library, roots):
    with open(library, "rb") as handle:
        loaded = plistlib.load(handle)
    present = find_missing_tracks.existing_under(roots)
    gone = []
    for entry in loaded["Tracks"].values():
        path = find_missing_tracks.location_of(entry)
        if path is not None and find_missing_tracks.key_for(path) not in present:
            gone.append(entry["Name"])
    return sorted(gone)


def test_a_deleted_file_is_reported(tmp_path):
    (tmp_path / "kept.mp3").write_bytes(b"x")
    library = write_library(
        tmp_path / "Library.xml",
        [track(tmp_path, "Kept", "kept.mp3"), track(tmp_path, "Gone", "gone.mp3")],
    )

    assert missing_from(library, [tmp_path]) == ["Gone"]


def test_a_decomposed_filename_is_not_reported_missing(tmp_path):
    """macOS stores filenames decomposed and most everything else composes them,
    so the umlauts in Motley Crue are two different byte strings depending on
    which side wrote the name."""
    composed = unicodedata.normalize("NFC", "Mötley Crüe.mp3")
    (tmp_path / composed).write_bytes(b"x")
    library = write_library(
        tmp_path / "Library.xml",
        [track(tmp_path, "Umlauts", unicodedata.normalize("NFD", "Mötley Crüe.mp3"))],
    )

    assert missing_from(library, [tmp_path]) == []


def test_a_case_difference_is_not_reported_missing(tmp_path):
    """The share is very likely case-insensitive, and a file is not missing
    because someone capitalised it differently."""
    (tmp_path / "Hells Bells.mp3").write_bytes(b"x")
    library = write_library(
        tmp_path / "Library.xml", [track(tmp_path, "Bells", "hells bells.MP3")]
    )

    assert missing_from(library, [tmp_path]) == []


def test_a_track_with_no_file_is_not_a_missing_file(tmp_path):
    """A streaming entry has never had a file to lose."""
    library = write_library(tmp_path / "Library.xml", [track(tmp_path, "Streamed", None)])

    assert missing_from(library, [tmp_path]) == []


def test_the_walk_finds_files_nested_below_the_root(tmp_path):
    nested = tmp_path / "Artist" / "Album"
    nested.mkdir(parents=True)
    (nested / "deep.mp3").write_bytes(b"x")

    found = find_missing_tracks.existing_under([tmp_path])

    assert find_missing_tracks.key_for(nested / "deep.mp3") in found


def test_the_root_is_derived_from_the_tracks(tmp_path):
    paths = [tmp_path / "a" / "one.mp3", tmp_path / "b" / "two.mp3"]

    assert find_missing_tracks.roots_of(paths, None) == [tmp_path]


def test_an_explicit_root_overrides_the_derived_one(tmp_path):
    """Worth using: tracks spread beyond the mirror can derive a common parent
    of / and send the walk across the whole disk."""
    paths = [tmp_path / "a" / "one.mp3"]

    assert find_missing_tracks.roots_of(paths, ["/Volumes/music-mp3"]) == [
        Path("/Volumes/music-mp3")
    ]


def test_the_whole_library_missing_is_called_out(tmp_path, capsys):
    """Almost always an unmounted share rather than an empty library."""
    library = write_library(
        tmp_path / "Library.xml", [track(tmp_path / "elsewhere", "Gone", "gone.mp3")]
    )

    find_missing_tracks.main([str(library), "--root", str(tmp_path)])

    assert "not mounted" in capsys.readouterr().err


def test_a_healthy_library_says_nothing_alarming(tmp_path, capsys):
    (tmp_path / "kept.mp3").write_bytes(b"x")
    library = write_library(tmp_path / "Library.xml", [track(tmp_path, "Kept", "kept.mp3")])

    find_missing_tracks.main([str(library), "--root", str(tmp_path)])

    captured = capsys.readouterr()
    assert "0 of 1 local tracks are missing" in captured.err
    assert "not mounted" not in captured.err
    assert captured.out == ""


@pytest.mark.parametrize(
    ("location", "expected"),
    [
        ("file:///music/a%20b.mp3", Path("/music/a b.mp3")),
        ("file:///music/plain.mp3", Path("/music/plain.mp3")),
        ("https://example.invalid/stream", None),
        ("", None),
    ],
)
def test_locations_are_decoded(location, expected):
    assert find_missing_tracks.location_of({"Location": location} if location else {}) == expected


def test_the_walk_costs_one_read_per_directory_not_one_per_file(tmp_path):
    """The whole point. Over SMB a per-file check is a round trip per track."""
    for album in range(20):
        directory = tmp_path / f"Album {album}"
        directory.mkdir()
        for index in range(25):
            (directory / f"{index}.mp3").write_bytes(b"x")

    reads = {"n": 0}
    real_walk = os.walk

    def counting_walk(*args, **kwargs):
        for entry in real_walk(*args, **kwargs):
            reads["n"] += 1
            yield entry

    find_missing_tracks.os.walk = counting_walk
    try:
        found = find_missing_tracks.existing_under([tmp_path])
    finally:
        find_missing_tracks.os.walk = real_walk

    assert len(found) == 500
    assert reads["n"] == 21  # the root and its twenty albums, not 500 files