Initial project skeleton for mopidy-smartplaylists
Some checks failed
CI / Build (push) Failing after 18m47s
CI / pyright (push) Failing after 3m39s
CI / pytest (3.13) (push) Failing after 6s
CI / ruff check (push) Failing after 6s
CI / ruff format (push) Failing after 7s
CI / pytest (3.14) (push) Failing after 6s

This commit is contained in:
2026-06-24 19:11:02 -04:00
commit a054575dea
17 changed files with 1202 additions and 0 deletions

8
.copier-answers.yml Normal file
View File

@ -0,0 +1,8 @@
_commit: v2.3.0
_src_path: gh:mopidy/mopidy-ext-template
author_email: chris@code.unbl.ink
author_full_name: Chris Powell
dist_name: mopidy-smartplaylists
ext_name: smartplaylists
github_username: secstate
short_description: Mopidy extension for generating smart playlists based on track properties

62
.github/workflows/ci.yml vendored Normal file
View File

@ -0,0 +1,62 @@
name: CI
on:
pull_request:
push:
branches:
- main
workflow_dispatch:
jobs:
build:
name: Build
runs-on: ubuntu-24.04
steps:
- uses: actions/checkout@v6
- uses: hynek/build-and-inspect-python-package@v2
main:
strategy:
fail-fast: false
matrix:
include:
- name: "pytest (3.13)"
python: "3.13"
tox: "3.13"
- name: "pytest (3.14)"
python: "3.14"
tox: "3.14"
coverage: true
- name: "pyright"
python: "3.14"
tox: "pyright"
- name: "ruff check"
python: "3.14"
tox: "ruff-check"
- name: "ruff format"
python: "3.14"
tox: "ruff-format"
name: ${{ matrix.name }}
runs-on: ubuntu-24.04
container: ghcr.io/mopidy/ci:latest
steps:
- uses: actions/checkout@v6
- name: Fix home dir permissions to enable pip caching
run: chown -R root /github/home
- uses: actions/setup-python@v6
with:
python-version: ${{ matrix.python }}
cache: pip
allow-prereleases: true
- run: python -m pip install tox
- run: python -m pip install mopidy
- run: python -m tox -e ${{ matrix.tox }}
if: ${{ ! matrix.coverage }}
- run: python -m tox -e ${{ matrix.tox }} -- --cov-report=xml
if: ${{ matrix.coverage }}
- uses: codecov/codecov-action@v6
if: ${{ matrix.coverage }}
with:
token: ${{ secrets.CODECOV_TOKEN }}

16
.github/workflows/release.yml vendored Normal file
View File

@ -0,0 +1,16 @@
name: Release
on:
release:
types:
- released
jobs:
release:
runs-on: ubuntu-24.04
permissions:
id-token: write
steps:
- uses: actions/checkout@v6
- uses: hynek/build-and-inspect-python-package@v2
- uses: pypa/gh-action-pypi-publish@v1.12.4

9
.gitignore vendored Normal file
View File

