music-curator/tests/test_music_curator.py

import urllib.error

import pytest
from conftest import FakeLastfm, FakeLidarr, make_loved, make_tracks

import music_curator

NOW = 1_700_000_000


def store_at(tmp_path):
    return music_curator.Store(tmp_path / "curator.db")


def client_for(api, **kwargs):
    kwargs.setdefault("delay", 0)
    kwargs.setdefault("backoff", 0)
    return music_curator.Lastfm("key", transport=api, **kwargs)


def ingest(store, api, backfill_limit=0):
    client = client_for(api)
    return music_curator.run_once(client, store, "lyra", NOW, backfill_limit)


def scrobble_of(artist, track, uts=NOW - 3600, mbid=""):
    """One recent-tracks entry, in the shape Last.fm actually sends."""
    return {
        "artist": {"#text": artist, "mbid": ""},
        "album": {"#text": "Some Album", "mbid": ""},
        "name": track,
        "mbid": mbid,
        "date": {"uts": str(uts)},
    }


LIBRARY = [
    {
        "name": "Yellowcard",
        "albums": [
            {
                "title": "Lift a Sail",
                "tracks": [
                    {"title": "Here I Am Alive", "recording_mbid": "rec-alive"},
                    {"title": "Transmission Home"},
                ],
            }
        ],
    },
    {
        "name": "AC/DC",
        "albums": [{"title": "Back in Black", "tracks": [{"title": "Hells Bells"}]}],
    },
    {
        "name": "The Prodigy",
        "albums": [
            {"title": "The Fat of the Land", "tracks": [{"title": "Breathe (Remastered)"}]}
        ],
    },
]


def indexed(tmp_path, scrobbles):
    """Ingest scrobbles, index the canned library, and match the two."""
    store = store_at(tmp_path)
    ingest(store, FakeLastfm(scrobbles))
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=FakeLidarr(LIBRARY)), store
    )
    music_curator.match_library(store)
    return store


def verdict(store, artist, track):
    row = store.connection.execute(
        "SELECT method, track_id FROM scrobble_key WHERE artist = ? AND track = ?",
        (artist, track),
    ).fetchone()
    return (row["method"], row["track_id"]) if row else (None, None)


def test_parse_interval_units():
    assert music_curator.parse_interval("90") == 90
    assert music_curator.parse_interval("30m") == 1800
    assert music_curator.parse_interval("6h") == 21600
    with pytest.raises(ValueError):
        music_curator.parse_interval("soon")


def test_artist_name_is_read_from_either_shape():
    """Recent tracks key it `#text`; loved tracks and extended=1 key it `name`."""
    assert music_curator.name_of({"#text": "Autechre"}) == "Autechre"
    assert music_curator.name_of({"name": "Autechre"}) == "Autechre"
    assert music_curator.name_of(None) == ""


def test_empty_mbid_becomes_none():
    """An empty string would otherwise look like a usable join key later on."""
    assert music_curator.mbid_of({"mbid": ""}) is None
    assert music_curator.mbid_of("") is None
    assert music_curator.mbid_of({"mbid": "abc"}) == "abc"


def test_backfill_ingests_the_whole_history(tmp_path):
    api = FakeLastfm(make_tracks(450))
    store = store_at(tmp_path)

    ingest(store, api)

    assert store.count() == 450
    assert store.get_state("backfill_complete") == "yes"


def test_backfill_resumes_after_a_capped_pass(tmp_path):
    """An interrupted backfill picks up from what the database holds."""
    api = FakeLastfm(make_tracks(450))
    store = store_at(tmp_path)

    ingest(store, api, backfill_limit=1)
    partial = store.count()
    assert 0 < partial < 450
    assert store.get_state("backfill_complete") != "yes"

    ingest(store, api)

    assert store.count() == 450
    assert store.get_state("backfill_complete") == "yes"


def test_now_playing_is_not_stored(tmp_path, now_playing):
    """It has no timestamp, and it would arrive again on every single pass."""
    api = FakeLastfm(make_tracks(10), nowplaying=now_playing)
    store = store_at(tmp_path)

    ingest(store, api)

    assert store.count() == 10
    titles = {row["track"] for row in store.connection.execute("SELECT track FROM scrobble")}
    assert "Currently Playing" not in titles


