music-mirror/tests/test_submit_scrobbles.py

import json
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from zoneinfo import ZoneInfo

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)

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

    assert result.skipped == 1
    assert [entry["timestamp"] for entry in result.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)

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

    assert result.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()
    ).played

    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)

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

    assert result.played == []
    # One: the skip and the clockless entry are filtered before the file is
    # looked for, since neither would be submitted either way.
    assert result.unresolved == 1
    # And that one play is kept, so a later run can try it again.
    assert len(result.retain) == 1


LONDON = ZoneInfo("Europe/London")

# Rockbox builds timestamps with mktime(get_time()), and its mktime applies no
# zone at all, so the number is the RTC's local wall clock read as if it were
# UTC. Under BST that puts every play an hour ahead of when it happened.
SUMMER_LOGGED = 1787839200   # 2026-08-27 14:00 written by the device
SUMMER_TRUE = 1787835600     # the instant that actually was, 13:00 UTC
WINTER_LOGGED = 1796479200   # 2026-12-05 14:00, when London is already UTC
WINTER_TRUE = 1796479200


def test_a_summer_timestamp_is_pulled_back_to_real_utc():
    """The device logged 14:00 local. That was 13:00 UTC, and UTC is what
    Last.fm stores."""
    assert submit_scrobbles.device_time_to_utc(SUMMER_LOGGED, LONDON) == SUMMER_TRUE


def test_a_winter_timestamp_is_left_alone():
    """London keeps UTC for half the year, so there is nothing to correct and
    the correction must not invent an offset anyway."""
    assert submit_scrobbles.device_time_to_utc(WINTER_LOGGED, LONDON) == WINTER_TRUE


def test_the_converted_time_reads_back_as_the_clock_the_device_showed():
    """The round trip, which is the property that actually matters: whatever
    the player's screen said is what Last.fm should show in local time."""
    shown = datetime.fromtimestamp(SUMMER_LOGGED, timezone.utc)
    corrected = submit_scrobbles.device_time_to_utc(SUMMER_LOGGED, LONDON)

    assert datetime.fromtimestamp(corrected, LONDON).strftime("%Y-%m-%d %H:%M") == (
        shown.strftime("%Y-%m-%d %H:%M")
    )


def test_one_log_spanning_a_clock_change_converts_each_side_separately():
    """A week's listening either side of the October change carries two
    different offsets. Correcting the log by a single figure would put half of
    it an hour out, which is why the offset is resolved per play."""
    before, after = 1792888200, 1792899000  # 2026-10-25, 00:30 BST and 03:30 GMT

    assert submit_scrobbles.device_time_to_utc(before, LONDON) == before - 3600
    assert submit_scrobbles.device_time_to_utc(after, LONDON) == after


def test_a_playback_log_is_corrected_before_submission(monkeypatch):
    monkeypatch.setattr(Path, "is_file", lambda self: True)
    log = f"{SUMMER_LOGGED}:180000:245000:/Music/Pendulum/Immersion/01.mp3\n"

    played = submit_scrobbles.plays_from_playback_log(
        log, "/Music", "/mnt/mirror", runner=tags_of(), zone=LONDON
    ).played

    assert played[0]["timestamp"] == str(SUMMER_TRUE)


def test_a_scrobbler_log_claiming_utc_is_not_corrected_twice():
    """The format's header exists for exactly this. A client that already did
    the conversion says so, and correcting it again would break it."""
    log = LOG.replace("#TZ/UNKNOWN", "#TZ/UTC")

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

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


def test_a_scrobbler_log_declaring_unknown_is_corrected():
    """Rockbox writes UNKNOWN, meaning local wall clock, so the times are ours
    to fix."""
    played, _, _ = submit_scrobbles.parse_log(LOG, LONDON)

    # 1700000100 and 1700000300 are November, when London is on UTC anyway;
    # the point is that the header did not exempt them from being looked at.
    assert submit_scrobbles.declares_utc(LOG) is False
    assert [entry["timestamp"] for entry in played] == ["1700000100", "1700000300"]


def test_the_system_zone_keeps_its_whole_name(monkeypatch, tmp_path):
    """/etc/localtime points into the tzdata tree. Taking only the last
    component gives "London", which no database holds -- and the fallback for
    an unknown name is a fixed offset, which is wrong for half the year. The
    region has to survive."""
    zoneinfo_dir = tmp_path / "share" / "zoneinfo" / "Europe"
    zoneinfo_dir.mkdir(parents=True)
    (zoneinfo_dir / "London").write_bytes(b"TZif")
    monkeypatch.setattr(
        submit_scrobbles.Path,
        "read_text",
        lambda self, **kw: (_ for _ in ()).throw(OSError),
    )
    monkeypatch.setattr(
        submit_scrobbles.Path,
        "resolve",
        lambda self: zoneinfo_dir / "London",
    )

    assert submit_scrobbles.system_zone_name() == "Europe/London"


def test_a_named_zone_beats_the_machine_default():
    """The player may not be in the same place as the laptop."""
    assert submit_scrobbles.device_zone("Asia/Tokyo") == ZoneInfo("Asia/Tokyo")