@ -0,0 +1,9 @@
*.egg-info/
/*.lock
/.*_cache/
/.coverage
/.tox/
/.venv/
/build/
/dist/
__pycache__/

1
.mailmap Normal file
View File

@ -0,0 +1 @@
Chris Powell <chris@code.unbl.ink>

11
LICENSE Normal file
View File

@ -0,0 +1,11 @@
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

140
pyproject.toml Normal file
View File

@ -0,0 +1,140 @@
[project]
name = "mopidy-smartplaylists"
description = "Mopidy extension for generating smart playlists based on track properties"
readme = "README.md"
requires-python = ">= 3.13"
license = { text = "Apache-2.0" }
authors = [{ name = "Chris Powell", email = "chris@code.unbl.ink" }]
classifiers = [
"Environment :: No Input/Output (Daemon)",
"Intended Audience :: End Users/Desktop",
"License :: OSI Approved :: Apache Software License",
"Operating System :: OS Independent",
"Topic :: Multimedia :: Sound/Audio :: Players",
]
dynamic = ["version"]
dependencies = [
"mopidy >= 4.0.0",
"pykka >= 4.1",
"tornado >= 6.4",
]
[project.urls]
Homepage = "https://github.com/secstate/mopidy-smartplaylists"
[project.entry-points."mopidy.ext"]
smartplaylists = "mopidy_smartplaylists:Extension"
[build-system]
requires = ["setuptools >= 78", "setuptools-scm >= 8.2"]
build-backend = "setuptools.build_meta"
[dependency-groups]
dev = [
"tox",
{ include-group = "ruff" },
{ include-group = "tests" },
{ include-group = "typing" },
]
ruff = ["ruff"]
tests = ["pytest", "pytest-cov"]
typing = ["pyright"]
[tool.coverage.paths]
source = ["src/", "*/site-packages/"]
[tool.coverage.run]
source_pkgs = ["mopidy_smartplaylists"]
[tool.coverage.report]
show_missing = true
[tool.pyright]
pythonVersion = "3.13"
typeCheckingMode = "standard"
reportMissingTypeStubs = false
reportPrivateImportUsage = false
[tool.pytest.ini_options]
filterwarnings = [
"error:::mopidy_smartplaylists",
]
[tool.ruff]
target-version = "py313"
[tool.ruff.lint]
select = ["ALL"]
ignore = [
"D",
"D203",
"D213",
"FIX002",
"G004",
"S101",
"TD002",
"TD003",
"COM812",
"ISC001",
]
[tool.ruff.lint.per-file-ignores]
"tests/*" = [
"ANN",
"ARG",
"D",
"FBT003",
"PLR2004",
"S101",
"SLF001",
]
[tool.setuptools.package-data]
"*" = ["*.conf"]
[tool.setuptools_scm]
[tool.tox]
env_list = [
"3.13",
"3.14",
"pyright",
"ruff-check",
"ruff-format",
]
[tool.tox.env_run_base]
package = "wheel"
wheel_build_env = ".pkg"
dependency_groups = ["tests"]
commands = [
[
"pytest",
"--cov",
"--basetemp={envtmpdir}",
{ replace = "posargs", extend = true },
],
]
[tool.tox.env.pyright]
dependency_groups = ["typing"]
commands = [["pyright", "{posargs:src}"]]
[tool.tox.env.ruff-check]
skip_install = true
dependency_groups = ["ruff"]
commands = [["ruff", "check", "{posargs:.}"]]
[tool.tox.env.ruff-format]
skip_install = true
dependency_groups = ["ruff"]
commands = [["ruff", "format", "--check", "--diff", "{posargs:.}"]]

View File

@ -0,0 +1,40 @@
import pathlib
from importlib.metadata import version
from mopidy import config, ext
from mopidy.config import ConfigSchema
__version__ = version("mopidy-smartplaylists")
class Extension(ext.Extension):
dist_name = "mopidy-smartplaylists"
ext_name = "smartplaylists"
version = __version__
def get_default_config(self):
return config.read(pathlib.Path(__file__).parent / "ext.conf")
def get_config_schema(self):
schema = super().get_config_schema()
schema["decades"] = config.String(optional=True)
schema["genres"] = config.String(optional=True)
schema["artists"] = config.String(optional=True)
schema["playlist_prefix"] = config.String(optional=True)
schema["refresh_interval"] = config.Integer(optional=True, minimum=0)
return schema
def validate_environment(self):
pass
def setup(self, registry):
from mopidy_smartplaylists.web import app_factory
registry.add("http:app", {
"name": self.ext_name,
"factory": app_factory,
})
from mopidy_smartplaylists.frontend import SmartPlaylistsFrontend
registry.add("frontend", SmartPlaylistsFrontend)

View File

@ -0,0 +1,17 @@
[smartplaylists]
enabled = true
# Comma-separated list of decades to auto-generate playlists for on startup.
# Example: 1960,1970,1980,1990,2000,2010,2020
# Leave empty to disable auto-generation of decade mixes.
decades =
# Comma-separated list of genres to auto-generate playlists for on startup.
# Example: Jazz,Classical,Rock,Electronic,Hip Hop
# Leave empty to disable auto-generation of genre mixes.
genres =
# Comma-separated list of artists to auto-generate playlists for on startup.
# Leave empty to disable auto-generation of artist mixes.
artists =
# Prefix used for all smart playlist names
playlist_prefix = [Smart]
# Refresh interval in hours (0 = only on startup)
refresh_interval = 0

View File

