Backup_Repos/beets
1
0
Fork 0
mirror of https://github.com/beetbox/beets.git synced 2025-12-06 08:39:17 +01:00
Code Issues Projects Releases Wiki Activity

Test aura (#5239)

Browse source 
In #4746 I was making a small adjustment in beetsplug/aura.py and found
that the module wasn't tested. So this PR adds some high-level tests to
act a safeguard for any future adjustments.
This commit is contained in:
Šarūnas Nejus 2024-06-10 13:07:48 +01:00 committed by GitHub
commit 0966e3c653
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 255 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

196
beetsplug/aura.py
View file

@ -17,8 +17,10 @@
import os.path import os.path
import re import re
from dataclasses import dataclass
from mimetypes import guess_type from mimetypes import guess_type
from os.path import getsize, isfile from os.path import getsize, isfile
from typing import ClassVar, Mapping, Type
from flask import ( from flask import (
Blueprint, Blueprint,
@ -28,6 +30,7 @@ from flask import (
request, request,
send_file, send_file,
) )
from typing_extensions import Self
from beets import config from beets import config
from beets.dbcore.query import ( from beets.dbcore.query import (
@ -38,8 +41,9 @@ from beets.dbcore.query import (
NotQuery, NotQuery,
RegexpQuery, RegexpQuery,
SlowFieldSort, SlowFieldSort,
SQLiteType,
) )
from beets.library import Album, Item from beets.library import Album, Item, LibModel, Library
from beets.plugins import BeetsPlugin from beets.plugins import BeetsPlugin
from beets.ui import Subcommand, _open_library from beets.ui import Subcommand, _open_library
from beets.util import py3_path from beets.util import py3_path
@ -117,9 +121,20 @@ ARTIST_ATTR_MAP = {
} }
@dataclass
class AURADocument: class AURADocument:
"""Base class for building AURA documents.""" """Base class for building AURA documents."""
model_cls: ClassVar[Type[LibModel]]
lib: Library
args: Mapping[str, str]
@classmethod
def from_app(cls) -> Self:
"""Initialise the document using the global app and request."""
return cls(current_app.config["lib"], request.args)
@staticmethod @staticmethod
def error(status, title, detail): def error(status, title, detail):
"""Make a response for an error following the JSON:API spec. """Make a response for an error following the JSON:API spec.
@ -135,13 +150,29 @@ class AURADocument:
} }
return make_response(document, status) return make_response(document, status)
@classmethod
def get_attribute_converter(cls, beets_attr: str) -> Type[SQLiteType]:
"""Work out what data type an attribute should be for beets.
Args:
beets_attr: The name of the beets attribute, e.g. "title".
"""
try:
# Look for field in list of Album fields
# and get python type of database type.
# See beets.library.Album and beets.dbcore.types
return cls.model_cls._fields[beets_attr].model_type
except KeyError:
# Fall back to string (NOTE: probably not good)
return str
def translate_filters(self): def translate_filters(self):
"""Translate filters from request arguments to a beets Query.""" """Translate filters from request arguments to a beets Query."""
# The format of each filter key in the request parameter is: # The format of each filter key in the request parameter is:
# filter[<attribute>]. This regex extracts <attribute>. # filter[<attribute>]. This regex extracts <attribute>.
pattern = re.compile(r"filter\[(?P<attribute>[a-zA-Z0-9_-]+)\]") pattern = re.compile(r"filter\[(?P<attribute>[a-zA-Z0-9_-]+)\]")
queries = [] queries = []
for key, value in request.args.items(): for key, value in self.args.items():
match = pattern.match(key) match = pattern.match(key)
if match: if match:
# Extract attribute name from key # Extract attribute name from key
@ -190,10 +221,10 @@ class AURADocument:
albums) or a list of strings (artists). albums) or a list of strings (artists).
""" """
# Pages start from zero # Pages start from zero
page = request.args.get("page", 0, int) page = self.args.get("page", 0, int)
# Use page limit defined in config by default. # Use page limit defined in config by default.
default_limit = config["aura"]["page_limit"].get(int) default_limit = config["aura"]["page_limit"].get(int)
limit = request.args.get("limit", default_limit, int) limit = self.args.get("limit", default_limit, int)
# start = offset of first item to return # start = offset of first item to return
start = page * limit start = page * limit
# end = offset of last item + 1 # end = offset of last item + 1
@ -203,10 +234,10 @@ class AURADocument:
next_url = None next_url = None
else: else:
# Not the last page so work out links.next url # Not the last page so work out links.next url
if not request.args: if not self.args:
# No existing arguments, so current page is 0 # No existing arguments, so current page is 0
next_url = request.url + "?page=1" next_url = request.url + "?page=1"
elif not request.args.get("page", None): elif not self.args.get("page", None):
# No existing page argument, so add one to the end # No existing page argument, so add one to the end
next_url = request.url + "&page=1" next_url = request.url + "&page=1"
else: else:
@ -215,7 +246,10 @@ class AURADocument:
f"page={page}", "page={}".format(page + 1) f"page={page}", "page={}".format(page + 1)
) )
# Get only the items in the page range # Get only the items in the page range
data = [self.resource_object(collection[i]) for i in range(start, end)] data = [
self.get_resource_object(self.lib, collection[i])
for i in range(start, end)
]
return data, next_url return data, next_url
def get_included(self, data, include_str): def get_included(self, data, include_str):
@ -249,18 +283,26 @@ class AURADocument:
res_type = identifier["type"] res_type = identifier["type"]
if res_type == "track": if res_type == "track":
track_id = int(identifier["id"]) track_id = int(identifier["id"])
track = current_app.config["lib"].get_item(track_id) track = self.lib.get_item(track_id)
included.append(TrackDocument.resource_object(track)) included.append(
TrackDocument.get_resource_object(self.lib, track)
)
elif res_type == "album": elif res_type == "album":
album_id = int(identifier["id"]) album_id = int(identifier["id"])
album = current_app.config["lib"].get_album(album_id) album = self.lib.get_album(album_id)
included.append(AlbumDocument.resource_object(album)) included.append(
AlbumDocument.get_resource_object(self.lib, album)
)
elif res_type == "artist": elif res_type == "artist":
artist_id = identifier["id"] artist_id = identifier["id"]
included.append(ArtistDocument.resource_object(artist_id)) included.append(
ArtistDocument.get_resource_object(self.lib, artist_id)
)
elif res_type == "image": elif res_type == "image":
image_id = identifier["id"] image_id = identifier["id"]
included.append(ImageDocument.resource_object(image_id)) included.append(
ImageDocument.get_resource_object(self.lib, image_id)
)
else: else:
raise ValueError(f"Invalid resource type: {res_type}") raise ValueError(f"Invalid resource type: {res_type}")
return included return included
@ -268,7 +310,7 @@ class AURADocument:
def all_resources(self): def all_resources(self):
"""Build document for /tracks, /albums or /artists.""" """Build document for /tracks, /albums or /artists."""
query = self.translate_filters() query = self.translate_filters()
sort_arg = request.args.get("sort", None) sort_arg = self.args.get("sort", None)
if sort_arg: if sort_arg:
sort = self.translate_sorts(sort_arg) sort = self.translate_sorts(sort_arg)
# For each sort field add a query which ensures all results # For each sort field add a query which ensures all results
@ -291,7 +333,7 @@ class AURADocument:
if next_url: if next_url:
document["links"] = {"next": next_url} document["links"] = {"next": next_url}
# Include related resources for each element in "data" # Include related resources for each element in "data"
include_str = request.args.get("include", None) include_str = self.args.get("include", None)
if include_str: if include_str:
document["included"] = self.get_included(data, include_str) document["included"] = self.get_included(data, include_str)
return document return document
@ -304,7 +346,7 @@ class AURADocument:
resource object. resource object.
""" """
document = {"data": resource_object} document = {"data": resource_object}
include_str = request.args.get("include", None) include_str = self.args.get("include", None)
if include_str: if include_str:
# [document["data"]] is because arg needs to be list # [document["data"]] is because arg needs to be list
document["included"] = self.get_included( document["included"] = self.get_included(
@ -316,6 +358,8 @@ class AURADocument:
class TrackDocument(AURADocument): class TrackDocument(AURADocument):
"""Class for building documents for /tracks endpoints.""" """Class for building documents for /tracks endpoints."""
model_cls = Item
attribute_map = TRACK_ATTR_MAP attribute_map = TRACK_ATTR_MAP
def get_collection(self, query=None, sort=None): def get_collection(self, query=None, sort=None):
@ -325,9 +369,10 @@ class TrackDocument(AURADocument):
query: A beets Query object or a beets query string. query: A beets Query object or a beets query string.
sort: A beets Sort object. sort: A beets Sort object.
""" """
return current_app.config["lib"].items(query, sort) return self.lib.items(query, sort)
def get_attribute_converter(self, beets_attr): @classmethod
def get_attribute_converter(cls, beets_attr: str) -> Type[SQLiteType]:
"""Work out what data type an attribute should be for beets. """Work out what data type an attribute should be for beets.
Args: Args:
@ -335,20 +380,12 @@ class TrackDocument(AURADocument):
""" """
# filesize is a special field (read from disk not db?) # filesize is a special field (read from disk not db?)
if beets_attr == "filesize": if beets_attr == "filesize":
converter = int return int
else:
try: return super().get_attribute_converter(beets_attr)
# Look for field in list of Item fields
# and get python type of database type.
# See beets.library.Item and beets.dbcore.types
converter = Item._fields[beets_attr].model_type
except KeyError:
# Fall back to string (NOTE: probably not good)
converter = str
return converter
@staticmethod @staticmethod
def resource_object(track): def get_resource_object(lib: Library, track):
"""Construct a JSON:API resource object from a beets Item. """Construct a JSON:API resource object from a beets Item.
Args: Args:
@ -386,7 +423,7 @@ class TrackDocument(AURADocument):
Args: Args:
track_id: The beets id of the track (integer). track_id: The beets id of the track (integer).
""" """
track = current_app.config["lib"].get_item(track_id) track = self.lib.get_item(track_id)
if not track: if not track:
return self.error( return self.error(
"404 Not Found", "404 Not Found",
@ -395,12 +432,16 @@ class TrackDocument(AURADocument):
track_id track_id
), ),
) )
return self.single_resource_document(self.resource_object(track)) return self.single_resource_document(
self.get_resource_object(self.lib, track)
)
class AlbumDocument(AURADocument): class AlbumDocument(AURADocument):
"""Class for building documents for /albums endpoints.""" """Class for building documents for /albums endpoints."""
model_cls = Album
attribute_map = ALBUM_ATTR_MAP attribute_map = ALBUM_ATTR_MAP
def get_collection(self, query=None, sort=None): def get_collection(self, query=None, sort=None):
@ -410,26 +451,10 @@ class AlbumDocument(AURADocument):
query: A beets Query object or a beets query string. query: A beets Query object or a beets query string.
sort: A beets Sort object. sort: A beets Sort object.
""" """
return current_app.config["lib"].albums(query, sort) return self.lib.albums(query, sort)
def get_attribute_converter(self, beets_attr):
"""Work out what data type an attribute should be for beets.
Args:
beets_attr: The name of the beets attribute, e.g. "title".
"""
try:
# Look for field in list of Album fields
# and get python type of database type.
# See beets.library.Album and beets.dbcore.types
converter = Album._fields[beets_attr].model_type
except KeyError:
# Fall back to string (NOTE: probably not good)
converter = str
return converter
@staticmethod @staticmethod
def resource_object(album): def get_resource_object(lib: Library, album):
"""Construct a JSON:API resource object from a beets Album. """Construct a JSON:API resource object from a beets Album.
Args: Args:
@ -448,7 +473,7 @@ class AlbumDocument(AURADocument):
# track number. Sorting is not required but it's nice. # track number. Sorting is not required but it's nice.
query = MatchQuery("album_id", album.id) query = MatchQuery("album_id", album.id)
sort = FixedFieldSort("track", ascending=True) sort = FixedFieldSort("track", ascending=True)
tracks = current_app.config["lib"].items(query, sort) tracks = lib.items(query, sort)
# JSON:API one-to-many relationship to tracks on the album # JSON:API one-to-many relationship to tracks on the album
relationships = { relationships = {
"tracks": { "tracks": {
@ -484,7 +509,7 @@ class AlbumDocument(AURADocument):
Args: Args:
album_id: The beets id of the album (integer). album_id: The beets id of the album (integer).
""" """
album = current_app.config["lib"].get_album(album_id) album = self.lib.get_album(album_id)
if not album: if not album:
return self.error( return self.error(
"404 Not Found", "404 Not Found",
@ -493,12 +518,16 @@ class AlbumDocument(AURADocument):
album_id album_id
), ),
) )
return self.single_resource_document(self.resource_object(album)) return self.single_resource_document(
self.get_resource_object(self.lib, album)
)
class ArtistDocument(AURADocument): class ArtistDocument(AURADocument):
"""Class for building documents for /artists endpoints.""" """Class for building documents for /artists endpoints."""
model_cls = Item
attribute_map = ARTIST_ATTR_MAP attribute_map = ARTIST_ATTR_MAP
def get_collection(self, query=None, sort=None): def get_collection(self, query=None, sort=None):
@ -509,7 +538,7 @@ class ArtistDocument(AURADocument):
sort: A beets Sort object. sort: A beets Sort object.
""" """
# Gets only tracks with matching artist information # Gets only tracks with matching artist information
tracks = current_app.config["lib"].items(query, sort) tracks = self.lib.items(query, sort)
collection = [] collection = []
for track in tracks: for track in tracks:
# Do not add duplicates # Do not add duplicates
@ -517,24 +546,8 @@ class ArtistDocument(AURADocument):
collection.append(track.artist) collection.append(track.artist)
return collection return collection
def get_attribute_converter(self, beets_attr):
"""Work out what data type an attribute should be for beets.
Args:
beets_attr: The name of the beets attribute, e.g. "artist".
"""
try:
# Look for field in list of Item fields
# and get python type of database type.
# See beets.library.Item and beets.dbcore.types
converter = Item._fields[beets_attr].model_type
except KeyError:
# Fall back to string (NOTE: probably not good)
converter = str
return converter
@staticmethod @staticmethod
def resource_object(artist_id): def get_resource_object(lib: Library, artist_id):
"""Construct a JSON:API resource object for the given artist. """Construct a JSON:API resource object for the given artist.
Args: Args:
@ -542,7 +555,7 @@ class ArtistDocument(AURADocument):
""" """
# Get tracks where artist field exactly matches artist_id # Get tracks where artist field exactly matches artist_id
query = MatchQuery("artist", artist_id) query = MatchQuery("artist", artist_id)
tracks = current_app.config["lib"].items(query) tracks = lib.items(query)
if not tracks: if not tracks:
return None return None
@ -564,7 +577,7 @@ class ArtistDocument(AURADocument):
} }
} }
album_query = MatchQuery("albumartist", artist_id) album_query = MatchQuery("albumartist", artist_id)
albums = current_app.config["lib"].albums(query=album_query) albums = lib.albums(query=album_query)
if len(albums) != 0: if len(albums) != 0:
relationships["albums"] = { relationships["albums"] = {
"data": [{"type": "album", "id": str(a.id)} for a in albums] "data": [{"type": "album", "id": str(a.id)} for a in albums]
@ -583,7 +596,7 @@ class ArtistDocument(AURADocument):
Args: Args:
artist_id: A string which is the artist's name. artist_id: A string which is the artist's name.
""" """
artist_resource = self.resource_object(artist_id) artist_resource = self.get_resource_object(self.lib, artist_id)
if not artist_resource: if not artist_resource:
return self.error( return self.error(
"404 Not Found", "404 Not Found",
@ -616,8 +629,10 @@ def safe_filename(fn):
class ImageDocument(AURADocument): class ImageDocument(AURADocument):
"""Class for building documents for /images/(id) endpoints.""" """Class for building documents for /images/(id) endpoints."""
model_cls = Album
@staticmethod @staticmethod
def get_image_path(image_id): def get_image_path(lib: Library, image_id):
"""Works out the full path to the image with the given id. """Works out the full path to the image with the given id.
Returns None if there is no such image. Returns None if there is no such image.
@ -639,7 +654,7 @@ class ImageDocument(AURADocument):
# Get the path to the directory parent's images are in # Get the path to the directory parent's images are in
if parent_type == "album": if parent_type == "album":
album = current_app.config["lib"].get_album(int(parent_id)) album = lib.get_album(int(parent_id))
if not album or not album.artpath: if not album or not album.artpath:
return None return None
# Cut the filename off of artpath # Cut the filename off of artpath
@ -659,7 +674,7 @@ class ImageDocument(AURADocument):
return None return None
@staticmethod @staticmethod
def resource_object(image_id): def get_resource_object(lib: Library, image_id):
"""Construct a JSON:API resource object for the given image. """Construct a JSON:API resource object for the given image.
Args: Args:
@ -668,7 +683,7 @@ class ImageDocument(AURADocument):
""" """
# Could be called as a static method, so can't use # Could be called as a static method, so can't use
# self.get_image_path() # self.get_image_path()
image_path = ImageDocument.get_image_path(image_id) image_path = ImageDocument.get_image_path(lib, image_id)
if not image_path: if not image_path:
return None return None
@ -708,7 +723,7 @@ class ImageDocument(AURADocument):
image_id: A string in the form image_id: A string in the form
"<parent_type>-<parent_id>-<img_filename>". "<parent_type>-<parent_id>-<img_filename>".
""" """
image_resource = self.resource_object(image_id) image_resource = self.get_resource_object(self.lib, image_id)
if not image_resource: if not image_resource:
return self.error( return self.error(
"404 Not Found", "404 Not Found",
@ -736,8 +751,7 @@ def server_info():
@aura_bp.route("/tracks") @aura_bp.route("/tracks")
def all_tracks(): def all_tracks():
"""Respond with a list of all tracks and related information.""" """Respond with a list of all tracks and related information."""
doc = TrackDocument() return TrackDocument.from_app().all_resources()
return doc.all_resources()
@aura_bp.route("/tracks/<int:track_id>") @aura_bp.route("/tracks/<int:track_id>")
@ -747,8 +761,7 @@ def single_track(track_id):
Args: Args:
track_id: The id of the track provided in the URL (integer). track_id: The id of the track provided in the URL (integer).
""" """
doc = TrackDocument() return TrackDocument.from_app().single_resource(track_id)
return doc.single_resource(track_id)
@aura_bp.route("/tracks/<int:track_id>/audio") @aura_bp.route("/tracks/<int:track_id>/audio")
@ -820,8 +833,7 @@ def audio_file(track_id):
@aura_bp.route("/albums") @aura_bp.route("/albums")
def all_albums(): def all_albums():
"""Respond with a list of all albums and related information.""" """Respond with a list of all albums and related information."""
doc = AlbumDocument() return AlbumDocument.from_app().all_resources()
return doc.all_resources()
@aura_bp.route("/albums/<int:album_id>") @aura_bp.route("/albums/<int:album_id>")
@ -831,8 +843,7 @@ def single_album(album_id):
Args: Args:
album_id: The id of the album provided in the URL (integer). album_id: The id of the album provided in the URL (integer).
""" """
doc = AlbumDocument() return AlbumDocument.from_app().single_resource(album_id)
return doc.single_resource(album_id)
# Artist endpoints # Artist endpoints
@ -842,8 +853,7 @@ def single_album(album_id):
@aura_bp.route("/artists") @aura_bp.route("/artists")
def all_artists(): def all_artists():
"""Respond with a list of all artists and related information.""" """Respond with a list of all artists and related information."""
doc = ArtistDocument() return ArtistDocument.from_app().all_resources()
return doc.all_resources()
# Using the path converter allows slashes in artist_id # Using the path converter allows slashes in artist_id
@ -855,8 +865,7 @@ def single_artist(artist_id):
artist_id: The id of the artist provided in the URL. A string artist_id: The id of the artist provided in the URL. A string
which is the artist's name. which is the artist's name.
""" """
doc = ArtistDocument() return ArtistDocument.from_app().single_resource(artist_id)
return doc.single_resource(artist_id)
# Image endpoints # Image endpoints
@ -872,8 +881,7 @@ def single_image(image_id):
image_id: The id of the image provided in the URL. A string in image_id: The id of the image provided in the URL. A string in
the form "<parent_type>-<parent_id>-<img_filename>". the form "<parent_type>-<parent_id>-<img_filename>".
""" """
doc = ImageDocument() return ImageDocument.from_app().single_resource(image_id)
return doc.single_resource(image_id)
@aura_bp.route("/images/<string:image_id>/file") @aura_bp.route("/images/<string:image_id>/file")
@ -884,7 +892,7 @@ def image_file(image_id):
image_id: The id of the image provided in the URL. A string in image_id: The id of the image provided in the URL. A string in
the form "<parent_type>-<parent_id>-<img_filename>". the form "<parent_type>-<parent_id>-<img_filename>".
""" """
img_path = ImageDocument.get_image_path(image_id) img_path = ImageDocument.get_image_path(current_app.config["lib"], image_id)
if not img_path: if not img_path:
return AURADocument.error( return AURADocument.error(
"404 Not Found", "404 Not Found",

1
setup.py
View file

@ -108,6 +108,7 @@ setup(
"pylast", "pylast",
"pytest", "pytest",
"pytest-cov", "pytest-cov",
"pytest-flask",
"python-mpd2", "python-mpd2",
"python3-discogs-client>=2.3.15", "python3-discogs-client>=2.3.15",
"py7zr", "py7zr",

152
test/plugins/test_aura.py Normal file
View file

@ -0,0 +1,152 @@
import os
from http import HTTPStatus
from pathlib import Path
from typing import Any, Dict, Optional
import pytest
from flask.testing import Client
from beets.test.helper import TestHelper
@pytest.fixture(scope="session", autouse=True)
def helper():
helper = TestHelper()
helper.setup_beets()
yield helper
helper.teardown_beets()
@pytest.fixture(scope="session")
def app(helper):
from beetsplug.aura import create_app
app = create_app()
app.config["lib"] = helper.lib
return app
@pytest.fixture(scope="session")
def item(helper):
return helper.add_item_fixture(
album="Album",
title="Title",
artist="Artist",
albumartist="Album Artist",
)
@pytest.fixture(scope="session")
def album(helper, item):
return helper.lib.add_album([item])
@pytest.fixture(scope="session", autouse=True)
def _other_album_and_item(helper):
"""Add another item and album to prove that filtering works."""
item = helper.add_item_fixture(
album="Other Album",
title="Other Title",
artist="Other Artist",
albumartist="Other Album Artist",
)
helper.lib.add_album([item])
class TestAuraResponse:
@pytest.fixture
def get_response_data(self, client: Client, item):
"""Return a callback accepting `endpoint` and `params` parameters."""
def get(
endpoint: str, params: Dict[str, str]
) -> Optional[Dict[str, Any]]:
"""Add additional `params` and GET the given endpoint.
`include` parameter is added to every call to check that the
functionality that fetches related entities works.
Before returning the response data, ensure that the request is
successful.
"""
response = client.get(
endpoint,
query_string={"include": "tracks,artists,albums", **params},
)
assert response.status_code == HTTPStatus.OK
return response.json
return get
@pytest.fixture(scope="class")
def track_document(self, item, album):
return {
"type": "track",
"id": str(item.id),
"attributes": {
"album": item.album,
"albumartist": item.albumartist,
"artist": item.artist,
"size": Path(os.fsdecode(item.path)).stat().st_size,
"title": item.title,
},
"relationships": {
"albums": {"data": [{"id": str(album.id), "type": "album"}]},
"artists": {"data": [{"id": item.artist, "type": "artist"}]},
},
}
@pytest.fixture(scope="class")
def artist_document(self, item):
return {
"type": "artist",
"id": item.artist,
"attributes": {"name": item.artist},
"relationships": {
"tracks": {"data": [{"id": str(item.id), "type": "track"}]}
},
}
@pytest.fixture(scope="class")
def album_document(self, album):
return {
"type": "album",
"id": str(album.id),
"attributes": {"artist": album.albumartist, "title": album.album},
"relationships": {
"tracks": {"data": [{"id": str(album.id), "type": "track"}]}
},
}
def test_tracks(
self,
get_response_data,
item,
album_document,
artist_document,
track_document,
):
data = get_response_data("/aura/tracks", {"filter[title]": item.title})
assert data == {
"data": [track_document],
"included": [artist_document, album_document],
}
def test_artists(
self, get_response_data, item, artist_document, track_document
):
data = get_response_data(
"/aura/artists", {"filter[artist]": item.artist}
)
assert data == {"data": [artist_document], "included": [track_document]}
def test_albums(
self, get_response_data, album, album_document, track_document
):
data = get_response_data("/aura/albums", {"filter[album]": album.album})
assert data == {"data": [album_document], "included": [track_document]}