music-mirror/tests/test_rsync_progress.py

import io
import sys
from pathlib import Path

import pytest

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

import rsync_progress  # noqa: E402


class NotATerminal(io.StringIO):
    def isatty(self):
        return False


class Terminal(io.StringIO):
    def isatty(self):
        return True


def run(lines, total=0, out=None, bytes_expected=0):
    out = out or NotATerminal()
    rsync_progress.main(
        ["--total", str(total), "--bytes", str(bytes_expected)],
        stream=io.StringIO(lines),
        out=out,
    )
    return out.getvalue()


def test_the_artist_and_album_are_pulled_from_the_path():
    assert rsync_progress.album_of("Pendulum/Immersion/01 - Watercolour.mp3") == (
        "Pendulum / Immersion"
    )


def test_a_shallower_path_degrades_rather_than_failing():
    assert rsync_progress.album_of("Pendulum/loose.mp3") == "Pendulum"
    assert rsync_progress.album_of("loose.mp3") == ""


def test_directories_are_not_counted():
    """rsync reports them too, and counting them puts the percentage past 100."""
    output = run("Artist/\nArtist/Album/\nArtist/Album/track.mp3\n", total=1)

    assert "1/1" in output
    assert "100%" in output


def test_the_percentage_tracks_the_total():
    output = run("".join(f"A/B/{i}.mp3\n" for i in range(5)), total=10)

    assert "5/10" in output
    assert "50%" in output


def test_without_a_total_it_counts_instead_of_guessing():
    output = run("A/B/one.mp3\nA/B/two.mp3\n")

    assert "2 files" in output
    assert "%" not in output


def test_a_final_line_is_always_printed():
    """Otherwise the last state of a rewriting line is whatever it happened to
    be when the interval last elapsed."""
    output = run("A/B/one.mp3\n", total=1)

    assert output.endswith("\n")
    assert "1/1" in output


def test_nothing_transferred_still_reports():
    output = run("", total=0)

    assert "0 files" in output


def test_a_log_gets_no_carriage_returns():
    """A non-terminal filling with \\r and escape codes is unreadable."""
    output = run("".join(f"A/B/{i}.mp3\n" for i in range(50)), total=50)

    assert "\r" not in output
    assert "\033" not in output


def test_a_terminal_rewrites_one_line():
    output = run("".join(f"A/B/{i}.mp3\n" for i in range(50)), total=50, out=Terminal())

    assert "\r\033[2K" in output


@pytest.mark.parametrize(
    ("text", "width", "expected"),
    [
        ("short", 20, "short"),
        ("King Gizzard / PetroDragonic Apocalypse", 20, "…agonic Apocalypse"),
    ],
)
def test_long_labels_are_trimmed_from_the_left(text, width, expected):
    """The album is the informative end, so the artist is what gets cut."""
    trimmed = rsync_progress.fit(text, width)

    assert len(trimmed) <= width
    if len(text) > width:
        assert trimmed.startswith("…")
        assert text.endswith(trimmed.lstrip("…"))


def test_the_size_and_path_are_parsed():
    assert rsync_progress.parse("5000 Artist/Album/Track.mp3\n") == (
        5000,
        "Artist/Album/Track.mp3",
    )


def test_a_filename_containing_spaces_survives():
    """Splitting on every space would lose most of the library."""
    assert rsync_progress.parse("1234 Artist/An Album/A Track With Spaces.mp3") == (
        1234,
        "Artist/An Album/A Track With Spaces.mp3",
    )


def test_a_bare_path_is_tolerated():
    """In case this is fed --out-format='%n' by something older."""
    assert rsync_progress.parse("Artist/Album/Track.mp3") == (0, "Artist/Album/Track.mp3")


def test_directory_sizes_do_not_inflate_the_total():
    """rsync reports directories with a 4096 inode size, which is several
    megabytes of nothing across six thousand albums."""
    output = run("4096 Artist/\n4096 Artist/Album/\n5000 Artist/Album/t.mp3\n", total=1)

    assert "4.9 KiB" in output
    assert "12" not in output.split("Artist")[0]


def test_a_rate_is_not_reported_until_it_means_something():
    """The first files arrive microseconds apart and would give a rate in the
    gigabytes per second and an ETA of zero."""
    rate = rsync_progress.Rate()
    rate.add(100.0, 0)
    rate.add(100.5, 5_000_000)

    assert rate.per_second() == 0.0


def test_a_rate_over_a_long_enough_window_is_reported():
    rate = rsync_progress.Rate()
    rate.add(100.0, 0)
    rate.add(110.0, 10_000_000)

    assert rate.per_second() == pytest.approx(1_000_000)


def test_the_window_forgets_the_distant_past():
    """So the estimate follows a device that slows down rather than averaging
    the slowdown away."""
    rate = rsync_progress.Rate(window=30.0)
    for second in range(0, 100, 10):
        rate.add(float(second), second * 1_000_000)
    rate.add(200.0, 100_000_000)

    assert rate.samples[0][0] >= 90.0


@pytest.mark.parametrize(
    ("seconds", "expected"),
    [(0, "0s"), (45, "45s"), (60, "1m00s"), (1092, "18m12s"), (7500, "2h05m")],
)
def test_durations_read_without_arithmetic(seconds, expected):
    assert rsync_progress.human_duration(seconds) == expected


def test_a_summary_is_printed_at_the_end():
    output = run("5000000 A/B/one.mp3\n", total=1, bytes_expected=5000000)

    assert "copied 4.8 MiB in" in output