@ -0,0 +1,35 @@
import logging
import pykka
from mopidy.config import Config
from mopidy.core import CoreListener, CoreProxy
from mopidy_smartplaylists.generators import refresh_smart_playlists
logger = logging.getLogger(__name__)
class SmartPlaylistsFrontend(pykka.ThreadingActor, CoreListener):
def __init__(self, config: Config, core: CoreProxy) -> None:
super().__init__()
self.config = config
self.core = core
def on_start(self) -> None:
section = dict(self.config.get("smartplaylists", {}))
has_recipes = any(section.get(k) for k in ("decades", "genres", "artists"))
if has_recipes:
logger.info("Generating smart playlists from config recipes...")
refresh_smart_playlists(self.core, section)
else:
logger.debug("No smart playlist recipes configured")
def on_stop(self) -> None:
pass
def playlists_loaded(self) -> None:
section = dict(self.config.get("smartplaylists", {}))
interval = int(section.get("refresh_interval", 0))
if interval > 0:
logger.info("Smart playlists loaded, running refresh...")
refresh_smart_playlists(self.core, section)

View File

@ -0,0 +1,159 @@
import logging
import re
from mopidy.core import CoreProxy
from mopidy.models import Playlist, Track, TrackField
logger = logging.getLogger(__name__)
def parse_decade(decade_str: str) -> str:
decade_str = decade_str.strip()
if len(decade_str) == 4:
return decade_str[:3] + "*"
if decade_str.endswith("s"):
return decade_str[:3] + "*"
return decade_str
def build_decade_mix(core: CoreProxy, decade: str) -> list[Track]:
query = {TrackField.DATE: [parse_decade(decade)]}
result = core.library.search(query).get()
tracks = _extract_tracks(result)
logger.info("Found %d tracks for decade %s", len(tracks), decade)
return tracks
def build_genre_mix(core: CoreProxy, genre: str) -> list[Track]:
query = {TrackField.GENRE: [genre]}
result = core.library.search(query).get()
tracks = _extract_tracks(result)
logger.info("Found %d tracks for genre %s", len(tracks), genre)
return tracks
def build_artist_mix(core: CoreProxy, artist: str) -> list[Track]:
query = {TrackField.ARTIST: [artist]}
result = core.library.search(query).get()
tracks = _extract_tracks(result)
logger.info("Found %d tracks for artist %s", len(tracks), artist)
return tracks
def build_album_mix(core: CoreProxy, album_uri: str) -> list[Track]:
tracks = core.library.lookup([album_uri]).get()
flat = _flatten_lookup(tracks)
logger.info("Found %d tracks for album %s", len(flat), album_uri)
return flat
def build_instant_mix(core: CoreProxy, track_uri: str, limit: int = 50) -> list[Track]:
lookup_result = core.library.lookup([track_uri]).get()
seed_tracks = _flatten_lookup(lookup_result)
if not seed_tracks:
logger.warning("No track found for URI: %s", track_uri)
return []
seed = seed_tracks[0]
genres: list[str] = []
artists: list[str] = []
if seed.genre:
genres = [g.strip() for g in seed.genre.split("/") if g.strip()]
if seed.artists:
for artist_ref in seed.artists:
if artist_ref.name:
artists.append(artist_ref.name)
similar: dict[str, Track] = {}
for genre in genres:
query = {TrackField.GENRE: [genre]}
result = core.library.search(query).get()
for batch in result:
for t in batch.tracks:
if t.uri and t.uri != track_uri:
similar[t.uri] = t
for artist in artists:
if len(similar) >= limit:
break
query = {TrackField.ARTIST: [artist]}
result = core.library.search(query).get()
for batch in result:
for t in batch.tracks:
if t.uri and t.uri != track_uri:
similar[t.uri] = t
tracks = list(similar.values())[:limit]
logger.info("Instant mix: found %d tracks similar to %s", len(tracks), seed.name)
return tracks
def save_smart_playlist(
core: CoreProxy,
prefix: str,
name: str,
tracks: list[Track],
) -> Playlist | None:
playlist_name = f"{prefix} {name}"
uri = f"mopidy:smartplaylists:{_sanitize_name(name)}"
existing = core.playlists.lookup(uri).get()
if existing:
core.playlists.delete(existing.uri).get()
playlist = Playlist(
name=playlist_name,
uri=uri,
tracks=tracks,
)
saved = core.playlists.save(playlist).get()
logger.info("Saved smart playlist: %s (%d tracks)", playlist_name, len(tracks))
return saved
def refresh_smart_playlists(core: CoreProxy, config_dict: dict) -> None:
prefix = config_dict.get("playlist_prefix", "[Smart]")
decades_raw = config_dict.get("decades", "")
if decades_raw:
decades = [d.strip() for d in decades_raw.split(",") if d.strip()]
for decade in decades:
tracks = build_decade_mix(core, decade)
if tracks:
save_smart_playlist(core, prefix, f"{decade}s Mix", tracks)
genres_raw = config_dict.get("genres", "")
if genres_raw:
genres = [g.strip() for g in genres_raw.split(",") if g.strip()]
for genre in genres:
tracks = build_genre_mix(core, genre)
if tracks:
save_smart_playlist(core, prefix, f"{genre} Mix", tracks)
artists_raw = config_dict.get("artists", "")
if artists_raw:
artists = [a.strip() for a in artists_raw.split(",") if a.strip()]
for artist in artists:
tracks = build_artist_mix(core, artist)
if tracks:
save_smart_playlist(core, prefix, f"{artist} Mix", tracks)
def _extract_tracks(search_result) -> list[Track]:
tracks: list[Track] = []
for batch in search_result:
tracks.extend(batch.tracks)
return tracks
def _flatten_lookup(lookup_result: dict[str, list[Track]]) -> list[Track]:
tracks: list[Track] = []
for uri_tracks in lookup_result.values():
tracks.extend(uri_tracks)
return tracks
def _sanitize_name(name: str) -> str:
return re.sub(r"[^a-zA-Z0-9_-]", "_", name).lower()