def test_plays_dated_after_now_are_reported():
    """A clock never put forward, or the wrong zone, produces plays that have
    not happened yet. Last.fm cannot tell those from real ones."""
    now = 1787835600
    played = [
        {"timestamp": str(now - 60)},
        {"timestamp": str(now + 3600)},
    ]

    ahead = submit_scrobbles.future_plays(played, now=now)

    assert [entry["timestamp"] for entry in ahead] == [str(now + 3600)]


def test_a_clock_a_minute_fast_is_not_reported():
    """Devices drift. Only an offset large enough to be a zone error matters."""
    now = 1787835600

    assert submit_scrobbles.future_plays([{"timestamp": str(now + 60)}], now=now) == []


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


MIRROR = "/mnt/mirror"


def stub_ffprobe(monkeypatch, **tags):
    """Answer for anything under the mirror without invoking ffprobe."""
    monkeypatch.setattr(submit_scrobbles, "_ffprobe", tags_of(**tags))


def only_mirror_files_exist(monkeypatch, resolvable=None):
    """Make the mirror's files appear to exist, and nothing else.

    Patching is_file wholesale makes the device directory look like a log file,
    which sends main() down the .scrobbler.log path instead.
    """
    real = Path.is_file

    def patched(self):
        text = str(self)
        if text.startswith(MIRROR):
            return resolvable is None or text == resolvable
        return real(self)

    monkeypatch.setattr(Path, "is_file", patched)


def playback_device(tmp_path, log=PLAYBACK_LOG):
    device = tmp_path / "IPOD"
    (device / ".rockbox").mkdir(parents=True)
    (device / ".rockbox" / "playback.log").write_text(log)
    return device


def test_a_failed_submission_keeps_every_log(tmp_path, monkeypatch):
    """Nothing got through, so nothing may be set aside."""
    only_mirror_files_exist(monkeypatch)
    stub_ffprobe(monkeypatch)
    monkeypatch.setattr(submit_scrobbles, "load_session", lambda: "sk")
    device = playback_device(tmp_path)
    transport = fake_transport([{"error": 29, "message": "Rate limit"}])

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

    assert code == 1
    assert (device / ".rockbox" / "playback.log").is_file()
    assert not list((device / ".rockbox").glob("*.submitted"))


def test_an_unmatched_play_is_written_back_not_lost(tmp_path, monkeypatch, capsys):
    """A track the mirror does not yet hold is still a play that happened. It
    is kept so a later run, after the file has been copied, can submit it."""
    monkeypatch.setattr(submit_scrobbles, "load_session", lambda: "sk")
    # Only the first track resolves; the rest are absent from the mirror.
    only_mirror_files_exist(monkeypatch, f"{MIRROR}/Pendulum/Immersion/01.mp3")
    stub_ffprobe(monkeypatch)
    log = (
        "1700000300:180000:245000:/Music/Pendulum/Immersion/01.mp3\n"
        "1700000400:180000:245000:/Music/Missing/Album/09.mp3\n"
    )
    device = playback_device(tmp_path, log)
    transport = fake_transport([{"scrobbles": {"@attr": {"accepted": 1, "ignored": 0}}}])

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

    assert code == 0
    rockbox = device / ".rockbox"
    # The original is preserved untouched...
    assert list(rockbox.glob("playback.log.*.submitted"))
    # ...and the unmatched play is back in a live log for the next attempt.
    written = (rockbox / "playback.log").read_text()
    assert "Missing/Album/09.mp3" in written
    assert "Pendulum/Immersion/01.mp3" not in written


def test_the_original_is_renamed_rather_than_deleted(tmp_path, monkeypatch):
    """If Last.fm quietly dropped something, the evidence stays on the device."""
    only_mirror_files_exist(monkeypatch)
    stub_ffprobe(monkeypatch)
    monkeypatch.setattr(submit_scrobbles, "load_session", lambda: "sk")
    device = playback_device(tmp_path)
    transport = fake_transport([{"scrobbles": {"@attr": {"accepted": 1, "ignored": 0}}}])

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

    aside = list((device / ".rockbox").glob("playback.log.*.submitted"))
    assert len(aside) == 1
    assert PLAYBACK_LOG.splitlines()[0] in aside[0].read_text()


def test_keep_leaves_everything_alone(tmp_path, monkeypatch):
    only_mirror_files_exist(monkeypatch)
    stub_ffprobe(monkeypatch)
    monkeypatch.setattr(submit_scrobbles, "load_session", lambda: "sk")
    device = playback_device(tmp_path)
    transport = fake_transport([{"scrobbles": {"@attr": {"accepted": 1, "ignored": 0}}}])

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

    assert (device / ".rockbox" / "playback.log").read_text() == PLAYBACK_LOG
    assert not list((device / ".rockbox").glob("*.submitted"))


def test_a_file_ffprobe_cannot_read_does_not_abandon_the_rest(monkeypatch):
    """CalledProcessError is not an OSError, so one bad file used to take the
    whole submission with it."""
    def broken(path):
        raise subprocess.CalledProcessError(1, "ffprobe")

    assert submit_scrobbles.read_tags(Path("/mnt/mirror/x.mp3"), runner=broken) == {}