def test_a_pass_over_unchanged_history_adds_nothing(tmp_path):
    api = FakeLastfm(make_tracks(300))
    store = store_at(tmp_path)

    ingest(store, api)
    before = store.count()

    assert ingest(store, api) == 0
    assert store.count() == before


def test_catch_up_adds_only_what_is_new(tmp_path):
    history = make_tracks(300)
    api = FakeLastfm(history)
    store = store_at(tmp_path)

    ingest(store, api)

    newest = int(history[-1]["date"]["uts"])
    api.tracks = sorted(
        [*history, *make_tracks(5, start=newest + 3600)],
        key=lambda track: int(track["date"]["uts"]),
        reverse=True,
    )

    assert ingest(store, api) == 5
    assert store.count() == 305


def test_a_lone_result_is_not_a_list(tmp_path):
    """Last.fm returns a bare object rather than a one-item list."""
    api = FakeLastfm(make_tracks(1))
    store = store_at(tmp_path)

    ingest(store, api)

    assert store.count() == 1


def test_a_full_page_sharing_one_second_is_paged_not_skipped(tmp_path):
    """Moving the window past a same-second cluster would lose the rest of it."""
    api = FakeLastfm(make_tracks(250, step=0))
    store = store_at(tmp_path)

    ingest(store, api)

    # 250 plays at one timestamp collapse to one row per distinct track.
    assert store.count() == 250
    assert store.oldest_uts() == store.newest_uts()


def test_loved_tracks_are_replaced_not_accumulated(tmp_path):
    api = FakeLastfm(make_tracks(5), loved=make_loved(["One", "Two", "Three"]))
    store = store_at(tmp_path)

    ingest(store, api)
    assert store.scalar("SELECT COUNT(*) FROM loved") == 3

    api.loved = make_loved(["One"])
    ingest(store, api)

    assert store.scalar("SELECT COUNT(*) FROM loved") == 1


def test_rate_limiting_is_retried(tmp_path):
    """Error 29 is the service asking for patience, not a broken request."""
    api = FakeLastfm(make_tracks(5), outcomes=[{"error": 29, "message": "Rate Limit Exceded"}])
    store = store_at(tmp_path)

    ingest(store, api)

    assert store.count() == 5


def test_server_errors_are_retried(tmp_path):
    api = FakeLastfm(
        make_tracks(5),
        outcomes=[urllib.error.HTTPError("url", 503, "unavailable", {}, None)],
    )
    store = store_at(tmp_path)

    ingest(store, api)

    assert store.count() == 5


def test_an_invalid_api_key_is_not_retried():
    """Error 10 will fail identically forever; failing fast says why."""
    api = FakeLastfm(outcomes=[{"error": 10, "message": "Invalid API key"}])
    client = client_for(api)

    with pytest.raises(music_curator.LastfmError, match="error 10"):
        client.call("user.getRecentTracks", {"user": "lyra"})

    assert len(api.calls) == 1


def test_giving_up_reports_the_last_failure():
    api = FakeLastfm(outcomes=[{"error": 29, "message": "slow down"}] * 3)
    client = client_for(api, attempts=3)

    with pytest.raises(music_curator.LastfmError, match="error 29"):
        client.call("user.getRecentTracks", {"user": "lyra"})

    assert len(api.calls) == 3


def test_the_store_rejects_a_schema_it_does_not_understand(tmp_path):
    store = store_at(tmp_path)
    store.set_state("schema_version", "999")
    store.close()

    with pytest.raises(RuntimeError, match="schema version"):
        store_at(tmp_path)


def test_missing_credentials_are_refused(tmp_path, monkeypatch):
    monkeypatch.delenv("MUSIC_CURATOR_LASTFM_USER", raising=False)
    monkeypatch.delenv("MUSIC_CURATOR_LASTFM_API_KEY", raising=False)

    assert music_curator.main(["--db", str(tmp_path / "curator.db")]) == 2


def test_report_only_needs_no_credentials(tmp_path):
    assert music_curator.main(["--db", str(tmp_path / "curator.db"), "--report-only"]) == 0


def test_a_second_pass_will_not_start_while_one_is_running(tmp_path, monkeypatch):
    database = tmp_path / "curator.db"
    held = music_curator.acquire_lock(database)
    assert held is not None

    try:
        assert music_curator.main(["--db", str(database), "--report-only"]) == 3
    finally:
        held.close()