View File

@ -0,0 +1,204 @@
import json
import logging
import tornado.web
from mopidy.config import Config
from mopidy.core import CoreProxy
from mopidy_smartplaylists.generators import (
build_album_mix,
build_artist_mix,
build_decade_mix,
build_genre_mix,
build_instant_mix,
refresh_smart_playlists,
save_smart_playlist,
)
logger = logging.getLogger(__name__)
class DecadeMixHandler(tornado.web.RequestHandler):
def initialize(self, core: CoreProxy, prefix: str) -> None:
self.core = core
self.prefix = prefix
def post(self) -> None:
data = json.loads(self.request.body)
decade = data.get("decade", "")
if not decade:
self.set_status(400)
self.write({"error": "Missing 'decade' in request body"})
return
tracks = build_decade_mix(self.core, decade)
if not tracks:
self.write({"playlist": None, "tracks": 0})
return
playlist = save_smart_playlist(self.core, self.prefix, f"{decade}s Mix", tracks)
self.write({
"playlist": {
"name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None,
"tracks": len(tracks),
},
})
class GenreMixHandler(tornado.web.RequestHandler):
def initialize(self, core: CoreProxy, prefix: str) -> None:
self.core = core
self.prefix = prefix
def post(self) -> None:
data = json.loads(self.request.body)
genre = data.get("genre", "")
if not genre:
self.set_status(400)
self.write({"error": "Missing 'genre' in request body"})
return
tracks = build_genre_mix(self.core, genre)
if not tracks:
self.write({"playlist": None, "tracks": 0})
return
playlist = save_smart_playlist(self.core, self.prefix, f"{genre} Mix", tracks)
self.write({
"playlist": {
"name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None,
"tracks": len(tracks),
},
})
class ArtistMixHandler(tornado.web.RequestHandler):
def initialize(self, core: CoreProxy, prefix: str) -> None:
self.core = core
self.prefix = prefix
def post(self) -> None:
data = json.loads(self.request.body)
artist = data.get("artist", "")
if not artist:
self.set_status(400)
self.write({"error": "Missing 'artist' in request body"})
return
tracks = build_artist_mix(self.core, artist)
if not tracks:
self.write({"playlist": None, "tracks": 0})
return
playlist = save_smart_playlist(self.core, self.prefix, f"{artist} Mix", tracks)
self.write({
"playlist": {
"name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None,
"tracks": len(tracks),
},
})
class AlbumMixHandler(tornado.web.RequestHandler):
def initialize(self, core: CoreProxy, prefix: str) -> None:
self.core = core
self.prefix = prefix
def post(self) -> None:
data = json.loads(self.request.body)
album_uri = data.get("uri", "")
if not album_uri:
self.set_status(400)
self.write({"error": "Missing 'uri' in request body"})
return
tracks = build_album_mix(self.core, album_uri)
if not tracks:
self.write({"playlist": None, "tracks": 0})
return
name = tracks[0].album.name if tracks[0].album and tracks[0].album.name else "Album"
playlist = save_smart_playlist(self.core, self.prefix, f"{name} Mix", tracks)
self.write({
"playlist": {
"name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None,
"tracks": len(tracks),
},
})
class InstantMixHandler(tornado.web.RequestHandler):
def initialize(self, core: CoreProxy, prefix: str) -> None:
self.core = core
self.prefix = prefix
def post(self) -> None:
data = json.loads(self.request.body)
track_uri = data.get("uri", "")
limit = data.get("limit", 50)
if not track_uri:
self.set_status(400)
self.write({"error": "Missing 'uri' in request body"})
return
tracks = build_instant_mix(self.core, track_uri, limit)
if not tracks:
self.write({"playlist": None, "tracks": 0})
return
lookup_result = self.core.library.lookup([track_uri]).get()
seed_name = "Instant Mix"
for uri_tracks in lookup_result.values():
for t in uri_tracks:
if t.name:
seed_name = t.name
break
playlist = save_smart_playlist(self.core, self.prefix, f"Instant Mix: {seed_name}", tracks)
self.write({
"playlist": {
"name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None,
"tracks": len(tracks),
},
})
class RefreshHandler(tornado.web.RequestHandler):
def initialize(self, core: CoreProxy, config: Config) -> None:
self.core = core
self.config = config
def post(self) -> None:
section = self.config.get("smartplaylists", {})
refresh_smart_playlists(self.core, section)
self.write({"status": "ok"})
class StatusHandler(tornado.web.RequestHandler):
def initialize(self, core: CoreProxy) -> None:
self.core = core
def get(self) -> None:
result = self.core.playlists.as_list().get()
smart = [p for p in result if p.uri and "smartplaylists" in p.uri]
self.write({
"smart_playlists": [
{
"name": p.name,
"uri": p.uri,
"tracks": len(p.tracks) if p.tracks else 0,
}
for p in smart
],
"count": len(smart),
})
def app_factory(config: Config, core: CoreProxy) -> list[tuple]:
prefix = config.get("smartplaylists", {}).get("playlist_prefix", "[Smart]")
return [
(r"/decade", DecadeMixHandler, {"core": core, "prefix": prefix}),
(r"/genre", GenreMixHandler, {"core": core, "prefix": prefix}),
(r"/artist", ArtistMixHandler, {"core": core, "prefix": prefix}),
(r"/album", AlbumMixHandler, {"core": core, "prefix": prefix}),
(r"/instant-mix", InstantMixHandler, {"core": core, "prefix": prefix}),
(r"/refresh", RefreshHandler, {"core": core, "config": config}),
(r"/status", StatusHandler, {"core": core}),
]

