music-mirror/tests/test_sync_to_ipod.py

"""The guards on sync-to-ipod.sh, which are the substance of the script.

rsync --delete is being aimed at a whole filesystem, so every refusal here is
protecting against emptying the wrong directory -- a mistake that does not
announce itself.
"""

import os
import shutil
import subprocess
from pathlib import Path

import pytest

SCRIPT = Path(__file__).resolve().parent.parent / "tools" / "sync-to-ipod.sh"

# Skipped rather than failed where the tools are absent: this is a host-side
# script, and a machine without rsync is not a machine that would run it.
REQUIRED = ("bash", "rsync", "findmnt")
pytestmark = pytest.mark.skipif(
    not all(shutil.which(tool) for tool in REQUIRED),
    reason=f"needs {', '.join(REQUIRED)} on PATH",
)


def run(*arguments):
    return subprocess.run(
        ["bash", str(SCRIPT), *arguments], capture_output=True, text=True
    )


@pytest.fixture
def mirror(tmp_path):
    source = tmp_path / "mirror"
    (source / "Album").mkdir(parents=True)
    (source / "Album" / "track.mp3").write_bytes(b"x")
    return source


def test_the_host_root_is_refused(mirror):
    """Stripping the trailing slash from "/" leaves an empty string, and an
    earlier version then reported it as "not a directory" instead."""
    result = run(str(mirror), "/")

    assert result.returncode == 1
    assert "refusing to sync onto /" in result.stderr


def test_an_empty_mirror_is_refused(tmp_path):
    """Mirroring nothing onto the device would delete everything on it."""
    empty = tmp_path / "empty"
    empty.mkdir()
    destination = tmp_path / "dest"
    destination.mkdir()

    result = run(str(empty), str(destination))

    assert result.returncode == 1
    assert "refusing to mirror nothing" in result.stderr


def test_syncing_a_directory_onto_itself_is_refused(mirror):
    result = run(str(mirror), str(mirror))

    assert result.returncode == 1
    assert "same directory" in result.stderr


def test_a_non_fat_destination_is_refused(mirror, tmp_path):
    """Which is also how an unmounted device is caught: /media/IPOD/Music then
    resolves to the host's own root filesystem."""
    destination = tmp_path / "dest"
    destination.mkdir()

    result = run(str(mirror), str(destination))

    assert result.returncode == 1
    assert "not FAT" in result.stderr
    assert "Is the device mounted?" in result.stderr


def test_a_missing_destination_is_refused(mirror, tmp_path):
    result = run(str(mirror), str(tmp_path / "nowhere"))

    assert result.returncode == 1
    assert "not a directory" in result.stderr


