music-curator/tests/test_music_curator.py

import json
import stat
import urllib.error
from pathlib import Path

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)"},
                    # Credits its remixer in the title, which is the only signal
                    # separating a real attribution miss from a title collision.
                    {"title": "Voodoo People (Pendulum Remix)"},
                ],
            }
        ],
    },
    # Held by the library in its own right, which is what puts its scrobbles
    # inside the "artist the library holds" filter at all.
    {
        "name": "Pendulum",
        "albums": [{"title": "Immersion", "tracks": [{"title": "Watercolour"}]}],
    },
]


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") == 4

    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"


def test_albums_come_from_the_unfiltered_endpoint(tmp_path):
    """One request, and the only path that skips albums it cannot hydrate."""
    api = FakeLidarr(LIBRARY)
    store = store_at(tmp_path)

    music_curator.index_library(music_curator.Lidarr("http://lidarr", "key", transport=api), store)

    album_calls = [query for path, query in api.calls if path == "album"]
    assert album_calls == [{}]
    assert store.scalar("SELECT COUNT(*) FROM lidarr_album") == 4


def test_a_bad_album_falls_back_to_asking_per_artist(tmp_path):
    """An album with two monitored releases throws in the resource mapper, so
    the bulk call dies wholesale. Per artist, only its owner is lost."""
    # The bulk call fails because artist 2 owns the offending album.
    api = FakeLidarr(LIBRARY, fail=[("album", 0), ("album", 2)])
    store = store_at(tmp_path)

    music_curator.index_library(music_curator.Lidarr("http://lidarr", "key", transport=api), store)

    assert store.scalar("SELECT COUNT(*) FROM lidarr_album WHERE artist_id = 2") == 0
    # The other two artists keep their albums.
    assert store.scalar("SELECT COUNT(*) FROM lidarr_album") == 3
    assert store.get_state("index_albums_skipped") == "1"


def test_a_bad_album_does_not_cost_that_artist_their_tracks(tmp_path):
    """Tracks come from a different endpoint with a different mapper, so
    matching survives an album Lidarr cannot serialise."""
    api = FakeLidarr(LIBRARY, fail=[("album", 0), ("album", 2)])
    store = store_at(tmp_path)

    music_curator.index_library(music_curator.Lidarr("http://lidarr", "key", transport=api), store)

    assert store.scalar("SELECT COUNT(*) FROM lidarr_track WHERE artist_id = 2") == 1
    assert store.get_state("index_skipped") == "0"


def test_an_artist_lidarr_cannot_serve_does_not_kill_the_index(tmp_path):
    # Artist 2 is AC/DC in LIBRARY; its track lookup fails.
    api = FakeLidarr(LIBRARY, fail=[("track", 2)])
    store = store_at(tmp_path)

    music_curator.index_library(music_curator.Lidarr("http://lidarr", "key", transport=api), store)

    assert store.scalar("SELECT COUNT(*) FROM lidarr_artist") == 4
    assert store.scalar("SELECT COUNT(*) FROM lidarr_track WHERE artist_id = 2") == 0
    # The other two artists are indexed in full.
    assert store.scalar("SELECT COUNT(*) FROM lidarr_track") == 5
    assert store.get_state("index_skipped") == "1"


def test_a_skipped_artist_is_recorded_so_the_report_can_disown_the_numbers(tmp_path):
    """An incomplete index makes played music look cold. It has to be loud."""
    api = FakeLidarr(LIBRARY, fail=[("trackfile", 1)])
    store = store_at(tmp_path)

    music_curator.index_library(music_curator.Lidarr("http://lidarr", "key", transport=api), store)
    music_curator.match_library(store)

    assert store.get_state("index_skipped") == "1"


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

    assert store.get_state("index_skipped") == "0"


def test_a_lidarr_error_carries_the_url_and_what_the_server_said():
    """A bare status code sends you looking at the wrong thing entirely."""
    api = FakeLidarr(LIBRARY, fail=[("album", 0)])
    client = music_curator.Lidarr("http://lidarr:8686", "key", transport=api)

    with pytest.raises(music_curator.LidarrError) as raised:
        client.get("album")

    message = str(raised.value)
    assert "http://lidarr:8686/api/v1/album" in message
    assert "HTTP 500" in message
    assert "boom" in message