0
tests/__init__.py Normal file
View File

48
tests/test_extension.py Normal file
View File

@ -0,0 +1,48 @@
from unittest import mock
from mopidy_smartplaylists import Extension
from mopidy_smartplaylists import frontend as frontend_lib
from mopidy_smartplaylists import web as web_lib
def test_get_default_config():
ext = Extension()
config = ext.get_default_config()
assert "[smartplaylists]" in config
assert "enabled = true" in config
assert "decades =" in config
assert "genres =" in config
assert "artists =" in config
assert "playlist_prefix = [Smart]" in config
assert "refresh_interval = 0" in config
def test_get_config_schema():
ext = Extension()
schema = ext.get_config_schema()
assert "decades" in schema
assert "genres" in schema
assert "artists" in schema
assert "playlist_prefix" in schema
assert "refresh_interval" in schema
def test_setup_registers_http_app():
ext = Extension()
registry = mock.Mock()
ext.setup(registry)
calls = [c.args for c in registry.add.call_args_list]
http_app_calls = [c for c in calls if c[0] == "http:app"]
assert len(http_app_calls) == 1
assert http_app_calls[0][1]["name"] == "smartplaylists"
assert http_app_calls[0][1]["factory"] == web_lib.app_factory
def test_setup_registers_frontend():
ext = Extension()
registry = mock.Mock()
ext.setup(registry)
calls = [c.args for c in registry.add.call_args_list]
frontend_calls = [c for c in calls if c[0] == "frontend"]
assert len(frontend_calls) == 1
assert frontend_calls[0][1] == frontend_lib.SmartPlaylistsFrontend

63
tests/test_frontend.py Normal file
View File

