music-mirror/tests/test_submit_scrobbles.py

import json
import sys
from pathlib import Path

import pytest

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

import submit_scrobbles  # noqa: E402

LOG = """#AUDIOSCROBBLER/1.1
#TZ/UNKNOWN
#CLIENT/Rockbox ipodvideo 4.0
#ARTIST\t#ALBUM\t#TITLE\t#TRACKNUM\t#LENGTH\t#RATING\t#TIMESTAMP\t#MUSICBRAINZ_TRACKID
Pendulum\tImmersion\tWatercolour\t3\t245\tL\t1700000300\t
Green Day\tDookie\tBasket Case\t7\t180\tS\t1700000200\t
Mötley Crüe\tDr. Feelgood\tKickstart My Heart\t2\t283\tL\t1700000100\tmb-1
"""


def test_only_listened_tracks_are_submitted():
    """A skip is not a play."""
    played, skipped, timeless = submit_scrobbles.parse_log(LOG)

    assert [entry["track"] for entry in played] == ["Kickstart My Heart", "Watercolour"]
    assert skipped == 1
    assert timeless == 0


def test_entries_come_back_oldest_first():
    played, _, _ = submit_scrobbles.parse_log(LOG)

    assert [entry["timestamp"] for entry in played] == ["1700000100", "1700000300"]


def test_a_timeless_log_is_counted_and_not_submitted():
    """Without a real-time clock Rockbox writes every timestamp as zero. Those
    cannot be scrobbled without inventing when they happened."""
    log = LOG + "Band\tAlbum\tTrack\t-1\t100\tL\t0\t\n"

    played, _, timeless = submit_scrobbles.parse_log(log)

    assert timeless == 1
    assert all(int(entry["timestamp"]) > 0 for entry in played)


def test_absent_optional_fields_are_dropped():
    played, _, _ = submit_scrobbles.parse_log(LOG)
    params = submit_scrobbles.batch_params(played)

    assert "mbid[0]" in params  # Kickstart My Heart has one
    assert "mbid[1]" not in params  # Watercolour does not
    assert params["trackNumber[0]"] == "2"


def test_a_track_number_of_minus_one_is_not_sent():
    """Rockbox writes -1 when it does not know, which is not a track number."""
    played, _, _ = submit_scrobbles.parse_log(
        "Band\tAlbum\tTrack\t-1\t100\tL\t1700000000\t\n"
    )

    assert submit_scrobbles.batch_params(played).get("trackNumber[0]") is None


def test_the_signature_sorts_names_by_ascii_not_by_number():
    """Last.fm sorts parameter names as strings, so artist[10] precedes
    artist[1]. Sorting numerically produces an invalid signature and nothing
    else."""
    params = {"artist[1]": "b", "artist[10]": "a", "api_key": "k"}

    expected = submit_scrobbles.hashlib.md5(
        ("api_keyk" + "artist[1]b" + "artist[10]a" + "s").encode()
    ).hexdigest()
    assert submit_scrobbles.sign(params, "s") != expected

    correct = submit_scrobbles.hashlib.md5(
        ("api_keyk" + "artist[10]a" + "artist[1]b" + "s").encode()
    ).hexdigest()
    assert submit_scrobbles.sign(params, "s") == correct


def fake_transport(responses):
    """Return a transport serving canned responses and recording requests."""
    sent = []

    def transport(request, timeout=None):
        sent.append(dict(submit_scrobbles.urllib.parse.parse_qsl(request.data.decode())))
        return json.dumps(responses[len(sent) - 1])

    transport.sent = sent
    return transport


def test_scrobbles_are_sent_in_batches_of_fifty():
    entries = [
        {"artist": "A", "track": f"T{i}", "timestamp": str(1700000000 + i)}
        for i in range(120)
    ]
    accepted = {"scrobbles": {"@attr": {"accepted": 50, "ignored": 0}}}
    transport = fake_transport([accepted, accepted, accepted])

    submit_scrobbles.submit(entries, "k", "s", "sk", transport, delay=0)

    assert len(transport.sent) == 3
    assert transport.sent[0]["method"] == "track.scrobble"
    assert "artist[49]" in transport.sent[0]
    assert "artist[50]" not in transport.sent[0]


