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