@ -0,0 +1,63 @@
from unittest import mock
from mopidy_smartplaylists.frontend import SmartPlaylistsFrontend
class TestSmartPlaylistsFrontend:
def test_on_start_with_recipes(self):
config = mock.Mock()
config.get.return_value = {
"decades": "1980,1990",
"genres": "",
"artists": "",
"playlist_prefix": "[Smart]",
"refresh_interval": "0",
}
core = mock.Mock()
frontend = SmartPlaylistsFrontend(config, core)
frontend.on_start()
core.library.search.assert_called()
def test_on_start_without_recipes(self):
config = mock.Mock()
config.get.return_value = {
"decades": "",
"genres": "",
"artists": "",
"playlist_prefix": "[Smart]",
"refresh_interval": "0",
}
core = mock.Mock()
frontend = SmartPlaylistsFrontend(config, core)
frontend.on_start()
core.library.search.assert_not_called()
def test_playlists_loaded_with_interval(self):
config = mock.Mock()
config.get.return_value = {
"playlist_prefix": "[Smart]",
"refresh_interval": "24",
}
core = mock.Mock()
frontend = SmartPlaylistsFrontend(config, core)
frontend.playlists_loaded()
core.library.search.assert_called()
def test_playlists_loaded_without_interval(self):
config = mock.Mock()
config.get.return_value = {
"playlist_prefix": "[Smart]",
"refresh_interval": "0",
}
core = mock.Mock()
frontend = SmartPlaylistsFrontend(config, core)
frontend.playlists_loaded()
core.library.search.assert_not_called()

241
tests/test_generators.py Normal file
View File