def test_every_request_carries_a_signature_and_session():
    entries = [{"artist": "A", "track": "T", "timestamp": "1700000000"}]
    transport = fake_transport([{"scrobbles": {"@attr": {"accepted": 1, "ignored": 0}}}])

    submit_scrobbles.submit(entries, "k", "s", "session-key", transport, delay=0)

    assert transport.sent[0]["sk"] == "session-key"
    assert len(transport.sent[0]["api_sig"]) == 32


def test_a_service_error_is_raised_not_swallowed():
    entries = [{"artist": "A", "track": "T", "timestamp": "1700000000"}]
    transport = fake_transport([{"error": 9, "message": "Invalid session key"}])

    with pytest.raises(submit_scrobbles.LastfmError, match="error 9"):
        submit_scrobbles.submit(entries, "k", "s", "sk", transport, delay=0)


def test_the_log_is_set_aside_after_a_successful_submission(tmp_path, monkeypatch):
    device = tmp_path / "IPOD"
    device.mkdir()
    (device / ".scrobbler.log").write_text(LOG, encoding="utf-8")
    transport = fake_transport([{"scrobbles": {"@attr": {"accepted": 2, "ignored": 0}}}])
    monkeypatch.setattr(submit_scrobbles, "load_session", lambda: "sk")

    submit_scrobbles.main(
        [str(device), "--api-key", "k", "--api-secret", "s"], transport=transport
    )

    assert not (device / ".scrobbler.log").exists()
    # Renamed, not deleted: if Last.fm quietly dropped one, the evidence remains.
    assert list(device.glob(".scrobbler.log.*.submitted"))


def test_a_failed_submission_leaves_the_log_alone(tmp_path, monkeypatch):
    device = tmp_path / "IPOD"
    device.mkdir()
    (device / ".scrobbler.log").write_text(LOG, encoding="utf-8")
    transport = fake_transport([{"error": 29, "message": "Rate limit"}])
    monkeypatch.setattr(submit_scrobbles, "load_session", lambda: "sk")

    code = submit_scrobbles.main(
        [str(device), "--api-key", "k", "--api-secret", "s"], transport=transport
    )

    assert code == 1
    assert (device / ".scrobbler.log").is_file()


def test_a_dry_run_submits_nothing(tmp_path):
    device = tmp_path / "IPOD"
    device.mkdir()
    (device / ".scrobbler.log").write_text(LOG, encoding="utf-8")
    transport = fake_transport([])

    submit_scrobbles.main([str(device), "--dry-run"], transport=transport)

    assert transport.sent == []
    assert (device / ".scrobbler.log").is_file()


def test_no_log_is_not_an_error(tmp_path):
    device = tmp_path / "IPOD"
    device.mkdir()

    assert submit_scrobbles.main([str(device)], transport=fake_transport([])) == 0


def test_write_credentials_are_required(tmp_path, capsys, monkeypatch):
    """The read-only key used elsewhere is not enough for a write method."""
    monkeypatch.delenv("LASTFM_API_KEY", raising=False)
    monkeypatch.delenv("LASTFM_API_SECRET", raising=False)
    device = tmp_path / "IPOD"
    device.mkdir()
    (device / ".scrobbler.log").write_text(LOG, encoding="utf-8")

    code = submit_scrobbles.main([str(device)], transport=fake_transport([]))

    assert code == 2
    assert "LASTFM_API_SECRET" in capsys.readouterr().err


PLAYBACK_LOG = """1700000300:180000:245000:/Music/Pendulum/Immersion/01.mp3
1700000200:9000:180000:/Music/Green Day/Dookie/07.mp3
0:180000:245000:/Music/No/Clock/track.mp3
malformed line without colons
"""


def tags_of(artist="Pendulum", title="Watercolour", album="Immersion", track="1/11"):
    def runner(path):
        return json.dumps(
            {"format": {"tags": {"ARTIST": artist, "TITLE": title,
                                 "ALBUM": album, "track": track}}}
        )

    return runner


def test_the_playback_log_format_is_four_fields():
    """timestamp:elapsed_ms:length_ms:path, written by Rockbox core."""
    plays = submit_scrobbles.parse_playback_log(PLAYBACK_LOG)

    assert len(plays) == 3  # the malformed line is dropped
    assert plays[0] == (1700000300, 180000, 245000, "/Music/Pendulum/Immersion/01.mp3")