def test_main_ingests_through_the_module_level_fetcher(tmp_path, monkeypatch):
    api = FakeLastfm(make_tracks(20))
    monkeypatch.setattr(music_curator, "http_get", api)

    code = music_curator.main(
        [
            "--db",
            str(tmp_path / "curator.db"),
            "--user",
            "lyra",
            "--api-key",
            "key",
            "--request-delay",
            "0",
        ],
        clock=lambda: NOW,
    )

    assert code == 0
    assert store_at(tmp_path).count() == 20


def test_normalise_flattens_the_ways_tags_and_scrobbles_disagree():
    assert music_curator.normalise("Yellowcard feat. Tay Jardine") == "yellowcard"
    assert music_curator.normalise("BABYMETAL Feat. F.Hero") == "babymetal"
    assert music_curator.normalise("AC/DC") == "ac dc"
    assert music_curator.normalise("The Prodigy") == "prodigy"
    assert music_curator.normalise("Beyoncé") == "beyonce"
    assert music_curator.normalise("Simon & Garfunkel") == "simon and garfunkel"
    assert music_curator.normalise("Breathe (Remastered 2011)") == "breathe"
    assert music_curator.normalise("Breathe (Live) (Remastered)") == "breathe"
    assert music_curator.normalise("Vice Grip - Live") == "vice grip"


def test_normalise_keeps_a_leading_parenthetical():
    """Stripping every bracket would destroy real titles."""
    assert music_curator.normalise("(Don't Fear) The Reaper") == "dont fear the reaper"


def test_normalise_treats_apostrophes_and_other_punctuation_differently():
    """The two rules pull opposite ways: one deletes, the other separates."""
    assert music_curator.normalise("Don't") == music_curator.normalise("Dont")
    assert music_curator.normalise("Don’t") == music_curator.normalise("Dont")
    assert music_curator.normalise("AC/DC") == music_curator.normalise("AC-DC")
    assert music_curator.normalise("AC/DC") == music_curator.normalise("AC DC")
    # ...and they must not collide with each other.
    assert music_curator.normalise("AC/DC") != music_curator.normalise("ACDC")


def test_normalise_leaves_with_alone():
    """`with` appears in too many genuine titles to cut on sight."""
    assert music_curator.normalise("Dancing with Myself") == "dancing with myself"


def test_a_recording_mbid_matches_exactly(tmp_path):
    store = indexed(tmp_path, [scrobble_of("Yellowcard", "Here I Am Alive", mbid="rec-alive")])

    assert verdict(store, "Yellowcard", "Here I Am Alive")[0] == "mbid"


def test_a_guest_credit_still_matches_by_name(tmp_path):
    """Last.fm puts the guest in the artist field; the file tag does not."""
    store = indexed(
        tmp_path, [scrobble_of("Yellowcard feat. Tay Jardine", "Here I Am Alive")]
    )

    method, track_id = verdict(store, "Yellowcard feat. Tay Jardine", "Here I Am Alive")
    assert method == "name"
    assert track_id is not None


def test_a_remaster_suffix_on_the_library_side_still_matches(tmp_path):
    store = indexed(tmp_path, [scrobble_of("The Prodigy", "Breathe")])

    assert verdict(store, "The Prodigy", "Breathe")[0] == "name"


def test_music_that_is_not_in_the_library_stays_unmatched(tmp_path):
    store = indexed(tmp_path, [scrobble_of("Some Band", "Some Song")])

    assert verdict(store, "Some Band", "Some Song") == ("none", None)


def test_the_mbid_tier_is_preferred_over_the_name_tier(tmp_path):
    store = indexed(
        tmp_path,
        [scrobble_of("Yellowcard", "Here I Am Alive", mbid="rec-alive")],
    )
    method, track_id = verdict(store, "Yellowcard", "Here I Am Alive")

    expected = store.connection.execute(
        "SELECT id FROM lidarr_track WHERE recording_mbid = 'rec-alive'"
    ).fetchone()["id"]
    assert (method, track_id) == ("mbid", expected)


def test_keys_carry_the_play_count_and_the_span(tmp_path):
    store = indexed(
        tmp_path,
        [
            scrobble_of("AC/DC", "Hells Bells", uts=NOW - 7200),
            scrobble_of("AC/DC", "Hells Bells", uts=NOW - 3600),
        ],
    )

    row = store.connection.execute(
        "SELECT plays, first_uts, last_uts FROM scrobble_key WHERE artist = 'AC/DC'"
    ).fetchone()
    assert row["plays"] == 2
    assert row["last_uts"] - row["first_uts"] == 3600


