Emma Thorpe
a7d16ca0b2
Build and publish container / build (pull_request) Successful in 2m40s
Playlists were being written owned by root. The image runs as root by default, deliberately, so that a bind-mounted dataset of any ownership stays writable -- but everything it writes then comes out root-owned, and a root-owned playlist inside a mirror owned by the apps account is unreadable to whatever serves it. The group-read bit does not help when the group is also root. Copy the ownership of the mirror the playlist is being written into. That needs no configuration, cannot drift from whatever the mirror actually is, and does nothing at all when the two already agree -- which is the case whenever the container is run with an explicit user. Applied to the temporary file before the rename, so a playlist is never briefly visible owned by the wrong account, and to the _playlists directory when this is the run that creates it. A chown that is refused is ignored rather than fatal: that only happens when not running as root, which is precisely the case where the ownership was already right.
1345 lines
48 KiB
Python
1345 lines
48 KiB
Python
import json
|
||
import os
|
||
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,
|
||
{
|
||
"Played Band": [{"name": "screamo", "count": 100}],
|
||
"Silent Band": [{"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, {"Played Band": [{"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, {"Played Band": [{"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))
|
||
|
||
|
||
def test_tags_are_looked_up_by_name_not_by_mbid(tmp_path):
|
||
"""Last.fm's mbid index is stale: it cannot find Devo or Escape the Fate by
|
||
one, though their pages plainly exist. Its name index can."""
|
||
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={"Played Band": [{"name": "screamo", "count": 90}]})
|
||
|
||
music_curator.sync_tags(client_for(lastfm), store, NOW)
|
||
|
||
asked = [call for call in lastfm.calls if call.get("method") == "artist.getTopTags"]
|
||
# The name is always tried first; the mbid only appears as a fallback for
|
||
# the artist that the name could not resolve.
|
||
assert "artist" in asked[0]
|
||
assert [call for call in asked if call.get("artist") == "Played Band"]
|
||
assert not [call for call in asked if call.get("mbid") == "artist-mbid-1"]
|
||
assert store.scalar("SELECT COUNT(*) FROM artist_tag WHERE tag = 'screamo'") == 1
|
||
|
||
|
||
def test_the_mbid_is_tried_when_the_name_is_not_found(tmp_path):
|
||
"""Kept only for a name Lidarr spells differently to Last.fm."""
|
||
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={"artist-mbid-1": [{"name": "dnb", "count": 80}]})
|
||
|
||
music_curator.sync_tags(client_for(lastfm), store, NOW)
|
||
|
||
assert store.scalar("SELECT COUNT(*) FROM artist_tag WHERE tag = 'dnb'") == 1
|
||
asked = [call for call in lastfm.calls if call.get("method") == "artist.getTopTags"]
|
||
assert any("mbid" in call for call in asked)
|
||
|
||
|
||
def test_an_artist_neither_key_resolves_is_recorded_and_not_retried(tmp_path):
|
||
"""Otherwise every pass spends a request on it again, for ever."""
|
||
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
|
||
spent = len(lastfm.calls)
|
||
|
||
assert music_curator.sync_tags(client_for(lastfm), store, NOW) == 0
|
||
assert len(lastfm.calls) == spent
|
||
|
||
|
||
def test_a_real_failure_is_not_recorded_so_the_next_pass_retries(tmp_path):
|
||
"""A rate limit is not the same as an artist not existing."""
|
||
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={"Played Band": [], "Silent Band": []},
|
||
outcomes=[{"error": 10, "message": "Invalid API key"}],
|
||
)
|
||
|
||
music_curator.sync_tags(client_for(lastfm), store, NOW)
|
||
|
||
# One artist failed hard and must still be pending.
|
||
assert store.scalar("SELECT COUNT(*) FROM artist_tag_fetched") == 1
|
||
|
||
|
||
# Tags that carry real weight in a real library, against tags that describe a
|
||
# passport, span most of the collection, or are somebody's artist name.
|
||
USELESS_TAGS = {
|
||
"american", "british", "australian", "canadian", "swedish", "dutch", "german",
|
||
"scottish", "english", "uk", "usa", "canada",
|
||
"rock", "electronic", "pop", "alternative", "metal ", "all", "heavy",
|
||
"female vocalists", "male vocalists", "female vocalist",
|
||
"my top songs", "cover", "covers", "not emo",
|
||
"green day", "paramore", "queen", "bon jovi", "shinedown", "aerosmith",
|
||
"journey", "fleetwood mac",
|
||
}
|
||
|
||
|
||
def test_no_mood_selects_on_a_useless_tag():
|
||
"""Nationality is not a sound; `rock` and `electronic` span most of the
|
||
library; and Last.fm's top tag for an artist is often their own name."""
|
||
for vibe in music_curator.DEFAULT_VIBES:
|
||
overlap = {tag.casefold() for tag in vibe["tags"]} & USELESS_TAGS
|
||
assert not overlap, f"{vibe['name']} selects on {overlap}"
|
||
|
||
|
||
def test_mood_names_are_unique():
|
||
names = [vibe["name"] for vibe in music_curator.DEFAULT_VIBES]
|
||
assert len(names) == len(set(names))
|
||
|
||
|
||
ROCK_TAGS = {
|
||
"classic rock", "hard rock", "blues rock", "southern rock", "arena rock",
|
||
"glam rock", "hair metal", "glam metal", "heavy metal", "metal", "art rock",
|
||
"psychedelic rock", "progressive rock", "rock and roll", "rock n roll",
|
||
}
|
||
|
||
|
||
def test_a_decade_tag_in_a_non_rock_mood_must_exclude_the_rock():
|
||
"""`80s` sits on Def Leppard and Bon Jovi as heavily as on Eurythmics, so a
|
||
mood that reaches for a decade without wanting rock has to say so. A mood
|
||
that does want it -- classic-rock reaching for 70s -- is exempt."""
|
||
for vibe in music_curator.DEFAULT_VIBES:
|
||
tags = {tag.casefold() for tag in vibe["tags"]}
|
||
decades = {"60s", "70s", "80s", "90s"} & tags
|
||
if not decades or tags & ROCK_TAGS:
|
||
continue
|
||
excluded = {tag.casefold() for tag in vibe.get("exclude", [])}
|
||
missing = {"hard rock", "hair metal"} - excluded
|
||
assert not missing, f"{vibe['name']} selects on {decades} without excluding {missing}"
|
||
|
||
|
||
def test_the_eighties_mood_covers_the_canon():
|
||
"""Depeche Mode, Duran Duran, Eurythmics and Frankie Goes to Hollywood --
|
||
checked against their live Last.fm tags. Three of the four carry "synth pop"
|
||
with a space; only one carries it without."""
|
||
synths = next(v for v in music_curator.DEFAULT_VIBES if v["name"] == "80s-synths")
|
||
tags = {t.casefold() for t in synths["tags"]}
|
||
for artist_tags in (
|
||
{"80s", "new wave", "pop", "female vocalists", "synth pop"}, # Eurythmics
|
||
{"80s", "new wave", "pop", "british", "dance"}, # Frankie
|
||
{"electronic", "synthpop", "new wave", "80s", "synth pop"}, # Depeche Mode
|
||
{"new wave", "80s", "pop", "synth pop", "rock"}, # Duran Duran
|
||
):
|
||
assert tags & artist_tags, artist_tags
|
||
assert synths["years"] == [1975, 1992]
|
||
|
||
|
||
def test_an_excluded_tag_drops_the_artist(tmp_path):
|
||
"""Weighting cannot separate the eighties synth acts from the eighties
|
||
stadium rock, because the tag they would be weighted on is the one they
|
||
share."""
|
||
store, source, mirror = tagged_store(
|
||
tmp_path,
|
||
{
|
||
"Played Band": [{"name": "80s", "count": 100}, {"name": "hard rock", "count": 90}],
|
||
"Silent Band": [{"name": "80s", "count": 100}, {"name": "synth pop", "count": 90}],
|
||
},
|
||
)
|
||
vibes = [{"name": "eighties", "tags": ["80s", "synth pop"], "exclude": ["hard rock"]}]
|
||
|
||
music_curator.build_vibe_playlists(store, vibes, mirror, str(source), 100, NOW)
|
||
|
||
written = (mirror / "_playlists" / "eighties.m3u").read_text()
|
||
assert "Silent Band" in written
|
||
assert "Played Band" not in written
|
||
|
||
|
||
def test_a_vibe_cannot_both_select_and_exclude_a_tag(tmp_path):
|
||
path = tmp_path / "vibes.json"
|
||
path.write_text(json.dumps([{"name": "x", "tags": ["80s"], "exclude": ["80s"]}]))
|
||
|
||
with pytest.raises(ValueError, match="selects on and excludes"):
|
||
music_curator.load_vibes(str(path))
|
||
|
||
|
||
def test_matching_ownership_is_a_no_op_when_it_already_agrees(tmp_path):
|
||
target = tmp_path / "file"
|
||
target.write_text("x")
|
||
|
||
assert music_curator.match_ownership(target, tmp_path) is False
|
||
|
||
|
||
def test_ownership_failure_is_tolerated(tmp_path):
|
||
"""Not permitted unless running as root -- which is exactly the case where
|
||
the ownership is already whatever the caller runs as."""
|
||
target = tmp_path / "file"
|
||
target.write_text("x")
|
||
|
||
# uid 0 from a non-root test process: refused, and must not raise.
|
||
assert music_curator.set_ownership(target, 0, 0) is False
|
||
|
||
|
||
def test_a_playlist_is_chowned_to_match_the_mirror(tmp_path, monkeypatch):
|
||
"""The image runs as root, so its output is root-owned, and a root-owned
|
||
playlist in an apps-owned mirror is unreadable to whatever serves it."""
|
||
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)
|
||
|
||
attempted = []
|
||
real_stat = music_curator.Path.stat
|
||
|
||
def pretend_mirror_is_owned_by_568(self, *args, **kwargs):
|
||
info = real_stat(self, *args, **kwargs)
|
||
if self == mirror:
|
||
return os.stat_result(
|
||
(info.st_mode, info.st_ino, info.st_dev, info.st_nlink, 568, 568,
|
||
info.st_size, int(info.st_atime), int(info.st_mtime), int(info.st_ctime))
|
||
)
|
||
return info
|
||
|
||
monkeypatch.setattr(music_curator.Path, "stat", pretend_mirror_is_owned_by_568)
|
||
monkeypatch.setattr(
|
||
music_curator.os, "chown", lambda p, u, g: attempted.append((str(p), u, g))
|
||
)
|
||
|
||
music_curator.build_playlists(store, mirror, str(source), 100, NOW)
|
||
|
||
assert attempted, "no ownership was applied"
|
||
assert all(tuple(owner) == (568, 568) for _, *owner in attempted)
|
||
# The temporary file, before the rename, never the finished playlist.
|
||
assert all(path.endswith(".part") or path.endswith("_playlists") for path, *_ in attempted)
|