@ -0,0 +1,241 @@
from unittest import mock
import pytest
from mopidy.models import Artist, Playlist, SearchResult, Track
from mopidy_smartplaylists.generators import (
_extract_tracks,
_flatten_lookup,
_sanitize_name,
build_artist_mix,
build_decade_mix,
build_genre_mix,
build_instant_mix,
parse_decade,
refresh_smart_playlists,
save_smart_playlist,
)
class TestParseDecade:
def test_four_digit_year(self):
assert parse_decade("1980") == "198*"
def test_with_suffix(self):
assert parse_decade("1980s") == "198*"
def test_three_digit(self):
assert parse_decade("199") == "199*"
def test_whitespace(self):
assert parse_decade(" 2000 ") == "200*"
class TestSanitizeName:
def test_lowercase(self):
assert _sanitize_name("Hello World") == "hello_world"
def test_special_chars(self):
assert _sanitize_name("Jazz/Funk Mix!") == "jazz_funk_mix_"
def test_already_sanitized(self):
assert _sanitize_name("hello_world") == "hello_world"
class TestExtractTracks:
def test_extracts_from_search_results(self):
tracks = [Track(uri="dummy:1", name="Track 1"), Track(uri="dummy:2", name="Track 2")]
results = [SearchResult(uri="dummy:search", tracks=tracks)]
assert _extract_tracks(results) == tracks
def test_empty_results(self):
assert _extract_tracks([]) == []
class TestFlattenLookup:
def test_flattens_dict_of_lists(self):
result = {
"dummy:1": [Track(uri="dummy:1", name="Track 1")],
"dummy:2": [Track(uri="dummy:2", name="Track 2")],
}
flat = _flatten_lookup(result)
assert len(flat) == 2
def test_empty_lookup(self):
assert _flatten_lookup({}) == []
class TestBuildDecadeMix:
def test_searches_with_decade_query(self):
core = mock.Mock()
expected_tracks = [Track(uri="dummy:1", name="Track 1")]
core.library.search.return_value.get.return_value = [
SearchResult(uri="dummy:search", tracks=expected_tracks)
]
result = build_decade_mix(core, "1980")
core.library.search.assert_called_once()
query_arg = core.library.search.call_args[0][0]
assert "date" in query_arg
assert query_arg["date"] == ["198*"]
assert result == expected_tracks
class TestBuildGenreMix:
def test_searches_with_genre_query(self):
core = mock.Mock()
expected_tracks = [Track(uri="dummy:1", name="Jazz Track")]
core.library.search.return_value.get.return_value = [
SearchResult(uri="dummy:search", tracks=expected_tracks)
]
result = build_genre_mix(core, "Jazz")
core.library.search.assert_called_once()
query_arg = core.library.search.call_args[0][0]
assert query_arg["genre"] == ["Jazz"]
assert result == expected_tracks
class TestBuildArtistMix:
def test_searches_with_artist_query(self):
core = mock.Mock()
expected_tracks = [Track(uri="dummy:1", name="Artist Track")]
core.library.search.return_value.get.return_value = [
SearchResult(uri="dummy:search", tracks=expected_tracks)
]
result = build_artist_mix(core, "Miles Davis")
core.library.search.assert_called_once()
query_arg = core.library.search.call_args[0][0]
assert query_arg["artist"] == ["Miles Davis"]
assert result == expected_tracks
class TestBuildInstantMix:
def test_returns_similar_tracks_by_genre(self):
core = mock.Mock()
seed = Track(
uri="dummy:seed",
name="Seed Track",
genre="Jazz",
artists=[Artist(name="Miles Davis")],
)
core.library.lookup.return_value.get.return_value = {
"dummy:seed": [seed],
}
similar = [
Track(uri="dummy:similar1", name="Similar 1", genre="Jazz"),
Track(uri="dummy:similar2", name="Similar 2", genre="Jazz"),
]
core.library.search.return_value.get.return_value = [
SearchResult(uri="dummy:search", tracks=similar),
]
result = build_instant_mix(core, "dummy:seed", limit=50)
assert len(result) == 2
assert result[0].uri == "dummy:similar1"
def test_excludes_seed_track(self):
core = mock.Mock()
seed = Track(uri="dummy:seed", name="Seed Track", genre="Rock")
core.library.lookup.return_value.get.return_value = {
"dummy:seed": [seed],
}
same_as_seed = [Track(uri="dummy:seed", name="Seed Track", genre="Rock")]
core.library.search.return_value.get.return_value = [
SearchResult(uri="dummy:search", tracks=same_as_seed),
]
result = build_instant_mix(core, "dummy:seed")
assert len(result) == 0
def test_empty_lookup_returns_empty(self):
core = mock.Mock()
core.library.lookup.return_value.get.return_value = {}
result = build_instant_mix(core, "dummy:nonexistent")
assert result == []
class TestSaveSmartPlaylist:
def test_creates_new_playlist(self):
core = mock.Mock()
core.playlists.lookup.return_value.get.return_value = None
saved = Playlist(
name="[Smart] Jazz Mix",
uri="mopidy:smartplaylists:jazz_mix",
tracks=[Track(uri="dummy:1", name="Track")],
)
core.playlists.save.return_value.get.return_value = saved
tracks = [Track(uri="dummy:1", name="Track")]
result = save_smart_playlist(core, "[Smart]", "Jazz Mix", tracks)
assert result == saved
core.playlists.save.assert_called_once()
def test_deletes_existing_before_save(self):
core = mock.Mock()
existing = Playlist(
name="[Smart] Old Jazz Mix",
uri="mopidy:smartplaylists:jazz_mix",
tracks=[],
)
core.playlists.lookup.return_value.get.return_value = existing
saved = Playlist(
name="[Smart] Jazz Mix",
uri="mopidy:smartplaylists:jazz_mix",
tracks=[Track(uri="dummy:1", name="Track")],
)
core.playlists.save.return_value.get.return_value = saved
tracks = [Track(uri="dummy:1", name="Track")]
result = save_smart_playlist(core, "[Smart]", "Jazz Mix", tracks)
assert result == saved
core.playlists.delete.assert_called_once_with(existing.uri)
class TestRefreshSmartPlaylists:
def test_skips_when_no_recipes(self):
core = mock.Mock()
config = {}
refresh_smart_playlists(core, config)
core.library.search.assert_not_called()
def test_generates_decade_mixes(self):
core = mock.Mock()
config = {"decades": "1980,1990", "playlist_prefix": "[Smart]"}
core.library.search.return_value.get.return_value = [
SearchResult(uri="dummy:search", tracks=[Track(uri="dummy:1", name="T")])
]
core.playlists.lookup.return_value.get.return_value = None
core.playlists.save.return_value.get.return_value = Playlist(
name="test", uri="dummy:playlist", tracks=[]
)
refresh_smart_playlists(core, config)
assert core.library.search.call_count == 2
assert core.playlists.save.call_count == 2
def test_generates_genre_mixes(self):
core = mock.Mock()
config = {"genres": "Jazz,Rock", "playlist_prefix": "[Smart]"}
core.library.search.return_value.get.return_value = [
SearchResult(uri="dummy:search", tracks=[Track(uri="dummy:1", name="T")])
]
core.playlists.lookup.return_value.get.return_value = None
core.playlists.save.return_value.get.return_value = Playlist(
name="test", uri="dummy:playlist", tracks=[]
)
refresh_smart_playlists(core, config)
assert core.library.search.call_count == 2
assert core.playlists.save.call_count == 2

