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 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_quick_mode_skips_the_counting_pass(mirror, tmp_path):
    """Over SMB the walk is the expensive part, and doing it twice for a
    percentage is not always the trade you want."""
    destination = tmp_path / "dest"
    destination.mkdir()

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

    assert result.returncode == 0, result.stderr
    assert "skipping the count" in 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"