def test_the_offending_album_is_named_not_just_its_artist(tmp_path, caplog):
    """An artist's whole discography is too much to click through by hand."""
    # AC/DC is artist 2; its only album is id 201, holding "Hells Bells".
    api = FakeLidarr(LIBRARY, fail=[("album", 0), ("album", 2), ("albumid", 201)])
    store = store_at(tmp_path)

    with caplog.at_level("WARNING"):
        music_curator.index_library(
            music_curator.Lidarr("http://lidarr", "key", transport=api), store
        )

    assert "album id 201" in caplog.text
    assert "Hells Bells" in caplog.text


def test_a_transient_failure_is_not_probed_album_by_album(tmp_path, caplog):
    """When the resolver is the problem every artist fails, and probing each of
    them multiplies the load that caused it."""

    def unresolvable(url, timeout=None, headers=None):
        if "artistId" in url:
            raise urllib.error.URLError("[Errno -3] Try again")
        return json.dumps(FakeLidarr(LIBRARY).artists) if url.endswith("artist") else "[]"

    store = store_at(tmp_path)
    client = music_curator.Lidarr("http://lidarr", "key", transport=unresolvable, backoff=0)

    with caplog.at_level("WARNING"):
        music_curator.index_library(client, store)

    assert "Try again" in caplog.text
    assert "cannot serialise" not in caplog.text


def test_a_transient_failure_is_retried():
    attempts = []

    def flaky(url, timeout=None, headers=None):
        attempts.append(url)
        if len(attempts) < 3:
            raise urllib.error.URLError("[Errno -3] Try again")
        return "[]"

    client = music_curator.Lidarr("http://lidarr", "key", transport=flaky, backoff=0)

    assert client.get("artist") == []
    assert len(attempts) == 3


def test_a_lidarr_500_is_not_retried():
    """It is an exception inside Lidarr's serialisation, not a busy server."""
    api = FakeLidarr(LIBRARY, fail=[("album", 0)])
    client = music_curator.Lidarr("http://lidarr", "key", transport=api, backoff=0)

    with pytest.raises(music_curator.LidarrError) as raised:
        client.get("album")

    assert raised.value.transient is False
    assert len(api.calls) == 1


def test_keep_alive_uses_one_connection_for_many_requests(http_server):
    """The point of the whole class: one DNS lookup and one socket, not N."""
    server, base = http_server
    transport = music_curator.KeepAlive()

    try:
        for index in range(5):
            body = transport(f"{base}/api/v1/artist?n={index}", headers={"X-Api-Key": "key"})
            assert json.loads(body)["key"] == "key"
    finally:
        transport.close()

    assert server.connections == 1


def test_keep_alive_maps_an_error_status_onto_httperror(http_server):
    _, base = http_server
    transport = music_curator.KeepAlive()

    try:
        with pytest.raises(urllib.error.HTTPError) as raised:
            transport(f"{base}/boom", headers={"X-Api-Key": "key"})
        assert raised.value.code == 500
        assert music_curator.error_detail(raised.value) == "boom"
    finally:
        transport.close()


def test_lidarr_talks_to_a_real_server_through_keep_alive(http_server):
    server, base = http_server
    client = music_curator.Lidarr(base, "secret")

    try:
        assert client.get("artist", {"x": 1})["key"] == "secret"
        assert client.get("album")["path"] == "/api/v1/album"
    finally:
        client.transport.close()

    assert server.connections == 1


# Titles taken verbatim from a real coverage report's unmatched list. Each one
# was a genuine miss before the normaliser handled it.
@pytest.mark.parametrize(
    ("scrobbled", "tagged"),
    [
        ("Self vs Self (feat. In Flames)", "Self vs Self"),
        ("Grime Battle of Hastings (feat. The Town Crier)", "Grime Battle of Hastings"),
        ("Gold Dust - Shy FX Re-Edit", "Gold Dust"),
        ("Back To Your Roots - Friction & K-Tee Remix", "Back To Your Roots"),
        ("Constellations - Forza Horizon 3 VIP", "Constellations"),
        ("Everyday (Netsky Remix)", "Everyday"),
        ("Voodoo People [Pendulum Remix] [Live At Brixton Academy]", "Voodoo People"),
    ],
)
def test_real_unmatched_titles_now_agree_with_their_tags(scrobbled, tagged):
    assert music_curator.normalise(scrobbled) == music_curator.normalise(tagged)


@pytest.mark.parametrize(
    "title",
    [
        "Dancing with Myself",
        "(Don't Fear) The Reaper",
        "Live and Let Die",
        "Radio Ga Ga",
        "Editors",
        "Mixed Emotions",
        "Vipassana",
    ],
)
def test_the_version_words_do_not_eat_ordinary_titles(title):
    """Every one of these contains a version word and must survive intact."""
    assert music_curator.normalise(title) == music_curator.normalise(title.lower())
    assert len(music_curator.normalise(title).split()) == len(title.split())