def test_re_indexing_drops_what_lidarr_no_longer_has(tmp_path):
    """The index is Lidarr's mirror, not an accumulation of everything ever seen."""
    store = indexed(tmp_path, [scrobble_of("AC/DC", "Hells Bells")])
    assert store.scalar("SELECT COUNT(*) FROM lidarr_artist") == 3

    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=FakeLidarr(LIBRARY[:1])), store
    )

    assert store.scalar("SELECT COUNT(*) FROM lidarr_artist") == 1
    assert store.scalar("SELECT COUNT(*) FROM lidarr_track WHERE artist_id NOT IN"
                        " (SELECT id FROM lidarr_artist)") == 0


def test_the_index_records_paths_and_when_a_file_landed(tmp_path):
    store = indexed(tmp_path, [])

    row = store.connection.execute(
        "SELECT path, added, has_file FROM lidarr_track WHERE title = 'Hells Bells'"
    ).fetchone()
    assert row["path"].endswith("Hells Bells.flac")
    assert row["has_file"] == 1
    assert row["added"] == music_curator.parse_added("2020-05-01T12:00:00Z")


def test_a_track_without_a_file_has_no_path(tmp_path):
    library = [
        {
            "name": "Ghost Artist",
            "albums": [{"title": "Unowned", "tracks": [{"title": "Missing", "has_file": False}]}],
        }
    ]
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=FakeLidarr(library)), store
    )

    row = store.connection.execute("SELECT path, has_file FROM lidarr_track").fetchone()
    assert row["has_file"] == 0
    assert row["path"] is None


def test_lidarr_requires_the_api_key_header(tmp_path):
    """The fake asserts on it, which is the point: a missing header is a 401."""
    client = music_curator.Lidarr("http://lidarr", "key", transport=FakeLidarr(LIBRARY))

    assert client.get("artist")


def test_a_lidarr_failure_is_reported_not_swallowed():
    def broken(url, timeout=None, headers=None):
        raise urllib.error.HTTPError(url, 401, "unauthorised", {}, None)

    client = music_curator.Lidarr("http://lidarr", "wrong", transport=broken)

    with pytest.raises(music_curator.LidarrError, match="HTTP 401"):
        client.get("artist")


def test_an_existing_version_one_store_is_migrated_not_discarded(tmp_path):
    """A rebuilt history is thousands of API requests; migration is additive."""
    store = store_at(tmp_path)
    store.add_scrobbles([music_curator.Scrobble(uts=NOW, artist="A", track="B")])
    store.set_state("schema_version", "1")
    store.close()

    reopened = store_at(tmp_path)

    assert reopened.count() == 1
    assert reopened.get_state("schema_version") == music_curator.SCHEMA_VERSION


def test_an_unmatched_track_by_a_known_artist_is_a_matcher_miss(tmp_path):
    """The split that matters: music you own and played, that failed to match."""
    store = indexed(
        tmp_path,
        [
            scrobble_of("AC/DC", "A Track Lidarr Has Never Heard Of"),
            scrobble_of("Some Band", "Some Song"),
        ],
    )

    misses = store.connection.execute(
        "SELECT k.artist FROM scrobble_key k WHERE k.track_id IS NULL"
        "   AND EXISTS (SELECT 1 FROM lidarr_artist a WHERE a.norm_name = k.norm_artist)"
    ).fetchall()

    # Only the AC/DC one: "Some Band" is not in the library, so it says nothing
    # about the matcher.
    assert [row["artist"] for row in misses] == ["AC/DC"]


def test_the_report_runs_over_a_matched_store(tmp_path):
    store = indexed(tmp_path, [scrobble_of("Yellowcard", "Here I Am Alive", mbid="rec-alive")])

    music_curator.report(store, NOW)


def test_matching_is_redone_when_new_scrobbles_arrive(tmp_path):
    """A pass with no re-index still has to give the new plays a verdict."""
    store = indexed(tmp_path, [scrobble_of("AC/DC", "Hells Bells")])
    api = FakeLastfm(
        [
            scrobble_of("AC/DC", "Hells Bells"),
            scrobble_of("Yellowcard", "Transmission Home", uts=NOW - 60),
        ]
    )

    music_curator.run_once(client_for(api), store, "lyra", NOW, 0)

    assert verdict(store, "Yellowcard", "Transmission Home")[0] == "name"