Lyra/worker/tests/test_discovery_albums.py

from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource
from lyra_worker.browser import ReleaseGroupInfo
from lyra_worker.discovery import DiscoveryConfig, run_discovery
from lyra_worker.similarity.base import SimilarArtist
from tests.conftest import insert_discovery_suggestion, insert_monitored_release, insert_watched_artist


def _album_rows(conn):
    with conn.cursor() as cur:
        cur.execute('SELECT "artistMbid", "rgMbid", album, "primaryType" '
                    'FROM "DiscoverySuggestion" WHERE kind = \'album\' ORDER BY album')
        return cur.fetchall()


def _artist_rows(conn):
    with conn.cursor() as cur:
        cur.execute("SELECT \"artistMbid\" FROM \"DiscoverySuggestion\" "
                    "WHERE kind = 'artist' AND status = 'pending' ORDER BY \"artistMbid\"")
        return cur.fetchall()


def _reasons(conn):
    with conn.cursor() as cur:
        cur.execute('SELECT "rgMbid", "albumReason" FROM "DiscoverySuggestion" '
                    'WHERE kind = \'album\' ORDER BY "rgMbid"')
        return dict(cur.fetchall())


class FakePopularity:
    """Maps artist NAME -> [[album_name, mbid_or_None, playcount], …] (Last.fm shape)."""
    def __init__(self, data):
        self._d = data
    def top_albums(self, conn, name):
        return self._d.get(name, [])


def test_derives_most_recent_core_album_for_surfaced_artist(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-old", "Debut", "Album", (), "2001-01-01"),
        ReleaseGroupInfo("rg-new", "Latest", "Album", (), "2020-01-01"),
        ReleaseGroupInfo("rg-live", "At The Hall", "Album", ("Live",), "2021-01-01"),
    ]})
    result = run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1))
    assert result.albums == 1
    assert _album_rows(conn) == [("c1", "rg-new", "Latest", "Album")]  # newest core, live excluded


def test_album_skips_release_groups_already_in_library(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    insert_monitored_release(conn, artist_mbid="c1", rg_mbid="rg-have", album="Owned")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-have", "Owned", "Album", (), "2019-01-01"),
        ReleaseGroupInfo("rg-fresh", "Fresh", "Album", (), "2018-01-01"),
    ]})
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1))
    assert _album_rows(conn) == [("c1", "rg-fresh", "Fresh", "Album")]  # owned rg excluded


def test_derives_newest_and_most_popular_albums(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-hit", "Big Hit", "Album", (), "2005-01-01"),
        ReleaseGroupInfo("rg-new", "Latest", "Album", (), "2022-01-01"),
    ]})
    pop = FakePopularity({"Cand": [["Big Hit", "rg-hit", 90000], ["Latest", "rg-new", 100]]})
    result = run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1), popularity=pop)
    assert result.albums == 2
    assert _reasons(conn) == {"rg-new": "newest", "rg-hit": "popular"}


def test_newest_and_popular_the_same_album_is_one_row(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-only", "Only", "Album", (), "2022-01-01"),
    ]})
    pop = FakePopularity({"Cand": [["Only", "rg-only", 90000]]})
    result = run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1), popularity=pop)
    assert result.albums == 1
    assert _reasons(conn) == {"rg-only": "newest,popular"}


def test_popular_matches_by_title_when_lastfm_mbid_missing_and_skips_owned(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    insert_monitored_release(conn, artist_mbid="c1", rg_mbid="rg-owned", album="Owned Hit")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-owned", "Owned Hit", "Album", (), "2010-01-01"),  # excluded (owned)
        ReleaseGroupInfo("rg-match", "Second Hit", "Album", (), "2012-01-01"),
        ReleaseGroupInfo("rg-new", "Latest", "Album", (), "2022-01-01"),
    ]})
    # top album is owned (skipped), next has no mbid -> matched by title into core
    pop = FakePopularity({"Cand": [["Owned Hit", "rg-owned", 99999], ["Second Hit", None, 80000]]})
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1), popularity=pop)
    assert _reasons(conn) == {"rg-new": "newest", "rg-match": "popular"}