def test_a_device_path_maps_onto_the_mirror():
    assert submit_scrobbles.device_to_local(
        "/Music/Pendulum/Immersion/01.mp3", "/Music", "/mnt/mirror"
    ) == Path("/mnt/mirror/Pendulum/Immersion/01.mp3")


def test_a_path_outside_the_prefix_is_not_mapped():
    """Something played from elsewhere on the card is not in the mirror."""
    assert submit_scrobbles.device_to_local(
        "/Podcasts/episode.mp3", "/Music", "/mnt/mirror"
    ) is None


def test_a_path_at_the_card_root_maps_straight_across():
    assert submit_scrobbles.device_to_local(
        "/Pendulum/Immersion/01.mp3", "/", "/mnt/mirror"
    ) == Path("/mnt/mirror/Pendulum/Immersion/01.mp3")


def test_a_short_play_is_a_skip_not_a_scrobble(tmp_path, monkeypatch):
    """Nine seconds of a three minute track. The on-device plugin uses the same
    fraction, so the two never disagree about what counted as a play."""
    monkeypatch.setattr(Path, "is_file", lambda self: True)

    played, skipped, unresolved, timeless = submit_scrobbles.plays_from_playback_log(
        PLAYBACK_LOG, "/Music", "/mnt/mirror", runner=tags_of()
    )

    assert skipped == 1
    assert [entry["timestamp"] for entry in played] == ["1700000300"]


def test_a_zero_timestamp_is_refused(tmp_path, monkeypatch):
    """Without a real-time clock Rockbox logs ticks, not dates. Scrobbling
    those would mean inventing when they happened."""
    monkeypatch.setattr(Path, "is_file", lambda self: True)

    _, _, _, timeless = submit_scrobbles.plays_from_playback_log(
        PLAYBACK_LOG, "/Music", "/mnt/mirror", runner=tags_of()
    )

    assert timeless == 1


def test_tags_come_from_the_mirror(tmp_path, monkeypatch):
    """The log carries only a path -- which is precisely why the on-device
    plugin exists. Off the mirror the tags are free."""
    monkeypatch.setattr(Path, "is_file", lambda self: True)

    played, _, _, _ = submit_scrobbles.plays_from_playback_log(
        PLAYBACK_LOG, "/Music", "/mnt/mirror", runner=tags_of()
    )

    assert played[0]["artist"] == "Pendulum"
    assert played[0]["album"] == "Immersion"
    assert played[0]["trackNumber"] == "1"       # "1/11" -> "1"
    assert played[0]["duration"] == "245"        # milliseconds -> seconds


def test_a_track_missing_from_the_mirror_is_counted_not_guessed(monkeypatch):
    monkeypatch.setattr(Path, "is_file", lambda self: False)

    played, _, unresolved, _ = submit_scrobbles.plays_from_playback_log(
        PLAYBACK_LOG, "/Music", "/mnt/mirror", runner=tags_of()
    )

    assert played == []
    # One: the skip and the clockless entry are filtered before the file is
    # looked for, since neither would be submitted either way.
    assert unresolved == 1


def test_playback_logs_are_found_including_rotations(tmp_path):
    """Rockbox rotates the log once it passes half a megabyte."""
    rockbox = tmp_path / ".rockbox"
    rockbox.mkdir()
    for name in ("playback.log", "playback_0001.log", "playback_0002.log"):
        (rockbox / name).write_text("1700000000:1:1:/Music/a.mp3\n")
    (rockbox / "empty.log").write_text("")

    found = submit_scrobbles.find_playback_logs(tmp_path)

    assert [path.name for path in found] == [
        "playback.log", "playback_0001.log", "playback_0002.log"
    ]


def test_without_a_mirror_it_says_what_is_needed(tmp_path, capsys):
    device = tmp_path / "IPOD"
    (device / ".rockbox").mkdir(parents=True)
    (device / ".rockbox" / "playback.log").write_text("1700000000:1:1:/Music/a.mp3\n")

    submit_scrobbles.main([str(device)], transport=fake_transport([]))

    assert "pass --mirror" in capsys.readouterr().err