feat: a sync script that submits scrobbles, copies, and unmounts cleanly
Build and publish container / build (pull_request) Canceled after 3m8s
Build and publish container / build (pull_request) Canceled after 3m8s
tools/sync-to-ipod.sh does the whole transfer to a Rockbox device, so the only manual part left is the disk-mode button sequence. The guards are the substance rather than decoration. rsync --delete is being aimed at a whole filesystem, so the destination must exist, be its own mount point, and be a FAT filesystem; the mirror must be non-empty and must not be the destination. Emptying the wrong directory is not a mistake that announces itself. It also excludes /.rockbox, the scrobbler logs and the usual filesystem metadata directories. The mirror does not contain them, so a sync to the card root would otherwise have deleted the Rockbox installation -- which the first draft of this script would have done. The unmount is why this is a script at all. FAT32 has no journal, the device is reached through the Apple firmware's disk mode because Rockbox's own mass storage is unreliable on an iFlash, and an interrupted write is corruption that needs fsck.vfat from another machine. tools/submit_scrobbles.py sends the Rockbox scrobbler log to Last.fm and sets it aside. Rockbox writes it in AUDIOSCROBBLER 1.1: tab-separated, one line per track, rated L for listened or S for skipped, and only the listened ones are a play. It runs before the copy, because the plays already happened and a failed transfer is no reason to lose them as well. Two things there differ from every other Last.fm call in these projects. Scrobbling is a write method, so it needs the API secret and a session key obtained once through the browser rather than the read-only key. And a target with no real-time clock gets /.scrobbler-timeless.log with every timestamp set to zero; those are counted and reported but never sent, since submitting them would mean inventing when they happened. Signature generation sorts parameter names by the ASCII table rather than numerically, so artist[10] precedes artist[1]. Sorting them the obvious way produces an invalid signature and no other symptom, so there is a test for it. The log is renamed rather than deleted once accepted, so that if Last.fm quietly dropped something the evidence is still on the device.
This commit is contained in:
@@ -0,0 +1,189 @@
|
||||
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
|
||||
Reference in New Issue
Block a user