def test_rederiving_replaces_stale_pending_album_suggestions(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    # a stale pending album suggestion for c1 that is no longer a pick (e.g. an old single)
    insert_discovery_suggestion(conn, kind="album", artist_mbid="c1", rg_mbid="rg-old-single",
                                album="Old Single", status="pending")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-new", "New Album", "Album", (), "2022-01-01"),
    ]})
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1))
    # the stale single is gone; only the freshly-derived album remains
    assert [r[1] for r in _album_rows(conn)] == ["rg-new"]


def test_rederiving_keeps_wanted_album_suggestions(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    insert_discovery_suggestion(conn, kind="album", artist_mbid="c1", rg_mbid="rg-wanted",
                                album="Wanted One", status="wanted")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-new", "New Album", "Album", (), "2022-01-01"),
    ]})
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1))
    with conn.cursor() as cur:  # wanted row untouched (only pending stale rows are replaced)
        cur.execute('SELECT count(*) FROM "DiscoverySuggestion" WHERE "rgMbid" = %s AND status = \'wanted\'', ("rg-wanted",))
        assert cur.fetchone()[0] == 1


def test_drops_artist_with_no_core_discography(conn):
    # Two candidates: one real (has a core album), one a "producer" whose only release is a
    # secondary-typed soundtrack (like Max Martin's & Juliet) — the latter must be dropped.
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [
        SimilarArtist("real", "Real Artist", 0.9), SimilarArtist("producer", "Producer", 0.8)]})
    browser = FakeMbBrowser(releases={
        "real": [ReleaseGroupInfo("rg-r", "Record", "Album", (), "2020-01-01")],
        "producer": [ReleaseGroupInfo("rg-st", "A Musical", "Album", ("Soundtrack",), "2019-01-01")],
    })
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1))
    assert [r[0] for r in _artist_rows(conn)] == ["real"]  # producer dropped


def test_drops_artist_with_empty_discography(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("member", "Band Member", 0.9)]})
    browser = FakeMbBrowser(releases={"member": []})  # no release groups at all
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1))
    assert _artist_rows(conn) == []


def test_no_popularity_provider_yields_newest_only(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-old", "Old", "Album", (), "2005-01-01"),
        ReleaseGroupInfo("rg-new", "New", "Album", (), "2022-01-01"),
    ]})
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=1))  # popularity=None
    assert _reasons(conn) == {"rg-new": "newest"}


def test_albums_only_excludes_singles_and_eps_by_default(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-single", "A Single", "Single", (), "2022-01-01"),
        ReleaseGroupInfo("rg-ep", "An EP", "EP", (), "2021-01-01"),
        ReleaseGroupInfo("rg-album", "A Record", "Album", (), "2020-01-01"),
    ]})
    # albums_only defaults True: only the full-length Album is suggested
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=5))
    assert _album_rows(conn) == [("c1", "rg-album", "A Record", "Album")]


def test_albums_only_false_includes_singles_and_eps(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [
        ReleaseGroupInfo("rg-single", "A Single", "Single", (), "2022-01-01"),
        ReleaseGroupInfo("rg-album", "A Record", "Album", (), "2020-01-01"),
    ]})
    run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=5, albums_only=False))
    assert _album_rows(conn) == [
        ("c1", "rg-album", "A Record", "Album"),
        ("c1", "rg-single", "A Single", "Single"),
    ]


def test_albums_per_artist_zero_derives_none(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed")
    src = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Cand", 0.9)]})
    browser = FakeMbBrowser(releases={"c1": [ReleaseGroupInfo("rg1", "A", "Album", (), "2020")]})
    assert run_discovery(conn, [src], browser, DiscoveryConfig(albums_per_artist=0)).albums == 0
    assert _album_rows(conn) == []