def test_a_bracketed_guest_credit_matches_the_bare_tag(tmp_path):
    """Whitespace-then-feat misses the bracketed form, which is most of them."""
    store = indexed(tmp_path, [scrobble_of("Yellowcard", "Here I Am Alive (feat. Someone)")])

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


def test_misses_are_split_by_whether_the_library_holds_the_title(tmp_path):
    """Owning an artist is a weak proxy for owning a track. Counting both as
    matcher failures overstates the problem and would over-block the cull."""
    store = indexed(
        tmp_path,
        [
            # The library holds "Hells Bells", but under AC/DC, not Yellowcard:
            # an attribution disagreement, and a real miss.
            scrobble_of("Yellowcard", "Hells Bells"),
            # Yellowcard is in the library; this track is not, under any artist.
            scrobble_of("Yellowcard", "A Single She Never Bought"),
        ],
    )

    def count(extra):
        return store.connection.execute(
            "SELECT COUNT(*) 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)"
            f"  {extra}"
        ).fetchone()[0]

    assert count("") == 2
    assert count("AND EXISTS (SELECT 1 FROM lidarr_track t WHERE t.norm_title = k.norm_track)") == 1


def test_the_report_survives_the_attribution_split(tmp_path):
    store = indexed(tmp_path, [scrobble_of("Yellowcard", "Hells Bells")])

    music_curator.report(store, NOW)


def attribution_pairs(store):
    """Unmatched pairs where the library's own title credits the scrobbled artist."""
    return [
        row["artist"]
        for row in 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)"
            "   AND EXISTS (SELECT 1 FROM lidarr_track t"
            "                WHERE t.norm_title = k.norm_track"
            "                  AND instr(lower(t.title), lower(k.artist)) > 0)"
        )
    ]


def test_a_shared_title_is_not_an_attribution_miss(tmp_path):
    """Across fifty thousand tracks, titles collide constantly: "Everyday" is
    Rusko and also Def Leppard. Matching those would be worse than missing."""
    store = indexed(tmp_path, [scrobble_of("Yellowcard", "Hells Bells")])

    # The library holds "Hells Bells", by AC/DC, and its title says nothing
    # about Yellowcard. A collision, not a miss.
    assert verdict(store, "Yellowcard", "Hells Bells") == ("none", None)
    assert attribution_pairs(store) == []


def test_a_remix_credited_in_the_library_title_is_an_attribution_miss(tmp_path):
    """The library has "Voodoo People (Pendulum Remix)" under The Prodigy; the
    scrobble credits Pendulum. Same song, different filing."""
    store = indexed(tmp_path, [scrobble_of("Pendulum", "Voodoo People")])

    assert attribution_pairs(store) == ["Pendulum"]


def test_the_title_index_is_used_for_the_report_lookup(tmp_path):
    """Without it the report scans every track for every unmatched key: forty-
    seven seconds on a real library."""
    store = indexed(tmp_path, [])

    plan = "\n".join(
        row[-1]
        for row in store.connection.execute(
            "EXPLAIN QUERY PLAN SELECT 1 FROM scrobble_key k WHERE k.track_id IS NULL"
            " AND EXISTS (SELECT 1 FROM lidarr_track t WHERE t.norm_title = k.norm_track)"
        )
    )
    assert "lidarr_track_title" in plan, plan


def playlist_library(tmp_path):
    """A library on disk, with mirror MP3s beside the Lidarr source paths."""
    source = tmp_path / "music"
    mirror = tmp_path / "mirror"
    library = [
        {
            "name": "Played Band",
            "albums": [
                {
                    "title": "Known",
                    "tracks": [{"title": "Hit"}, {"title": "Album Track"}],
                }
            ],
        },
        {
            "name": "Silent Band",
            "albums": [{"title": "Unknown", "tracks": [{"title": "Never Heard"}]}],
        },
    ]
    api = FakeLidarr(library)
    # FakeLidarr invents /music/<artist>/<album>/<title>.flac; put the mirror
    # MP3s at the paths music-mirror would have produced from those.
    for handle in api.files:
        relative = Path(handle["path"]).relative_to("/music")
        handle["path"] = str(source / relative)
        mp3 = (mirror / relative).with_suffix(".mp3")
        mp3.parent.mkdir(parents=True, exist_ok=True)
        mp3.write_bytes(b"not really an mp3")
    for artist in api.artists:
        artist["path"] = str(source / Path(artist["path"]).name)
    return api, source, mirror


