[deps] Fix so it works with older mopidy
Some checks failed
CI / Build (push) Failing after 4m58s
CI / pyright (push) Failing after 6s
CI / pytest (3.13) (push) Failing after 5s
CI / ruff check (push) Failing after 5s
CI / ruff format (push) Failing after 7s
CI / pytest (3.14) (push) Failing after 7s

This commit is contained in:
2026-06-24 19:43:20 -04:00
parent a054575dea
commit 51e0c8a9ad
9 changed files with 355 additions and 162 deletions

View File

@ -2,7 +2,7 @@
name = "mopidy-smartplaylists" name = "mopidy-smartplaylists"
description = "Mopidy extension for generating smart playlists based on track properties" description = "Mopidy extension for generating smart playlists based on track properties"
readme = "README.md" readme = "README.md"
requires-python = ">= 3.13" requires-python = ">= 3.10"
license = { text = "Apache-2.0" } license = { text = "Apache-2.0" }
authors = [{ name = "Chris Powell", email = "chris@code.unbl.ink" }] authors = [{ name = "Chris Powell", email = "chris@code.unbl.ink" }]
classifiers = [ classifiers = [
@ -14,9 +14,9 @@ classifiers = [
] ]
dynamic = ["version"] dynamic = ["version"]
dependencies = [ dependencies = [
"mopidy >= 4.0.0", "mopidy >= 3.0.0",
"pykka >= 4.1", "pykka >= 2.0.1",
"tornado >= 6.4", "tornado >= 4.4",
] ]
[project.urls] [project.urls]
@ -54,7 +54,7 @@ show_missing = true
[tool.pyright] [tool.pyright]
pythonVersion = "3.13" pythonVersion = "3.10"
typeCheckingMode = "standard" typeCheckingMode = "standard"
reportMissingTypeStubs = false reportMissingTypeStubs = false
reportPrivateImportUsage = false reportPrivateImportUsage = false
@ -67,7 +67,7 @@ filterwarnings = [
[tool.ruff] [tool.ruff]
target-version = "py313" target-version = "py310"
[tool.ruff.lint] [tool.ruff.lint]
select = ["ALL"] select = ["ALL"]
@ -105,6 +105,9 @@ ignore = [
[tool.tox] [tool.tox]
env_list = [ env_list = [
"3.10",
"3.11",
"3.12",
"3.13", "3.13",
"3.14", "3.14",
"pyright", "pyright",

View File

@ -2,7 +2,6 @@ import pathlib
from importlib.metadata import version from importlib.metadata import version
from mopidy import config, ext from mopidy import config, ext
from mopidy.config import ConfigSchema
__version__ = version("mopidy-smartplaylists") __version__ = version("mopidy-smartplaylists")
@ -12,10 +11,10 @@ class Extension(ext.Extension):
ext_name = "smartplaylists" ext_name = "smartplaylists"
version = __version__ version = __version__
def get_default_config(self): def get_default_config(self) -> str:
return config.read(pathlib.Path(__file__).parent / "ext.conf") return config.read(pathlib.Path(__file__).parent / "ext.conf")
def get_config_schema(self): def get_config_schema(self) -> config.ConfigSchema:
schema = super().get_config_schema() schema = super().get_config_schema()
schema["decades"] = config.String(optional=True) schema["decades"] = config.String(optional=True)
schema["genres"] = config.String(optional=True) schema["genres"] = config.String(optional=True)
@ -24,16 +23,19 @@ class Extension(ext.Extension):
schema["refresh_interval"] = config.Integer(optional=True, minimum=0) schema["refresh_interval"] = config.Integer(optional=True, minimum=0)
return schema return schema
def validate_environment(self): def validate_environment(self) -> None:
pass pass
def setup(self, registry): def setup(self, registry: ext.Registry) -> None:
from mopidy_smartplaylists.web import app_factory from mopidy_smartplaylists.web import app_factory
registry.add("http:app", { registry.add(
"http:app",
{
"name": self.ext_name, "name": self.ext_name,
"factory": app_factory, "factory": app_factory,
}) },
)
from mopidy_smartplaylists.frontend import SmartPlaylistsFrontend from mopidy_smartplaylists.frontend import SmartPlaylistsFrontend

View File

@ -0,0 +1,23 @@
"""
Compatibility layer for Mopidy 3 vs 4 differences.
Provides import aliases for APIs that differ between Mopidy 3 and 4:
- ``CoreProxy`` / ``Config`` are available in both but under different modules.
- ``mopidy.types`` (``Uri``, ``SearchField``, ``Query``) only exists in Mopidy 4.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from mopidy.config import Config
from mopidy.core import CoreProxy
from mopidy.types import Query, SearchField, Uri
else:
CoreProxy = Any
Config = dict[str, dict[str, Any]]
Uri = str
SearchField = str
Query = dict

View File

@ -1,11 +1,16 @@
from __future__ import annotations
import logging import logging
from typing import TYPE_CHECKING
import pykka import pykka
from mopidy.config import Config from mopidy.core import CoreListener
from mopidy.core import CoreListener, CoreProxy
from mopidy_smartplaylists.generators import refresh_smart_playlists from mopidy_smartplaylists.generators import refresh_smart_playlists
if TYPE_CHECKING:
from mopidy_smartplaylists.compat import Config, CoreProxy
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)

View File

@ -1,23 +1,32 @@
from __future__ import annotations
import logging import logging
import re import re
from typing import TYPE_CHECKING, cast
from mopidy.core import CoreProxy from mopidy.models import Playlist, Track
from mopidy.models import Playlist, Track, TrackField
if TYPE_CHECKING:
from mopidy_smartplaylists.compat import CoreProxy, Query, SearchField, Uri
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def parse_decade(decade_str: str) -> str: def parse_decade(decade_str: str) -> str:
decade_str = decade_str.strip() decade_str = decade_str.strip().rstrip("s")
if len(decade_str) == 4: if len(decade_str) >= 3:
return decade_str[:3] + "*"
if decade_str.endswith("s"):
return decade_str[:3] + "*" return decade_str[:3] + "*"
return decade_str return decade_str
def _search(core: CoreProxy, field: str, value: str) -> list[Track]:
query = cast("Query[SearchField]", {field: [value]})
result = core.library.search(query).get()
return _extract_tracks(result)
def build_decade_mix(core: CoreProxy, decade: str) -> list[Track]: def build_decade_mix(core: CoreProxy, decade: str) -> list[Track]:
query = {TrackField.DATE: [parse_decade(decade)]} query = cast("Query[SearchField]", {"date": [parse_decade(decade)]})
result = core.library.search(query).get() result = core.library.search(query).get()
tracks = _extract_tracks(result) tracks = _extract_tracks(result)
logger.info("Found %d tracks for decade %s", len(tracks), decade) logger.info("Found %d tracks for decade %s", len(tracks), decade)
@ -25,30 +34,22 @@ def build_decade_mix(core: CoreProxy, decade: str) -> list[Track]:
def build_genre_mix(core: CoreProxy, genre: str) -> list[Track]: def build_genre_mix(core: CoreProxy, genre: str) -> list[Track]:
query = {TrackField.GENRE: [genre]} return _search(core, "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]: def build_artist_mix(core: CoreProxy, artist: str) -> list[Track]:
query = {TrackField.ARTIST: [artist]} return _search(core, "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]: def build_album_mix(core: CoreProxy, album_uri: str) -> list[Track]:
tracks = core.library.lookup([album_uri]).get() tracks = core.library.lookup(cast("list[Uri]", [album_uri])).get()
flat = _flatten_lookup(tracks) flat = _flatten_lookup(tracks)
logger.info("Found %d tracks for album %s", len(flat), album_uri) logger.info("Found %d tracks for album %s", len(flat), album_uri)
return flat return flat
def build_instant_mix(core: CoreProxy, track_uri: str, limit: int = 50) -> list[Track]: def build_instant_mix(core: CoreProxy, track_uri: str, limit: int = 50) -> list[Track]:
lookup_result = core.library.lookup([track_uri]).get() lookup_result = core.library.lookup(cast("list[Uri]", [track_uri])).get()
seed_tracks = _flatten_lookup(lookup_result) seed_tracks = _flatten_lookup(lookup_result)
if not seed_tracks: if not seed_tracks:
logger.warning("No track found for URI: %s", track_uri) logger.warning("No track found for URI: %s", track_uri)
@ -61,16 +62,14 @@ def build_instant_mix(core: CoreProxy, track_uri: str, limit: int = 50) -> list[
if seed.genre: if seed.genre:
genres = [g.strip() for g in seed.genre.split("/") if g.strip()] genres = [g.strip() for g in seed.genre.split("/") if g.strip()]
if seed.artists: if seed.artists:
for artist_ref in seed.artists: artists.extend(a_ref.name for a_ref in seed.artists if a_ref.name)
if artist_ref.name:
artists.append(artist_ref.name)
similar: dict[str, Track] = {} similar: dict[str, Track] = {}
for genre in genres: for genre in genres:
query = {TrackField.GENRE: [genre]} g_query = cast("Query[SearchField]", {"genre": [genre]})
result = core.library.search(query).get() genre_result = core.library.search(g_query).get()
for batch in result: for batch in genre_result:
for t in batch.tracks: for t in batch.tracks:
if t.uri and t.uri != track_uri: if t.uri and t.uri != track_uri:
similar[t.uri] = t similar[t.uri] = t
@ -78,9 +77,9 @@ def build_instant_mix(core: CoreProxy, track_uri: str, limit: int = 50) -> list[
for artist in artists: for artist in artists:
if len(similar) >= limit: if len(similar) >= limit:
break break
query = {TrackField.ARTIST: [artist]} a_query = cast("Query[SearchField]", {"artist": [artist]})
result = core.library.search(query).get() artist_result = core.library.search(a_query).get()
for batch in result: for batch in artist_result:
for t in batch.tracks: for t in batch.tracks:
if t.uri and t.uri != track_uri: if t.uri and t.uri != track_uri:
similar[t.uri] = t similar[t.uri] = t
@ -97,16 +96,16 @@ def save_smart_playlist(
tracks: list[Track], tracks: list[Track],
) -> Playlist | None: ) -> Playlist | None:
playlist_name = f"{prefix} {name}" playlist_name = f"{prefix} {name}"
uri = f"mopidy:smartplaylists:{_sanitize_name(name)}" uri: str = f"mopidy:smartplaylists:{_sanitize_name(name)}"
existing = core.playlists.lookup(uri).get() existing = core.playlists.lookup(cast("Uri", uri)).get()
if existing: if existing:
core.playlists.delete(existing.uri).get() core.playlists.delete(existing.uri).get()
playlist = Playlist( playlist = Playlist(
name=playlist_name, name=playlist_name,
uri=uri, uri=cast("Uri", uri),
tracks=tracks, tracks=tuple(tracks),
) )
saved = core.playlists.save(playlist).get() saved = core.playlists.save(playlist).get()
logger.info("Saved smart playlist: %s (%d tracks)", playlist_name, len(tracks)) logger.info("Saved smart playlist: %s (%d tracks)", playlist_name, len(tracks))
@ -141,14 +140,14 @@ def refresh_smart_playlists(core: CoreProxy, config_dict: dict) -> None:
save_smart_playlist(core, prefix, f"{artist} Mix", tracks) save_smart_playlist(core, prefix, f"{artist} Mix", tracks)
def _extract_tracks(search_result) -> list[Track]: def _extract_tracks(search_result: list) -> list[Track]:
tracks: list[Track] = [] tracks: list[Track] = []
for batch in search_result: for batch in search_result:
tracks.extend(batch.tracks) tracks.extend(batch.tracks)
return tracks return tracks
def _flatten_lookup(lookup_result: dict[str, list[Track]]) -> list[Track]: def _flatten_lookup(lookup_result: dict[Uri, list[Track]]) -> list[Track]:
tracks: list[Track] = [] tracks: list[Track] = []
for uri_tracks in lookup_result.values(): for uri_tracks in lookup_result.values():
tracks.extend(uri_tracks) tracks.extend(uri_tracks)

View File

@ -1,9 +1,16 @@
from __future__ import annotations
import json import json
import logging import logging
from typing import TYPE_CHECKING, cast
import tornado.web import tornado.web
from mopidy.config import Config
from mopidy.core import CoreProxy if TYPE_CHECKING:
from mopidy_smartplaylists.compat import Config, CoreProxy
if TYPE_CHECKING:
from mopidy.models import Playlist
from mopidy_smartplaylists.generators import ( from mopidy_smartplaylists.generators import (
build_album_mix, build_album_mix,
@ -35,13 +42,15 @@ class DecadeMixHandler(tornado.web.RequestHandler):
self.write({"playlist": None, "tracks": 0}) self.write({"playlist": None, "tracks": 0})
return return
playlist = save_smart_playlist(self.core, self.prefix, f"{decade}s Mix", tracks) playlist = save_smart_playlist(self.core, self.prefix, f"{decade}s Mix", tracks)
self.write({ self.write(
{
"playlist": { "playlist": {
"name": playlist.name if playlist else None, "name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None, "uri": playlist.uri if playlist else None,
"tracks": len(tracks), "tracks": len(tracks),
}, },
}) }
)
class GenreMixHandler(tornado.web.RequestHandler): class GenreMixHandler(tornado.web.RequestHandler):
@ -61,13 +70,15 @@ class GenreMixHandler(tornado.web.RequestHandler):
self.write({"playlist": None, "tracks": 0}) self.write({"playlist": None, "tracks": 0})
return return
playlist = save_smart_playlist(self.core, self.prefix, f"{genre} Mix", tracks) playlist = save_smart_playlist(self.core, self.prefix, f"{genre} Mix", tracks)
self.write({ self.write(
{
"playlist": { "playlist": {
"name": playlist.name if playlist else None, "name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None, "uri": playlist.uri if playlist else None,
"tracks": len(tracks), "tracks": len(tracks),
}, },
}) }
)
class ArtistMixHandler(tornado.web.RequestHandler): class ArtistMixHandler(tornado.web.RequestHandler):
@ -87,13 +98,15 @@ class ArtistMixHandler(tornado.web.RequestHandler):
self.write({"playlist": None, "tracks": 0}) self.write({"playlist": None, "tracks": 0})
return return
playlist = save_smart_playlist(self.core, self.prefix, f"{artist} Mix", tracks) playlist = save_smart_playlist(self.core, self.prefix, f"{artist} Mix", tracks)
self.write({ self.write(
{
"playlist": { "playlist": {
"name": playlist.name if playlist else None, "name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None, "uri": playlist.uri if playlist else None,
"tracks": len(tracks), "tracks": len(tracks),
}, },
}) }
)
class AlbumMixHandler(tornado.web.RequestHandler): class AlbumMixHandler(tornado.web.RequestHandler):
@ -112,15 +125,21 @@ class AlbumMixHandler(tornado.web.RequestHandler):
if not tracks: if not tracks:
self.write({"playlist": None, "tracks": 0}) self.write({"playlist": None, "tracks": 0})
return return
name = tracks[0].album.name if tracks[0].album and tracks[0].album.name else "Album" 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) playlist = save_smart_playlist(self.core, self.prefix, f"{name} Mix", tracks)
self.write({ self.write(
{
"playlist": { "playlist": {
"name": playlist.name if playlist else None, "name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None, "uri": playlist.uri if playlist else None,
"tracks": len(tracks), "tracks": len(tracks),
}, },
}) }
)
class InstantMixHandler(tornado.web.RequestHandler): class InstantMixHandler(tornado.web.RequestHandler):
@ -149,14 +168,18 @@ class InstantMixHandler(tornado.web.RequestHandler):
seed_name = t.name seed_name = t.name
break break
playlist = save_smart_playlist(self.core, self.prefix, f"Instant Mix: {seed_name}", tracks) playlist = save_smart_playlist(
self.write({ self.core, self.prefix, f"Instant Mix: {seed_name}", tracks
)
self.write(
{
"playlist": { "playlist": {
"name": playlist.name if playlist else None, "name": playlist.name if playlist else None,
"uri": playlist.uri if playlist else None, "uri": playlist.uri if playlist else None,
"tracks": len(tracks), "tracks": len(tracks),
}, },
}) }
)
class RefreshHandler(tornado.web.RequestHandler): class RefreshHandler(tornado.web.RequestHandler):
@ -176,8 +199,10 @@ class StatusHandler(tornado.web.RequestHandler):
def get(self) -> None: def get(self) -> None:
result = self.core.playlists.as_list().get() result = self.core.playlists.as_list().get()
smart = [p for p in result if p.uri and "smartplaylists" in p.uri] result_typed = cast("list[Playlist]", result)
self.write({ smart = [p for p in result_typed if p.uri and "smartplaylists" in p.uri]
self.write(
{
"smart_playlists": [ "smart_playlists": [
{ {
"name": p.name, "name": p.name,
@ -187,7 +212,8 @@ class StatusHandler(tornado.web.RequestHandler):
for p in smart for p in smart
], ],
"count": len(smart), "count": len(smart),
}) }
)
def app_factory(config: Config, core: CoreProxy) -> list[tuple]: def app_factory(config: Config, core: CoreProxy) -> list[tuple]:

View File

@ -1,5 +1,7 @@
from unittest import mock from unittest import mock
from mopidy.models import Playlist
from mopidy_smartplaylists.frontend import SmartPlaylistsFrontend from mopidy_smartplaylists.frontend import SmartPlaylistsFrontend
@ -14,11 +16,16 @@ class TestSmartPlaylistsFrontend:
"refresh_interval": "0", "refresh_interval": "0",
} }
core = mock.Mock() core = mock.Mock()
core.library.search.return_value.get.return_value = []
core.playlists.lookup.return_value.get.return_value = None
core.playlists.save.return_value.get.return_value = Playlist(
name="test", uri="dummy:pl", tracks=[]
)
frontend = SmartPlaylistsFrontend(config, core) frontend = SmartPlaylistsFrontend(config, core)
frontend.on_start() frontend.on_start()
core.library.search.assert_called() assert core.library.search.call_count == 2
def test_on_start_without_recipes(self): def test_on_start_without_recipes(self):
config = mock.Mock() config = mock.Mock()
@ -39,10 +46,16 @@ class TestSmartPlaylistsFrontend:
def test_playlists_loaded_with_interval(self): def test_playlists_loaded_with_interval(self):
config = mock.Mock() config = mock.Mock()
config.get.return_value = { config.get.return_value = {
"decades": "1980",
"playlist_prefix": "[Smart]", "playlist_prefix": "[Smart]",
"refresh_interval": "24", "refresh_interval": "24",
} }
core = mock.Mock() core = mock.Mock()
core.library.search.return_value.get.return_value = []
core.playlists.lookup.return_value.get.return_value = None
core.playlists.save.return_value.get.return_value = Playlist(
name="test", uri="dummy:pl", tracks=[]
)
frontend = SmartPlaylistsFrontend(config, core) frontend = SmartPlaylistsFrontend(config, core)
frontend.playlists_loaded() frontend.playlists_loaded()

View File

@ -1,6 +1,5 @@
from unittest import mock from unittest import mock
import pytest
from mopidy.models import Artist, Playlist, SearchResult, Track from mopidy.models import Artist, Playlist, SearchResult, Track
from mopidy_smartplaylists.generators import ( from mopidy_smartplaylists.generators import (
@ -44,7 +43,10 @@ class TestSanitizeName:
class TestExtractTracks: class TestExtractTracks:
def test_extracts_from_search_results(self): def test_extracts_from_search_results(self):
tracks = [Track(uri="dummy:1", name="Track 1"), Track(uri="dummy:2", name="Track 2")] tracks = [
Track(uri="dummy:1", name="Track 1"),
Track(uri="dummy:2", name="Track 2"),
]
results = [SearchResult(uri="dummy:search", tracks=tracks)] results = [SearchResult(uri="dummy:search", tracks=tracks)]
assert _extract_tracks(results) == tracks assert _extract_tracks(results) == tracks

View File

@ -1,7 +1,8 @@
import json import json
from unittest import mock from unittest import mock
import pytest import tornado.testing
import tornado.web
from mopidy.models import Playlist, Track from mopidy.models import Playlist, Track
from mopidy_smartplaylists.web import ( from mopidy_smartplaylists.web import (
@ -30,9 +31,51 @@ class TestAppFactory:
assert "core" in kwargs assert "core" in kwargs
class _HandlerTestBase:
def make_app(self, handler_cls, **kwargs):
return tornado.web.Application(
[
(r"/test", handler_cls, kwargs),
]
)
def call_handler(self, handler_cls, method="POST", body=None, **kwargs):
app = self.make_app(handler_cls, **kwargs)
request = tornado.httputil.HTTPServerRequest(
method=method,
uri="/test",
body=json.dumps(body).encode() if isinstance(body, dict) else body,
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
return handler_cls(app, request, **kwargs)
class TestDecadeMixHandler: class TestDecadeMixHandler:
@pytest.fixture def test_missing_decade_returns_400(self):
def handler(self): app = tornado.web.Application(
[
(
r"/decade",
DecadeMixHandler,
{"core": mock.Mock(), "prefix": "[Smart]"},
),
]
)
request = tornado.httputil.HTTPServerRequest(
method="POST",
uri="/decade",
body=json.dumps({}).encode(),
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = DecadeMixHandler(app, request, core=mock.Mock(), prefix="[Smart]")
handler.set_status = mock.Mock()
handler.write = mock.Mock()
handler.post()
handler.set_status.assert_called_once_with(400)
def test_success(self):
core = mock.Mock() core = mock.Mock()
core.library.search.return_value.get.return_value = [ core.library.search.return_value.get.return_value = [
mock.Mock(tracks=[Track(uri="dummy:1", name="T")]) mock.Mock(tracks=[Track(uri="dummy:1", name="T")])
@ -44,70 +87,125 @@ class TestDecadeMixHandler:
tracks=[Track(uri="dummy:1", name="T")], tracks=[Track(uri="dummy:1", name="T")],
) )
h = DecadeMixHandler(mock.Mock(), mock.Mock(), core=core, prefix="[Smart]") app = tornado.web.Application(
h.request = mock.Mock() [
h.request.body = json.dumps({"decade": "1980"}).encode() (r"/decade", DecadeMixHandler, {"core": core, "prefix": "[Smart]"}),
return h ]
)
def test_returns_playlist_info(self, handler): request = tornado.httputil.HTTPServerRequest(
method="POST",
uri="/decade",
body=json.dumps({"decade": "1980"}).encode(),
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = DecadeMixHandler(app, request, core=core, prefix="[Smart]")
handler._transforms = []
handler.write = mock.Mock() handler.write = mock.Mock()
handler.post() handler.post()
data = handler.write.call_args[0][0] data = handler.write.call_args[0][0]
assert data["playlist"]["name"] == "[Smart] 1980s Mix" assert data["playlist"]["name"] == "[Smart] 1980s Mix"
assert data["tracks"] == 1 assert data["playlist"]["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: class TestGenreMixHandler:
def test_missing_genre_returns_400(self): def test_missing_genre_returns_400(self):
h = GenreMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]") app = tornado.web.Application(
h.request = mock.Mock() [
h.request.body = json.dumps({}).encode() (
h.set_status = mock.Mock() r"/genre",
h.write = mock.Mock() GenreMixHandler,
h.post() {"core": mock.Mock(), "prefix": "[Smart]"},
h.set_status.assert_called_once_with(400) ),
]
)
request = tornado.httputil.HTTPServerRequest(
method="POST",
uri="/genre",
body=json.dumps({}).encode(),
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = GenreMixHandler(app, request, core=mock.Mock(), prefix="[Smart]")
handler.set_status = mock.Mock()
handler.write = mock.Mock()
handler.post()
handler.set_status.assert_called_once_with(400)
class TestArtistMixHandler: class TestArtistMixHandler:
def test_missing_artist_returns_400(self): def test_missing_artist_returns_400(self):
h = ArtistMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]") app = tornado.web.Application(
h.request = mock.Mock() [
h.request.body = json.dumps({}).encode() (
h.set_status = mock.Mock() r"/artist",
h.write = mock.Mock() ArtistMixHandler,
h.post() {"core": mock.Mock(), "prefix": "[Smart]"},
h.set_status.assert_called_once_with(400) ),
]
)
request = tornado.httputil.HTTPServerRequest(
method="POST",
uri="/artist",
body=json.dumps({}).encode(),
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = ArtistMixHandler(app, request, core=mock.Mock(), prefix="[Smart]")
handler.set_status = mock.Mock()
handler.write = mock.Mock()
handler.post()
handler.set_status.assert_called_once_with(400)
class TestAlbumMixHandler: class TestAlbumMixHandler:
def test_missing_uri_returns_400(self): def test_missing_uri_returns_400(self):
h = AlbumMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]") app = tornado.web.Application(
h.request = mock.Mock() [
h.request.body = json.dumps({}).encode() (
h.set_status = mock.Mock() r"/album",
h.write = mock.Mock() AlbumMixHandler,
h.post() {"core": mock.Mock(), "prefix": "[Smart]"},
h.set_status.assert_called_once_with(400) ),
]
)
request = tornado.httputil.HTTPServerRequest(
method="POST",
uri="/album",
body=json.dumps({}).encode(),
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = AlbumMixHandler(app, request, core=mock.Mock(), prefix="[Smart]")
handler.set_status = mock.Mock()
handler.write = mock.Mock()
handler.post()
handler.set_status.assert_called_once_with(400)
class TestInstantMixHandler: class TestInstantMixHandler:
def test_missing_uri_returns_400(self): def test_missing_uri_returns_400(self):
h = InstantMixHandler(mock.Mock(), mock.Mock(), core=mock.Mock(), prefix="[Smart]") app = tornado.web.Application(
h.request = mock.Mock() [
h.request.body = json.dumps({}).encode() (
h.set_status = mock.Mock() r"/instant-mix",
h.write = mock.Mock() InstantMixHandler,
h.post() {"core": mock.Mock(), "prefix": "[Smart]"},
h.set_status.assert_called_once_with(400) ),
]
)
request = tornado.httputil.HTTPServerRequest(
method="POST",
uri="/instant-mix",
body=json.dumps({}).encode(),
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = InstantMixHandler(app, request, core=mock.Mock(), prefix="[Smart]")
handler.set_status = mock.Mock()
handler.write = mock.Mock()
handler.post()
handler.set_status.assert_called_once_with(400)
class TestRefreshHandler: class TestRefreshHandler:
@ -116,12 +214,22 @@ class TestRefreshHandler:
config.get.return_value = {"playlist_prefix": "[Smart]"} config.get.return_value = {"playlist_prefix": "[Smart]"}
core = mock.Mock() core = mock.Mock()
h = RefreshHandler(mock.Mock(), mock.Mock(), core=core, config=config) app = tornado.web.Application(
h.request = mock.Mock() [
h.request.body = b"{}" (r"/refresh", RefreshHandler, {"core": core, "config": config}),
h.write = mock.Mock() ]
h.post() )
assert h.write.call_args[0][0] == {"status": "ok"} request = tornado.httputil.HTTPServerRequest(
method="POST",
uri="/refresh",
body=b"{}",
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = RefreshHandler(app, request, core=core, config=config)
handler.write = mock.Mock()
handler.post()
assert handler.write.call_args[0][0] == {"status": "ok"}
class TestStatusHandler: class TestStatusHandler:
@ -140,9 +248,21 @@ class TestStatusHandler:
), ),
] ]
h = StatusHandler(mock.Mock(), mock.Mock(), core=core) app = tornado.web.Application(
h.write = mock.Mock() [
h.get() (r"/status", StatusHandler, {"core": core}),
data = h.write.call_args[0][0] ]
)
request = tornado.httputil.HTTPServerRequest(
method="GET",
uri="/status",
body=None,
connection=mock.Mock(),
headers=tornado.httputil.HTTPHeaders(),
)
handler = StatusHandler(app, request, core=core)
handler.write = mock.Mock()
handler.get()
data = handler.write.call_args[0][0]
assert data["count"] == 1 assert data["count"] == 1
assert data["smart_playlists"][0]["name"] == "[Smart] Jazz Mix" assert data["smart_playlists"][0]["name"] == "[Smart] Jazz Mix"