148
tests/test_web.py Normal file
View File

@ -0,0 +1,148 @@
import json
from unittest import mock
import pytest
from mopidy.models import Playlist, Track
from mopidy_smartplaylists.web import (
AlbumMixHandler,
ArtistMixHandler,
DecadeMixHandler,
GenreMixHandler,
InstantMixHandler,
RefreshHandler,
StatusHandler,
app_factory,
)
class TestAppFactory:
def test_returns_list_of_routes(self):
config = mock.Mock()
config.get.return_value = {"playlist_prefix": "[Smart]"}
core = mock.Mock()
routes = app_factory(config, core)
assert isinstance(routes, list)
assert len(routes) > 0
for pattern, handler_cls, kwargs in routes:
assert isinstance(pattern, str)
assert issubclass(handler_cls, object)
assert "core" in kwargs
class TestDecadeMixHandler:
@pytest.fixture
def handler(self):
core = mock.Mock()
core.library.search.return_value.get.return_value = [
mock.Mock(tracks=[Track(uri="dummy:1", name="T")])
]
core.playlists.lookup.return_value.get.return_value = None
core.playlists.save.return_value.get.return_value = Playlist(
name="[Smart] 1980s Mix",
uri="mopidy:smartplaylists:1980s_mix",
tracks=[Track(uri="dummy:1", name="T")],
)
h = DecadeMixHandler(mock.Mock(), mock.Mock(), core=core, prefix="[Smart]")
h.request = mock.Mock()
h.request.body = json.dumps({"decade": "1980"}).encode()
return h
def test_returns_playlist_info(self, handler):
handler.write = mock.Mock()
handler.post()
data = handler.write.call_args[0][0]
assert data["playlist"]["name"] == "[Smart] 1980s Mix"
assert data["tracks"] == 1
def test_missing_decade_returns_400(self):
h = DecadeMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]")
h.request = mock.Mock()
h.request.body = json.dumps({}).encode()
h.set_status = mock.Mock()
h.write = mock.Mock()
h.post()
h.set_status.assert_called_once_with(400)
class TestGenreMixHandler:
def test_missing_genre_returns_400(self):
h = GenreMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]")
h.request = mock.Mock()
h.request.body = json.dumps({}).encode()
h.set_status = mock.Mock()
h.write = mock.Mock()
h.post()
h.set_status.assert_called_once_with(400)
class TestArtistMixHandler:
def test_missing_artist_returns_400(self):
h = ArtistMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]")
h.request = mock.Mock()
h.request.body = json.dumps({}).encode()
h.set_status = mock.Mock()
h.write = mock.Mock()
h.post()
h.set_status.assert_called_once_with(400)
class TestAlbumMixHandler:
def test_missing_uri_returns_400(self):
h = AlbumMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]")
h.request = mock.Mock()
h.request.body = json.dumps({}).encode()
h.set_status = mock.Mock()
h.write = mock.Mock()
h.post()
h.set_status.assert_called_once_with(400)
class TestInstantMixHandler:
def test_missing_uri_returns_400(self):
h = InstantMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]")
h.request = mock.Mock()
h.request.body = json.dumps({}).encode()
h.set_status = mock.Mock()
h.write = mock.Mock()
h.post()
h.set_status.assert_called_once_with(400)
class TestRefreshHandler:
def test_calls_refresh(self):
config = mock.Mock()
config.get.return_value = {"playlist_prefix": "[Smart]"}
core = mock.Mock()
h = RefreshHandler(mock.Mock(), mock.Mock(), core=core, config=config)
h.request = mock.Mock()
h.request.body = b"{}"
h.write = mock.Mock()
h.post()
assert h.write.call_args[0][0] == {"status": "ok"}
class TestStatusHandler:
def test_returns_smart_playlists(self):
core = mock.Mock()
core.playlists.as_list.return_value.get.return_value = [
Playlist(
name="[Smart] Jazz Mix",
uri="mopidy:smartplaylists:jazz_mix",
tracks=[Track(uri="dummy:1", name="T")],
),
Playlist(
name="Normal Playlist",
uri="mopidy:playlist:normal",
tracks=[],
),
]
h = StatusHandler(mock.Mock(), mock.Mock(), core=core)
h.write = mock.Mock()
h.get()
data = h.write.call_args[0][0]
assert data["count"] == 1
assert data["smart_playlists"][0]["name"] == "[Smart] Jazz Mix"