def test_playlists_are_written_into_the_mirror(tmp_path):
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    ingest(store, FakeLastfm([scrobble_of("Played Band", "Hit")] * 1))
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    music_curator.match_library(store)

    music_curator.build_playlists(store, mirror, str(source), 100, NOW)

    written = sorted(p.name for p in (mirror / "_playlists").glob("*.m3u"))
    assert written == [
        "all-time.m3u",
        "deep-cuts.m3u",
        "heavy-rotation.m3u",
        "neglected.m3u",
        "unheard-favourites.m3u",
        "unheard.m3u",
    ]

    played = (mirror / "_playlists" / "all-time.m3u").read_text().splitlines()
    assert played[0] == "#EXTM3U"
    assert played[1].startswith("#EXTINF:")
    assert "Played Band - Hit" in played[1]
    # Relative to the playlist file, so the same file works from any mount.
    assert played[2] == "../Played Band/Known/Hit.mp3"
    assert (mirror / "_playlists" / "all-time.m3u").parent.joinpath(played[2]).resolve().is_file()


def test_an_unplayed_track_lands_in_the_unheard_playlist(tmp_path):
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    ingest(store, FakeLastfm([scrobble_of("Played Band", "Hit")]))
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    music_curator.match_library(store)

    music_curator.build_playlists(store, mirror, str(source), 100, NOW)

    unheard = (mirror / "_playlists" / "unheard.m3u").read_text()
    assert "Never Heard" in unheard
    assert "Album Track" in unheard
    # The one thing that was played must not be in it.
    assert "Played Band - Hit\n" not in unheard


def test_a_track_with_no_mirror_file_is_left_out(tmp_path):
    """The index knows what Lidarr holds; whether music-mirror has encoded the
    MP3 yet is a different question, and is checked rather than assumed."""
    api, source, mirror = playlist_library(tmp_path)
    for mp3 in mirror.rglob("*.mp3"):
        mp3.unlink()
    store = store_at(tmp_path)
    ingest(store, FakeLastfm([scrobble_of("Played Band", "Hit")]))
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    music_curator.match_library(store)

    total = music_curator.build_playlists(store, mirror, str(source), 100, NOW)

    assert total == 0
    assert (mirror / "_playlists" / "all-time.m3u").read_text() == "#EXTM3U\n"


def test_the_library_root_is_derived_from_the_artist_folders(tmp_path):
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )

    assert music_curator.library_root_of(store) == str(source)


def test_the_playlist_limit_is_honoured(tmp_path):
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    music_curator.match_library(store)

    music_curator.build_playlists(store, mirror, str(source), 1, NOW)

    unheard = (mirror / "_playlists" / "unheard.m3u").read_text().splitlines()
    assert len([line for line in unheard if line.startswith("#EXTINF")]) == 1


def test_the_rotation_moves_weekly_not_every_pass(tmp_path):
    """A playlist that reorders on every pass is one that has to be re-imported
    on every pass; the Music app imports a snapshot, it does not track a file."""
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    music_curator.match_library(store)

    def unheard_at(when):
        music_curator.build_playlists(store, mirror, str(source), 100, when)
        return (mirror / "_playlists" / "unheard.m3u").read_text()

    same_week = unheard_at(NOW), unheard_at(NOW + 3600)
    assert same_week[0] == same_week[1]


def test_playlists_are_group_readable(tmp_path):
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    music_curator.match_library(store)

    music_curator.build_playlists(store, mirror, str(source), 100, NOW)

    for playlist in (mirror / "_playlists").glob("*.m3u"):
        assert playlist.stat().st_mode & stat.S_IRGRP, playlist


def test_tag_weights_fall_back_to_rank_when_no_count_is_sent():
    """The documented sample carries only a name and a URL; a live response also
    carries a count. Neither may be relied on alone."""
    with_count = music_curator.parse_tags(
        {"toptags": {"tag": [{"name": "Screamo", "count": 100}, {"name": "emo", "count": 40}]}}
    )
    assert with_count == [("screamo", 100), ("emo", 40)]

    without = music_curator.parse_tags(
        {"toptags": {"tag": [{"name": "screamo"}, {"name": "emo"}]}}
    )
    assert [tag for tag, _ in without] == ["screamo", "emo"]
    assert without[0][1] > without[1][1]


def test_a_lone_tag_is_not_a_list():
    assert music_curator.parse_tags({"toptags": {"tag": {"name": "dnb", "count": 90}}}) == [
        ("dnb", 90)
    ]


