Adds the per-kind generator modules and registers them with the
PlaylistKindRegistry so the manager's `refresh_playlist` can dispatch
to any of them.
Generators (each in its own module under
`core/personalized/generators/`):
Singletons (variant=''):
- hidden_gems -> wraps service.get_hidden_gems
- discovery_shuffle -> wraps service.get_discovery_shuffle
- popular_picks -> wraps service.get_popular_picks
Variant-bearing kinds:
- time_machine -> variant = decade label ('1980s', '1990s', ...).
Variant resolver returns 7 standard decades.
Generator parses '1980s' -> 1980 + delegates
to service.get_decade_playlist.
- genre_playlist -> variant = URL-safe genre key
('electronic_dance', 'hip_hop_rap', ...).
Resolver normalizes parent-genre keys from
service.GENRE_MAPPING; free-form keywords
pass through to service.get_genre_playlist.
- daily_mix -> variant = top-genre rank ('1' / '2' / '3' / '4').
Generator looks up user's Nth-ranked library
genre and returns discovery picks within it.
Library half (was a stub returning []) is
intentionally dropped: tracks table has no
source IDs, so library rows can't sync. Fixed
the stub to return [] cleanly without the
misleading log warning.
- fresh_tape -> Spotify Release Radar. Reads curated track
IDs from discovery_curated_playlists (tries
'release_radar_<source>' first, falls back to
'release_radar') and hydrates against the
discovery pool.
- archives -> Spotify Discover Weekly. Same hydration path
as fresh_tape but uses 'discovery_weekly'.
- seasonal_mix -> variant = season key ('halloween' / 'christmas'
/ 'valentines' / 'summer' / 'spring' / 'autumn').
Reads curated IDs via SeasonalDiscoveryService
then hydrates from seasonal_tracks (which
carries full track_data_json).
Each module:
- Defines `generate(deps, variant, config) -> List[Track]`.
- Defines `SPEC = PlaylistKindSpec(...)` and registers it on import
(idempotent — re-import safe via `if registry.get(...) is None`).
- For variant-bearing kinds, also defines `variant_resolver(deps)`.
Shared helpers in `_common.py`:
- `get_service(deps)` pulls the legacy
`PersonalizedPlaylistsService` instance (deps.service or
deps['service']).
- `coerce_tracks(rows)` runs each dict through `Track.from_dict`,
tolerates None / non-list inputs.
Tests (50 new, total 85 across personalized subsystem):
- Singletons: registration + display name + dispatch + limit
forwarding + empty/None tolerance + missing-deps error +
dict-form deps acceptance (16 tests).
- Variants: variant_resolver listing + label parsing + invalid
variant errors + parent-key normalization + free-form passthrough
(13 tests).
- Curated/hybrid: daily_mix rank-to-genre resolution + rank-out-of-
range empty + invalid-variant error; fresh_tape & archives
hydration order + missing-id skip + source-specific-then-fallback
key dispatch + limit + missing-database-dep error; seasonal_mix
curated-id hydration order + missing-id skip + JSON round-trip +
empty-curated empty + limit + missing-service error (21 tests).
3304+ tests pass. No regression on existing 62 personalized tests.
147 lines
5.5 KiB
Python
147 lines
5.5 KiB
Python
"""Boundary tests for the singleton-kind personalized generators
|
|
(`hidden_gems`, `discovery_shuffle`, `popular_picks`).
|
|
|
|
Each generator wraps the legacy
|
|
``PersonalizedPlaylistsService`` method 1:1, so the tests pin:
|
|
- registration side-effect at import
|
|
- generator forwards `config.limit` correctly
|
|
- empty / None / non-dict service output → []
|
|
- tracks coerced through `Track.from_dict`
|
|
- missing service in deps raises a clear error"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import Any, List
|
|
|
|
import pytest
|
|
|
|
# Importing each generator triggers registration as a side-effect.
|
|
from core.personalized.generators import discovery_shuffle as _ds_mod
|
|
from core.personalized.generators import hidden_gems as _hg_mod
|
|
from core.personalized.generators import popular_picks as _pp_mod
|
|
from core.personalized.specs import get_registry
|
|
from core.personalized.types import PlaylistConfig
|
|
|
|
|
|
class _StubService:
|
|
"""Records every call so tests can assert on dispatched limits."""
|
|
|
|
def __init__(self, return_value=None):
|
|
self.calls: List[dict] = []
|
|
self.return_value = return_value if return_value is not None else []
|
|
|
|
def get_hidden_gems(self, limit):
|
|
self.calls.append({'method': 'get_hidden_gems', 'limit': limit})
|
|
return self.return_value
|
|
|
|
def get_discovery_shuffle(self, limit):
|
|
self.calls.append({'method': 'get_discovery_shuffle', 'limit': limit})
|
|
return self.return_value
|
|
|
|
def get_popular_picks(self, limit):
|
|
self.calls.append({'method': 'get_popular_picks', 'limit': limit})
|
|
return self.return_value
|
|
|
|
|
|
def _deps(svc):
|
|
return SimpleNamespace(service=svc)
|
|
|
|
|
|
# ─── registration ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestRegistration:
|
|
def test_hidden_gems_registered(self):
|
|
spec = get_registry().get('hidden_gems')
|
|
assert spec is not None
|
|
assert spec.kind == 'hidden_gems'
|
|
assert spec.requires_variant is False
|
|
assert spec.default_config.limit == 50
|
|
|
|
def test_discovery_shuffle_registered(self):
|
|
spec = get_registry().get('discovery_shuffle')
|
|
assert spec is not None
|
|
assert spec.requires_variant is False
|
|
|
|
def test_popular_picks_registered(self):
|
|
spec = get_registry().get('popular_picks')
|
|
assert spec is not None
|
|
assert spec.requires_variant is False
|
|
|
|
def test_display_names(self):
|
|
assert get_registry().get('hidden_gems').display_name('') == 'Hidden Gems'
|
|
assert get_registry().get('discovery_shuffle').display_name('') == 'Discovery Shuffle'
|
|
assert get_registry().get('popular_picks').display_name('') == 'Popular Picks'
|
|
|
|
|
|
# ─── generator dispatch ──────────────────────────────────────────────
|
|
|
|
|
|
class TestHiddenGemsGenerator:
|
|
def test_forwards_limit(self):
|
|
svc = _StubService()
|
|
_hg_mod.generate(_deps(svc), '', PlaylistConfig(limit=75))
|
|
assert svc.calls == [{'method': 'get_hidden_gems', 'limit': 75}]
|
|
|
|
def test_uses_default_limit_when_config_default(self):
|
|
svc = _StubService()
|
|
_hg_mod.generate(_deps(svc), '', PlaylistConfig())
|
|
assert svc.calls[0]['limit'] == 50
|
|
|
|
def test_coerces_tracks(self):
|
|
svc = _StubService(return_value=[
|
|
{'track_name': 'A', 'artist_name': 'X', 'spotify_track_id': 'sp-1'},
|
|
{'track_name': 'B', 'artist_name': 'Y', 'spotify_track_id': 'sp-2'},
|
|
])
|
|
out = _hg_mod.generate(_deps(svc), '', PlaylistConfig())
|
|
assert len(out) == 2
|
|
assert out[0].track_name == 'A'
|
|
assert out[0].spotify_track_id == 'sp-1'
|
|
|
|
def test_empty_service_output_returns_empty_list(self):
|
|
svc = _StubService(return_value=[])
|
|
out = _hg_mod.generate(_deps(svc), '', PlaylistConfig())
|
|
assert out == []
|
|
|
|
def test_none_service_output_returns_empty_list(self):
|
|
svc = _StubService(return_value=None)
|
|
out = _hg_mod.generate(_deps(svc), '', PlaylistConfig())
|
|
assert out == []
|
|
|
|
|
|
class TestDiscoveryShuffleGenerator:
|
|
def test_forwards_limit(self):
|
|
svc = _StubService()
|
|
_ds_mod.generate(_deps(svc), '', PlaylistConfig(limit=42))
|
|
assert svc.calls == [{'method': 'get_discovery_shuffle', 'limit': 42}]
|
|
|
|
def test_coerces_tracks(self):
|
|
svc = _StubService(return_value=[{'track_name': 'Z', 'artist_name': 'Q'}])
|
|
out = _ds_mod.generate(_deps(svc), '', PlaylistConfig())
|
|
assert out[0].track_name == 'Z'
|
|
|
|
|
|
class TestPopularPicksGenerator:
|
|
def test_forwards_limit(self):
|
|
svc = _StubService()
|
|
_pp_mod.generate(_deps(svc), '', PlaylistConfig(limit=10))
|
|
assert svc.calls == [{'method': 'get_popular_picks', 'limit': 10}]
|
|
|
|
|
|
# ─── deps validation ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestDepsValidation:
|
|
def test_missing_service_raises(self):
|
|
# No `service` attribute on deps.
|
|
deps = SimpleNamespace()
|
|
with pytest.raises(RuntimeError, match='missing `service`'):
|
|
_hg_mod.generate(deps, '', PlaylistConfig())
|
|
|
|
def test_dict_form_deps_accepted(self):
|
|
# generators._common.get_service tolerates dict deps too.
|
|
svc = _StubService()
|
|
out = _hg_mod.generate({'service': svc}, '', PlaylistConfig())
|
|
assert isinstance(out, list)
|
|
assert svc.calls
|