def test_a_subdirectory_of_the_device_is_a_valid_target(mirror, tmp_path):
    """The better target, in fact: --delete is confined to it."""
    destination = tmp_path / "dest" / "Music"
    destination.mkdir(parents=True)

    result = run("-f", "-n", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert "dry run, nothing was written" in result.stderr


def test_the_device_prefix_is_derived_from_the_destination(mirror, tmp_path):
    """Derived rather than configured, so it cannot disagree with where the
    files are actually going -- and the device's path limit applies to it."""
    destination = tmp_path / "dest" / "Music"
    destination.mkdir(parents=True)

    result = run("-f", "-n", str(mirror), str(destination))

    assert "the device will see this as /" in result.stderr


def test_a_dry_run_writes_nothing(mirror, tmp_path):
    destination = tmp_path / "dest"
    destination.mkdir()

    run("-f", "-n", str(mirror), str(destination))

    assert list(destination.iterdir()) == []


def test_rockbox_is_never_deleted(mirror, tmp_path):
    """A sync to the card root would otherwise remove the Rockbox install,
    since the mirror does not contain it."""
    destination = tmp_path / "dest"
    destination.mkdir()
    (destination / ".rockbox").mkdir()
    (destination / ".rockbox" / "rockbox.ipod").write_bytes(b"firmware")
    (destination / ".scrobbler.log").write_bytes(b"#AUDIOSCROBBLER/1.1\n")
    (destination / "Stale.mp3").write_bytes(b"old")

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert (destination / ".rockbox" / "rockbox.ipod").is_file()
    assert (destination / ".scrobbler.log").is_file()
    # But a track whose source has gone is still removed. That is the point.
    assert not (destination / "Stale.mp3").exists()
    assert (destination / "Album" / "track.mp3").is_file()


def test_help_goes_to_stdout_and_exits_clean():
    """Asking for help is not an error; getting the arguments wrong is."""
    result = run("--help")

    assert result.returncode == 0
    assert result.stdout.startswith("usage:")
    assert result.stderr == ""


def test_short_help_behaves_the_same():
    result = run("-h")

    assert result.returncode == 0
    assert result.stdout.startswith("usage:")


def test_misuse_goes_to_stderr_and_does_not():
    result = run("only-one-argument")

    assert result.returncode == 2
    assert result.stderr.startswith("usage:")
    assert result.stdout == ""


def test_the_help_explains_what_the_destination_should_be():
    """The question this script actually gets asked."""
    help_text = run("--help").stdout

    assert "/media/IPOD/Music" in help_text
    assert "artist folders" in help_text
    assert ".rockbox" in help_text


def test_the_help_says_how_to_reach_and_leave_disk_mode():
    help_text = run("--help").stdout

    assert "Menu+Select" in help_text
    assert "holding Play" in help_text


def test_counting_is_off_by_default(mirror, tmp_path):
    """The counting pass walks and compares both trees exactly as the transfer
    does. On a FAT card of fifty thousand files that costs more than moving the
    data, so the percentage has to be asked for."""
    destination = tmp_path / "dest"
    destination.mkdir()

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert "files to copy" not in result.stderr
    assert (destination / "Album" / "track.mp3").is_file()


def test_the_delta_algorithm_is_disabled(mirror, tmp_path):
    """It would read every destination file back over USB to checksum it, to
    avoid resending an MP3 that has changed in its entirety anyway."""
    script = SCRIPT.read_text()

    assert "--whole-file" in script


def test_directory_timestamps_are_not_set(mirror, tmp_path):
    """One setattr round trip per directory, across six thousand albums, to set
    timestamps nothing reads."""
    script = SCRIPT.read_text()

    assert "--omit-dir-times" in script


def test_an_interrupt_is_trapped_so_the_filesystem_is_flushed():
    """Ctrl-C during a transfer would otherwise skip the sync and the unmount,
    leaving a journal-less FAT filesystem with dirty buffers -- which is the
    corruption this script exists to prevent.

    Structural rather than timed: reproducing a mid-transfer signal needs a
    payload large enough to be slow, and a test that depends on losing a race
    is a test that fails in CI for no reason.
    """
    script = SCRIPT.read_text()

    assert "trap interrupted INT TERM" in script
    assert "exit 130" in script


def test_the_flush_and_unmount_happen_on_every_exit_path():
    script = SCRIPT.read_text()

    # Both the normal path and the interrupt path go through the same function,
    # so one cannot drift from the other.
    assert script.count("finish\n") >= 2
    assert "--partial" not in script, "rsync must delete partial files, not keep them"


def test_the_database_step_is_skipped_without_a_tool(mirror, tmp_path, monkeypatch):
    """Opt-in, like the scrobbler: absent configuration is not an error."""
    destination = tmp_path / "dest"
    destination.mkdir()
    monkeypatch.delenv("MUSIC_MIRROR_DATABASE_TOOL", raising=False)

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert "no database tool configured" in result.stderr


def test_a_missing_database_tool_is_refused(mirror, tmp_path, monkeypatch):
    destination = tmp_path / "dest"
    destination.mkdir()
    monkeypatch.setenv("MUSIC_MIRROR_DATABASE_TOOL", str(tmp_path / "nonexistent"))

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 1
    assert "not executable" in result.stderr


def test_the_database_step_can_be_skipped(mirror, tmp_path, monkeypatch):
    destination = tmp_path / "dest"
    destination.mkdir()
    monkeypatch.setenv("MUSIC_MIRROR_DATABASE_TOOL", str(tmp_path / "nonexistent"))

    result = run("-f", "-S", "-U", "-B", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert "not executable" not in result.stderr


def test_the_scan_reads_from_the_mirror_not_the_device():
    """The whole point: tags come off the mirror, only the .tcd files go over
    USB. Reading 49,600 files through an iPod's USB bridge is the slow path."""
    script = SCRIPT.read_text()

    assert 'ln -s "$mirror"' in script
    assert 'cd "$scratch"' in script


def can_bind_mount():
    """User namespaces let an unprivileged process bind mount. Not everywhere,
    notably not inside some containers, so the test that needs it skips."""
    return (
        subprocess.run(
            ["unshare", "-Umr", "true"], capture_output=True, check=False
        ).returncode
        == 0
    )


@pytest.mark.skipif(not can_bind_mount(), reason="needs unprivileged user namespaces")
def test_the_database_lands_at_the_device_root_not_the_music_folder(tmp_path):
    """The two tools disagree about where the root is. rsync copies artist
    folders into <device>/Music; the database tool must run one level up, where
    .rockbox lives, and must record /Music/... paths while reading the bytes
    from the mirror. device_prefix is what reconciles them.
    """
    mirror = tmp_path / "mirror" / "Pendulum" / "Immersion"
    mirror.mkdir(parents=True)
    (mirror / "01.mp3").write_bytes(b"not really an mp3")
    card = tmp_path / "card"
    (card / ".rockbox").mkdir(parents=True)
    (card / "Music").mkdir()
    device = tmp_path / "device"
    device.mkdir()

    tool = tmp_path / "fake-database"
    # Records where it was run and what it could see, which is the whole
    # question; producing a real database needs Rockbox's builder.
    tool.write_text(
        "#!/bin/sh\n"
        "printf '%s\\n' \"$PWD\" > .rockbox/where.txt\n"
        "ls Music/ > .rockbox/saw.txt\n"
        "echo db > .rockbox/database_0.tcd\n"
    )
    tool.chmod(0o755)

    script = (
        f"mount --bind {card} {device} && "
        f"XDG_CACHE_HOME={tmp_path / 'cache'} MUSIC_MIRROR_DATABASE_TOOL={tool} "
        f"bash {SCRIPT} -f -S -U {tmp_path / 'mirror'} {device / 'Music'}"
    )
    result = subprocess.run(
        ["unshare", "-Umr", "sh", "-c", script], capture_output=True, text=True
    )

    assert result.returncode == 0, result.stderr
    # The database lands beside the device root, not inside Music.
    assert (card / ".rockbox" / "database_0.tcd").is_file()
    # Only *.tcd is copied across, so the markers stay in the scratch root --
    # which is itself the point: nothing else is written to the device.
    scratch = tmp_path / "cache" / "music-mirror" / "database" / ".rockbox"
    assert not (card / ".rockbox" / "where.txt").exists()

    # It ran in the scratch root, not on the card.
    where = (scratch / "where.txt").read_text().strip()
    assert where.endswith("music-mirror/database"), where
    # ...and could walk into the mirror through a symlink named for the device
    # prefix, which is how the paths come out as /Music/... while the bytes are
    # read from somewhere else entirely.
    assert "Pendulum" in (scratch / "saw.txt").read_text()


def test_counting_can_be_asked_for(mirror, tmp_path):
    """When the destination is cheap to traverse, the percentage is worth it."""
    destination = tmp_path / "dest"
    destination.mkdir()

    result = run("-f", "-S", "-U", "-P", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert "files to copy" in result.stderr


# A ReplayGain re-level rewrites a track's tags in the padding the previous
# write left behind, so neither the size nor the mtime changes -- and those are
# the two things rsync's quick check compares.


def stale_copy(mirror, destination, relative, current, previous):
    """Put a file on the device that differs only in content from the mirror's."""
    source = mirror / relative
    source.parent.mkdir(parents=True, exist_ok=True)
    source.write_bytes(current)
    device = destination / relative
    device.parent.mkdir(parents=True, exist_ok=True)
    device.write_bytes(previous)
    os.utime(device, (source.stat().st_atime, source.stat().st_mtime))
    return device


def test_a_tag_only_change_reaches_the_device_with_the_track_that_caused_it(
    mirror, tmp_path
):
    """The new track is visible to rsync; its re-levelled sibling is not, and
    would otherwise keep the old album gain on the device forever."""
    destination = tmp_path / "dest"
    destination.mkdir()
    sibling = stale_copy(mirror, destination, "Album/sibling.mp3", b"NEW", b"OLD")

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert sibling.read_bytes() == b"NEW"
    assert (destination / "Album" / "track.mp3").is_file()


def test_a_deletion_also_relevels_what_is_left_behind(mirror, tmp_path):
    destination = tmp_path / "dest"
    destination.mkdir()
    sibling = stale_copy(mirror, destination, "Album/sibling.mp3", b"NEW", b"OLD")
    (destination / "Album" / "gone.mp3").write_bytes(b"old")

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert sibling.read_bytes() == b"NEW"
    assert not (destination / "Album" / "gone.mp3").exists()


def test_an_untouched_album_is_not_copied_again(mirror, tmp_path):
    """The second pass is scoped to albums that changed. An album whose file
    set is the same is left where it is, which is the whole point of not
    running --ignore-times over the library."""
    destination = tmp_path / "dest"
    destination.mkdir()
    quiet = stale_copy(mirror, destination, "Quiet/only.mp3", b"NEW", b"OLD")
    (destination / "Album").mkdir()
    shutil.copy2(mirror / "Album" / "track.mp3", destination / "Album" / "track.mp3")

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert quiet.read_bytes() == b"OLD"


def test_a_dry_run_says_how_many_extra_tracks_are_involved(mirror, tmp_path):
    destination = tmp_path / "dest"
    destination.mkdir()
    stale_copy(mirror, destination, "Album/sibling.mp3", b"NEW", b"OLD")

    result = run("-f", "-S", "-U", "-n", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert "and 1 more in those albums" in result.stderr


def test_a_first_sync_does_not_copy_anything_twice(mirror, tmp_path):
    """Everything is transferred by the main pass, so there is nothing left for
    the second one and it must not announce itself."""
    destination = tmp_path / "dest"
    destination.mkdir()

    result = run("-f", "-S", "-U", str(mirror), str(destination))

    assert result.returncode == 0, result.stderr
    assert "re-levelled" not in result.stderr