def test_an_artist_lastfm_cannot_answer_for_is_not_asked_again(tmp_path):
    """Recording the fetch even when it returns nothing is what stops a pass
    spending a request per unknown artist, forever."""
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    lastfm = FakeLastfm(tags={})

    assert music_curator.sync_tags(client_for(lastfm), store, NOW) == 2
    before = len(lastfm.calls)

    assert music_curator.sync_tags(client_for(lastfm), store, NOW) == 0
    assert len(lastfm.calls) == before


def test_tags_are_refetched_once_they_go_stale(tmp_path):
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    lastfm = FakeLastfm(tags={})
    music_curator.sync_tags(client_for(lastfm), store, NOW)

    later = NOW + music_curator.TAG_REFRESH_SECONDS + 1
    assert music_curator.sync_tags(client_for(lastfm), store, later) == 2


def tagged_store(tmp_path, tags):
    api, source, mirror = playlist_library(tmp_path)
    store = store_at(tmp_path)
    music_curator.index_library(
        music_curator.Lidarr("http://lidarr", "key", transport=api), store
    )
    music_curator.sync_tags(client_for(FakeLastfm(tags=tags)), store, NOW)
    return store, source, mirror


def test_a_vibe_selects_by_tag(tmp_path):
    store, source, mirror = tagged_store(
        tmp_path,
        {
            "artist-mbid-1": [{"name": "screamo", "count": 100}],
            "artist-mbid-2": [{"name": "classic rock", "count": 100}],
        },
    )
    vibes = [{"name": "screamo", "tags": ["screamo"]}]

    music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW)

    written = (mirror / "_playlists" / "screamo.m3u").read_text()
    assert "Played Band" in written
    assert "Silent Band" not in written


def test_a_weakly_tagged_artist_is_below_the_threshold(tmp_path):
    """A single low-weight tag is not a genre, it is somebody's stray opinion."""
    store, source, mirror = tagged_store(
        tmp_path, {"artist-mbid-1": [{"name": "screamo", "count": 3}]}
    )
    vibes = [{"name": "screamo", "tags": ["screamo"]}]

    music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW)

    assert (mirror / "_playlists" / "screamo.m3u").read_text() == "#EXTM3U\n"


def test_a_vibe_can_be_restricted_by_release_year(tmp_path):
    """What separates eighties synth records from everything else a synthpop
    tag drags in."""
    store, source, mirror = tagged_store(
        tmp_path, {"artist-mbid-1": [{"name": "synthpop", "count": 100}]}
    )
    inside = [{"name": "eighties", "tags": ["synthpop"], "years": [1975, 1992]}]
    outside = [{"name": "nineties", "tags": ["synthpop"], "years": [1993, 1999]}]

    # FakeLidarr dates every album 2019, so neither window should catch it.
    music_curator.build_vibe_playlists(store, inside, mirror, str(source), 100, NOW)
    music_curator.build_vibe_playlists(store, outside, mirror, str(source), 100, NOW)
    assert (mirror / "_playlists" / "eighties.m3u").read_text() == "#EXTM3U\n"
    assert (mirror / "_playlists" / "nineties.m3u").read_text() == "#EXTM3U\n"

    modern = [{"name": "modern", "tags": ["synthpop"], "years": [2000, 2030]}]
    music_curator.build_vibe_playlists(store, modern, mirror, str(source), 100, NOW)
    assert "Played Band" in (mirror / "_playlists" / "modern.m3u").read_text()


def test_the_built_in_vibes_are_all_usable_filenames():
    for vibe in music_curator.DEFAULT_VIBES:
        assert music_curator.SAFE_VIBE_NAME.fullmatch(vibe["name"]), vibe["name"]
        assert vibe["tags"]


def test_a_vibes_file_replaces_the_built_in_set(tmp_path):
    path = tmp_path / "vibes.json"
    path.write_text(json.dumps([{"name": "mine", "tags": ["shoegaze"]}]))

    assert music_curator.load_vibes(str(path)) == ({"name": "mine", "tags": ["shoegaze"]},)
    assert music_curator.load_vibes(None) is music_curator.DEFAULT_VIBES


@pytest.mark.parametrize(
    "content",
    [
        '{"not": "a list"}',
        '[{"tags": ["x"]}]',
        '[{"name": "../escape", "tags": ["x"]}]',
        '[{"name": "ok"}]',
        "not json at all",
    ],
)
def test_a_bad_vibes_file_is_refused_up_front(tmp_path, content):
    """A bad name would otherwise surface as a file written somewhere
    unintended, which is a poor way to learn about a typo."""
    path = tmp_path / "vibes.json"
    path.write_text(content)

    with pytest.raises(ValueError):
        music_curator.load_vibes(str(path))