Hirad/pylast
2
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Merge pull request #336 from kvanzuijlen/master

Browse source
This commit is contained in:
Hugo van Kemenade 2020-12-30 12:55:30 +02:00 committed by GitHub
parent 66b4cb7d10 c851b82a1d
commit 34690f68cc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 231 additions and 193 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -103,7 +103,7 @@ export PYLAST_API_SECRET=TODO_ENTER_YOURS_HERE
To run all unit and integration tests: To run all unit and integration tests:
```sh ```sh
pip install pytest flaky pip install -e ".[tests]"
pytest pytest
``` ```

380
src/pylast/__init__.py
View file

@ -665,8 +665,7 @@ class LastFMNetwork(_Network):
password_hash="", password_hash="",
token="", token="",
): ):
_Network.__init__( super().__init__(
self,
name="Last.fm", name="Last.fm",
homepage="https://www.last.fm", homepage="https://www.last.fm",
ws_server=("ws.audioscrobbler.com", "/2.0/"), ws_server=("ws.audioscrobbler.com", "/2.0/"),
@ -733,8 +732,7 @@ class LibreFMNetwork(_Network):
self, api_key="", api_secret="", session_key="", username="", password_hash="" self, api_key="", api_secret="", session_key="", username="", password_hash=""
): ):
_Network.__init__( super().__init__(
self,
name="Libre.fm", name="Libre.fm",
homepage="https://libre.fm", homepage="https://libre.fm",
ws_server=("libre.fm", "/2.0/"), ws_server=("libre.fm", "/2.0/"),
@ -1146,21 +1144,29 @@ class _BaseObject:
return first_child.wholeText.strip() return first_child.wholeText.strip()
def _get_things(self, method, thing_type, params=None, cacheable=True): def _get_things(
"""Returns a list of the most played thing_types.""" self, method, thing_type, params=None, cacheable=True, stream=False
):
"""Returns a list of the most played thing_types by this thing."""
limit = params.get("limit", 50) def _stream_get_things():
seq = [] limit = params.get("limit", 50)
for node in _collect_nodes( nodes = _collect_nodes(
limit, self, self.ws_prefix + "." + method, cacheable, params limit,
): self,
title = _extract(node, "name") self.ws_prefix + "." + method,
artist = _extract(node, "name", 1) cacheable,
playcount = _number(_extract(node, "playcount")) params,
stream=stream,
)
for node in nodes:
title = _extract(node, "name")
artist = _extract(node, "name", 1)
playcount = _number(_extract(node, "playcount"))
seq.append(TopItem(thing_type(artist, title, self.network), playcount)) yield TopItem(thing_type(artist, title, self.network), playcount)
return seq return _stream_get_things() if stream else list(_stream_get_things())
def get_wiki_published_date(self): def get_wiki_published_date(self):
""" """
@ -1201,11 +1207,11 @@ class _BaseObject:
return _extract(node, section) return _extract(node, section)
class _Chartable: class _Chartable(_BaseObject):
"""Common functions for classes with charts.""" """Common functions for classes with charts."""
def __init__(self, ws_prefix): def __init__(self, network, ws_prefix):
self.ws_prefix = ws_prefix # TODO move to _BaseObject? super().__init__(network=network, ws_prefix=ws_prefix)
def get_weekly_chart_dates(self): def get_weekly_chart_dates(self):
"""Returns a list of From and To tuples for the available charts.""" """Returns a list of From and To tuples for the available charts."""
@ -1272,11 +1278,11 @@ class _Chartable:
return seq return seq
class _Taggable: class _Taggable(_BaseObject):
"""Common functions for classes with tags.""" """Common functions for classes with tags."""
def __init__(self, ws_prefix): def __init__(self, network, ws_prefix):
self.ws_prefix = ws_prefix # TODO move to _BaseObject super().__init__(network=network, ws_prefix=ws_prefix)
def add_tags(self, tags): def add_tags(self, tags):
"""Adds one or several tags. """Adds one or several tags.
@ -1381,9 +1387,9 @@ class _Taggable:
for element in elements: for element in elements:
tag_name = _extract(element, "name") tag_name = _extract(element, "name")
tagcount = _extract(element, "count") tag_count = _extract(element, "count")
seq.append(TopItem(Tag(tag_name, self.network), tagcount)) seq.append(TopItem(Tag(tag_name, self.network), tag_count))
if limit: if limit:
seq = seq[:limit] seq = seq[:limit]
@ -1459,7 +1465,7 @@ class NetworkError(Exception):
return "NetworkError: %s" % str(self.underlying_error) return "NetworkError: %s" % str(self.underlying_error)
class _Opus(_BaseObject, _Taggable): class _Opus(_Taggable):
"""An album or track.""" """An album or track."""
artist = None artist = None
@ -1480,8 +1486,7 @@ class _Opus(_BaseObject, _Taggable):
if info is None: if info is None:
info = {} info = {}
_BaseObject.__init__(self, network, ws_prefix) super().__init__(network=network, ws_prefix=ws_prefix)
_Taggable.__init__(self, ws_prefix)
if isinstance(artist, Artist): if isinstance(artist, Artist):
self.artist = artist self.artist = artist
@ -1649,7 +1654,7 @@ class Album(_Opus):
} }
class Artist(_BaseObject, _Taggable): class Artist(_Taggable):
"""An artist.""" """An artist."""
name = None name = None
@ -1666,8 +1671,7 @@ class Artist(_BaseObject, _Taggable):
if info is None: if info is None:
info = {} info = {}
_BaseObject.__init__(self, network, "artist") super().__init__(network=network, ws_prefix="artist")
_Taggable.__init__(self, "artist")
self.name = name self.name = name
self.username = username self.username = username
@ -1806,21 +1810,21 @@ class Artist(_BaseObject, _Taggable):
return artists return artists
def get_top_albums(self, limit=None, cacheable=True): def get_top_albums(self, limit=None, cacheable=True, stream=False):
"""Returns a list of the top albums.""" """Returns a list of the top albums."""
params = self._get_params() params = self._get_params()
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things("getTopAlbums", Album, params, cacheable) return self._get_things("getTopAlbums", Album, params, cacheable, stream=stream)
def get_top_tracks(self, limit=None, cacheable=True): def get_top_tracks(self, limit=None, cacheable=True, stream=False):
"""Returns a list of the most played Tracks by this artist.""" """Returns a list of the most played Tracks by this artist."""
params = self._get_params() params = self._get_params()
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things("getTopTracks", Track, params, cacheable) return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
def get_url(self, domain_name=DOMAIN_ENGLISH): def get_url(self, domain_name=DOMAIN_ENGLISH):
"""Returns the URL of the artist page on the network. """Returns the URL of the artist page on the network.
@ -1853,7 +1857,7 @@ class Country(_BaseObject):
__hash__ = _BaseObject.__hash__ __hash__ = _BaseObject.__hash__
def __init__(self, name, network): def __init__(self, name, network):
_BaseObject.__init__(self, network, "geo") super().__init__(network=network, ws_prefix="geo")
self.name = name self.name = name
@ -1888,13 +1892,13 @@ class Country(_BaseObject):
return _extract_top_artists(doc, self) return _extract_top_artists(doc, self)
def get_top_tracks(self, limit=None, cacheable=True): def get_top_tracks(self, limit=None, cacheable=True, stream=False):
"""Returns a sequence of the most played tracks""" """Returns a sequence of the most played tracks"""
params = self._get_params() params = self._get_params()
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things("getTopTracks", Track, params, cacheable) return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
def get_url(self, domain_name=DOMAIN_ENGLISH): def get_url(self, domain_name=DOMAIN_ENGLISH):
"""Returns the URL of the country page on the network. """Returns the URL of the country page on the network.
@ -1928,7 +1932,7 @@ class Library(_BaseObject):
__hash__ = _BaseObject.__hash__ __hash__ = _BaseObject.__hash__
def __init__(self, user, network): def __init__(self, user, network):
_BaseObject.__init__(self, network, "library") super().__init__(network=network, ws_prefix="library")
if isinstance(user, User): if isinstance(user, User):
self.user = user self.user = user
@ -1949,27 +1953,27 @@ class Library(_BaseObject):
"""Returns the user who owns this library.""" """Returns the user who owns this library."""
return self.user return self.user
def get_artists(self, limit=50, cacheable=True): def get_artists(self, limit=50, cacheable=True, stream=False):
""" """
Returns a sequence of Album objects Returns a sequence of Album objects
if limit==None it will return all (may take a while) if limit==None it will return all (may take a while)
""" """
seq = [] def _get_artists():
for node in _collect_nodes( for node in _collect_nodes(
limit, self, self.ws_prefix + ".getArtists", cacheable limit, self, self.ws_prefix + ".getArtists", cacheable, stream=stream
): ):
name = _extract(node, "name") name = _extract(node, "name")
playcount = _number(_extract(node, "playcount")) playcount = _number(_extract(node, "playcount"))
tagcount = _number(_extract(node, "tagcount")) tagcount = _number(_extract(node, "tagcount"))
seq.append(LibraryItem(Artist(name, self.network), playcount, tagcount)) yield LibraryItem(Artist(name, self.network), playcount, tagcount)
return seq return _get_artists() if stream else list(_get_artists())
class Tag(_BaseObject, _Chartable): class Tag(_Chartable):
"""A Last.fm object tag.""" """A Last.fm object tag."""
name = None name = None
@ -1977,8 +1981,7 @@ class Tag(_BaseObject, _Chartable):
__hash__ = _BaseObject.__hash__ __hash__ = _BaseObject.__hash__
def __init__(self, name, network): def __init__(self, name, network):
_BaseObject.__init__(self, network, "tag") super().__init__(network=network, ws_prefix="tag")
_Chartable.__init__(self, "tag")
self.name = name self.name = name
@ -2018,13 +2021,13 @@ class Tag(_BaseObject, _Chartable):
return _extract_top_albums(doc, self.network) return _extract_top_albums(doc, self.network)
def get_top_tracks(self, limit=None, cacheable=True): def get_top_tracks(self, limit=None, cacheable=True, stream=False):
"""Returns a list of the most played Tracks for this tag.""" """Returns a list of the most played Tracks for this tag."""
params = self._get_params() params = self._get_params()
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things("getTopTracks", Track, params, cacheable) return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
def get_top_artists(self, limit=None, cacheable=True): def get_top_artists(self, limit=None, cacheable=True):
"""Returns a sequence of the most played artists.""" """Returns a sequence of the most played artists."""
@ -2180,7 +2183,7 @@ class Track(_Opus):
} }
class User(_BaseObject, _Chartable): class User(_Chartable):
"""A Last.fm user.""" """A Last.fm user."""
name = None name = None
@ -2188,8 +2191,7 @@ class User(_BaseObject, _Chartable):
__hash__ = _BaseObject.__hash__ __hash__ = _BaseObject.__hash__
def __init__(self, user_name, network): def __init__(self, user_name, network):
_BaseObject.__init__(self, network, "user") super().__init__(network=network, ws_prefix="user")
_Chartable.__init__(self, "user")
self.name = user_name self.name = user_name
@ -2212,6 +2214,16 @@ class User(_BaseObject, _Chartable):
def _get_params(self): def _get_params(self):
return {self.ws_prefix: self.get_name()} return {self.ws_prefix: self.get_name()}
def _extract_played_track(self, track_node):
title = _extract(track_node, "name")
track_artist = _extract(track_node, "artist")
date = _extract(track_node, "date")
album = _extract(track_node, "album")
timestamp = track_node.getElementsByTagName("date")[0].getAttribute("uts")
return PlayedTrack(
Track(track_artist, title, self.network), album, date, timestamp
)
def get_name(self, properly_capitalized=False): def get_name(self, properly_capitalized=False):
"""Returns the user name.""" """Returns the user name."""
@ -2222,47 +2234,53 @@ class User(_BaseObject, _Chartable):
return self.name return self.name
def get_friends(self, limit=50, cacheable=False): def get_friends(self, limit=50, cacheable=False, stream=False):
"""Returns a list of the user's friends. """ """Returns a list of the user's friends. """
seq = [] def _get_friends():
for node in _collect_nodes( for node in _collect_nodes(
limit, self, self.ws_prefix + ".getFriends", cacheable limit, self, self.ws_prefix + ".getFriends", cacheable, stream=stream
): ):
seq.append(User(_extract(node, "name"), self.network)) yield User(_extract(node, "name"), self.network)
return seq return _get_friends() if stream else list(_get_friends())
def get_loved_tracks(self, limit=50, cacheable=True): def get_loved_tracks(self, limit=50, cacheable=True, stream=False):
""" """
Returns this user's loved track as a sequence of LovedTrack objects in Returns this user's loved track as a sequence of LovedTrack objects in
reverse order of their timestamp, all the way back to the first track. reverse order of their timestamp, all the way back to the first track.
If limit==None, it will try to pull all the available data. If limit==None, it will try to pull all the available data.
If stream=True, it will yield tracks as soon as a page has been retrieved.
This method uses caching. Enable caching only if you're pulling a This method uses caching. Enable caching only if you're pulling a
large amount of data. large amount of data.
""" """
params = self._get_params() def _get_loved_tracks():
if limit: params = self._get_params()
params["limit"] = limit if limit:
params["limit"] = limit
seq = [] for track in _collect_nodes(
for track in _collect_nodes( limit,
limit, self, self.ws_prefix + ".getLovedTracks", cacheable, params self,
): self.ws_prefix + ".getLovedTracks",
try: cacheable,
artist = _extract(track, "name", 1) params,
except IndexError: # pragma: no cover stream=stream,
continue ):
title = _extract(track, "name") try:
date = _extract(track, "date") artist = _extract(track, "name", 1)
timestamp = track.getElementsByTagName("date")[0].getAttribute("uts") except IndexError: # pragma: no cover
continue
title = _extract(track, "name")
date = _extract(track, "date")
timestamp = track.getElementsByTagName("date")[0].getAttribute("uts")
seq.append(LovedTrack(Track(artist, title, self.network), date, timestamp)) yield LovedTrack(Track(artist, title, self.network), date, timestamp)
return seq return _get_loved_tracks() if stream else list(_get_loved_tracks())
def get_now_playing(self): def get_now_playing(self):
""" """
@ -2290,7 +2308,15 @@ class User(_BaseObject, _Chartable):
return Track(artist, title, self.network, self.name, info=info) return Track(artist, title, self.network, self.name, info=info)
def get_recent_tracks(self, limit=10, cacheable=True, time_from=None, time_to=None): def get_recent_tracks(
self,
limit=10,
cacheable=True,
time_from=None,
time_to=None,
stream=False,
now_playing=False,
):
""" """
Returns this user's played track as a sequence of PlayedTrack objects Returns this user's played track as a sequence of PlayedTrack objects
in reverse order of playtime, all the way back to the first track. in reverse order of playtime, all the way back to the first track.
@ -2305,45 +2331,39 @@ class User(_BaseObject, _Chartable):
before this time, in UNIX timestamp format (integer number of before this time, in UNIX timestamp format (integer number of
seconds since 00:00:00, January 1st 1970 UTC). This must be in seconds since 00:00:00, January 1st 1970 UTC). This must be in
the UTC time zone. the UTC time zone.
stream: If True, it will yield tracks as soon as a page has been retrieved.
This method uses caching. Enable caching only if you're pulling a This method uses caching. Enable caching only if you're pulling a
large amount of data. large amount of data.
""" """
params = self._get_params() def _get_recent_tracks():
if limit: params = self._get_params()
params["limit"] = limit + 1 # in case we remove the now playing track if limit:
if time_from: params["limit"] = limit + 1 # in case we remove the now playing track
params["from"] = time_from if time_from:
if time_to: params["from"] = time_from
params["to"] = time_to if time_to:
params["to"] = time_to
seq = [] track_count = 0
for track in _collect_nodes( for track_node in _collect_nodes(
limit + 1 if limit else None, limit + 1 if limit else None,
self, self,
self.ws_prefix + ".getRecentTracks", self.ws_prefix + ".getRecentTracks",
cacheable, cacheable,
params, params,
): stream=stream,
):
if track_node.hasAttribute("nowplaying") and not now_playing:
continue # to prevent the now playing track from sneaking in
if track.hasAttribute("nowplaying"): if limit and track_count >= limit:
continue # to prevent the now playing track from sneaking in break
yield self._extract_played_track(track_node=track_node)
track_count += 1
title = _extract(track, "name") return _get_recent_tracks() if stream else list(_get_recent_tracks())
artist = _extract(track, "artist")
date = _extract(track, "date")
album = _extract(track, "album")
timestamp = track.getElementsByTagName("date")[0].getAttribute("uts")
seq.append(
PlayedTrack(Track(artist, title, self.network), album, date, timestamp)
)
if limit:
# Slice, in case we didn't remove a now playing track
seq = seq[:limit]
return seq
def get_country(self): def get_country(self):
"""Returns the name of the country of the user.""" """Returns the name of the country of the user."""
@ -2482,7 +2502,9 @@ class User(_BaseObject, _Chartable):
return seq return seq
def get_top_tracks(self, period=PERIOD_OVERALL, limit=None, cacheable=True): def get_top_tracks(
self, period=PERIOD_OVERALL, limit=None, cacheable=True, stream=False
):
"""Returns the top tracks played by a user. """Returns the top tracks played by a user.
* period: The period of time. Possible values: * period: The period of time. Possible values:
o PERIOD_OVERALL o PERIOD_OVERALL
@ -2498,33 +2520,29 @@ class User(_BaseObject, _Chartable):
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things("getTopTracks", Track, params, cacheable) return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
def get_track_scrobbles(self, artist, track, cacheable=False): def get_track_scrobbles(self, artist, track, cacheable=False, stream=False):
""" """
Get a list of this user's scrobbles of this artist's track, Get a list of this user's scrobbles of this artist's track,
including scrobble time. including scrobble time.
""" """
params = self._get_params() params = self._get_params()
params["artist"] = artist params["artist"] = artist
params["track"] = track params["track"] = track
seq = [] def _get_track_scrobbles():
for track in _collect_nodes( for track_node in _collect_nodes(
None, self, self.ws_prefix + ".getTrackScrobbles", cacheable, params None,
): self,
title = _extract(track, "name") self.ws_prefix + ".getTrackScrobbles",
artist = _extract(track, "artist") cacheable,
date = _extract(track, "date") params,
album = _extract(track, "album") stream=stream,
timestamp = track.getElementsByTagName("date")[0].getAttribute("uts") ):
yield self._extract_played_track(track_node)
seq.append( return _get_track_scrobbles() if stream else list(_get_track_scrobbles())
PlayedTrack(Track(artist, title, self.network), album, date, timestamp)
)
return seq
def get_image(self, size=SIZE_EXTRA_LARGE): def get_image(self, size=SIZE_EXTRA_LARGE):
""" """
@ -2569,21 +2587,21 @@ class User(_BaseObject, _Chartable):
class AuthenticatedUser(User): class AuthenticatedUser(User):
def __init__(self, network): def __init__(self, network):
User.__init__(self, network.username, network) super().__init__(user_name=network.username, network=network)
def _get_params(self): def _get_params(self):
return {"user": self.get_name()} return {"user": self.get_name()}
def get_name(self): def get_name(self, properly_capitalized=False):
"""Returns the name of the authenticated user.""" """Returns the name of the authenticated user."""
return self.name return super().get_name(properly_capitalized=properly_capitalized)
class _Search(_BaseObject): class _Search(_BaseObject):
"""An abstract class. Use one of its derivatives.""" """An abstract class. Use one of its derivatives."""
def __init__(self, ws_prefix, search_terms, network): def __init__(self, ws_prefix, search_terms, network):
_BaseObject.__init__(self, network, ws_prefix) super().__init__(network, ws_prefix)
self._ws_prefix = ws_prefix self._ws_prefix = ws_prefix
self.search_terms = search_terms self.search_terms = search_terms
@ -2623,8 +2641,9 @@ class AlbumSearch(_Search):
"""Search for an album by name.""" """Search for an album by name."""
def __init__(self, album_name, network): def __init__(self, album_name, network):
super().__init__(
_Search.__init__(self, "album", {"album": album_name}, network) ws_prefix="album", search_terms={"album": album_name}, network=network
)
def get_next_page(self): def get_next_page(self):
"""Returns the next page of results as a sequence of Album objects.""" """Returns the next page of results as a sequence of Album objects."""
@ -2649,7 +2668,9 @@ class ArtistSearch(_Search):
"""Search for an artist by artist name.""" """Search for an artist by artist name."""
def __init__(self, artist_name, network): def __init__(self, artist_name, network):
_Search.__init__(self, "artist", {"artist": artist_name}, network) super().__init__(
ws_prefix="artist", search_terms={"artist": artist_name}, network=network
)
def get_next_page(self): def get_next_page(self):
"""Returns the next page of results as a sequence of Artist objects.""" """Returns the next page of results as a sequence of Artist objects."""
@ -2676,9 +2697,10 @@ class TrackSearch(_Search):
""" """
def __init__(self, artist_name, track_title, network): def __init__(self, artist_name, track_title, network):
super().__init__(
_Search.__init__( ws_prefix="track",
self, "track", {"track": track_title, "artist": artist_name}, network search_terms={"track": track_title, "artist": artist_name},
network=network,
) )
def get_next_page(self): def get_next_page(self):
@ -2734,59 +2756,59 @@ def cleanup_nodes(doc):
return doc return doc
def _collect_nodes(limit, sender, method_name, cacheable, params=None): def _collect_nodes(limit, sender, method_name, cacheable, params=None, stream=False):
""" """
Returns a sequence of dom.Node objects about as close to limit as possible Returns a sequence of dom.Node objects about as close to limit as possible
""" """
if not params: if not params:
params = sender._get_params() params = sender._get_params()
nodes = [] def _stream_collect_nodes():
page = 1 node_count = 0
end_of_pages = False page = 1
end_of_pages = False
while not end_of_pages and (not limit or (limit and len(nodes) < limit)): while not end_of_pages and (not limit or (limit and node_count < limit)):
params["page"] = str(page) params["page"] = str(page)
tries = 1 tries = 1
while True: while True:
try: try:
doc = sender._request(method_name, cacheable, params) doc = sender._request(method_name, cacheable, params)
break # success break # success
except Exception as e: except Exception as e:
if tries >= 3: if tries >= 3:
raise e raise e
# Wait and try again # Wait and try again
time.sleep(1) time.sleep(1)
tries += 1 tries += 1
doc = cleanup_nodes(doc) doc = cleanup_nodes(doc)
# break if there are no child nodes # break if there are no child nodes
if not doc.documentElement.childNodes: if not doc.documentElement.childNodes:
break break
main = doc.documentElement.childNodes[0] main = doc.documentElement.childNodes[0]
if main.hasAttribute("totalPages"): if main.hasAttribute("totalPages") or main.hasAttribute("totalpages"):
total_pages = _number(main.getAttribute("totalPages")) total_pages = _number(
elif main.hasAttribute("totalpages"): main.getAttribute("totalPages") or main.getAttribute("totalpages")
total_pages = _number(main.getAttribute("totalpages")) )
else: else:
raise Exception("No total pages attribute") raise Exception("No total pages attribute")
for node in main.childNodes: for node in main.childNodes:
if not node.nodeType == xml.dom.Node.TEXT_NODE and ( if not node.nodeType == xml.dom.Node.TEXT_NODE and (
not limit or (len(nodes) < limit) not limit or (node_count < limit)
): ):
nodes.append(node) node_count += 1
yield node
if page >= total_pages: end_of_pages = page >= total_pages
end_of_pages = True
page += 1 page += 1
return nodes return _stream_collect_nodes() if stream else list(_stream_collect_nodes())
def _extract(node, name, index=0): def _extract(node, name, index=0):
@ -2880,8 +2902,6 @@ def _number(string):
if not string: if not string:
return 0 return 0
elif string == "":
return 0
else: else:
try: try:
return int(string) return int(string)

2
tests/test_album.py
View file

@ -32,7 +32,7 @@ class TestPyLastAlbum(TestPyLastWithLastFm):
# Act # Act
# limit=2 to ignore now-playing: # limit=2 to ignore now-playing:
track = lastfm_user.get_recent_tracks(limit=2)[0] track = list(lastfm_user.get_recent_tracks(limit=2))[0]
# Assert # Assert
assert hasattr(track, "album") assert hasattr(track, "album")

2
tests/test_artist.py
View file

@ -90,7 +90,7 @@ class TestPyLastArtist(TestPyLastWithLastFm):
artist = self.network.get_top_artists(limit=1)[0].item artist = self.network.get_top_artists(limit=1)[0].item
# Act # Act
things = artist.get_top_albums(limit=2) things = list(artist.get_top_albums(limit=2))
# Assert # Assert
self.helper_two_different_things_in_top_list(things, pylast.Album) self.helper_two_different_things_in_top_list(things, pylast.Album)

2
tests/test_network.py
View file

@ -27,7 +27,7 @@ class TestPyLastNetwork(TestPyLastWithLastFm):
# Assert # Assert
# limit=2 to ignore now-playing: # limit=2 to ignore now-playing:
last_scrobble = lastfm_user.get_recent_tracks(limit=2)[0] last_scrobble = list(lastfm_user.get_recent_tracks(limit=2))[0]
assert str(last_scrobble.track.artist).lower() == artist assert str(last_scrobble.track.artist).lower() == artist
assert str(last_scrobble.track.title).lower() == title assert str(last_scrobble.track.title).lower() == title

8
tests/test_pylast.py
View file

@ -88,9 +88,9 @@ class TestPyLastWithLastFm(PyLastTestCase):
assert a is not None assert a is not None
assert b is not None assert b is not None
assert c is not None assert c is not None
assert len(a) >= 0 assert isinstance(len(a), int)
assert len(b) >= 0 assert isinstance(len(b), int)
assert len(c) >= 0 assert isinstance(len(c), int)
assert a == b assert a == b
assert b == c assert b == c
@ -102,7 +102,7 @@ class TestPyLastWithLastFm(PyLastTestCase):
# Act # Act
result1 = func(limit=1, cacheable=False) result1 = func(limit=1, cacheable=False)
result2 = func(limit=1, cacheable=True) result2 = func(limit=1, cacheable=True)
result3 = func(limit=1) result3 = list(func(limit=1))
# Assert # Assert
self.helper_validate_results(result1, result2, result3) self.helper_validate_results(result1, result2, result3)

6
tests/test_track.py
View file

@ -24,7 +24,7 @@ class TestPyLastTrack(TestPyLastWithLastFm):
track.love() track.love()
# Assert # Assert
loved = lastfm_user.get_loved_tracks(limit=1) loved = list(lastfm_user.get_loved_tracks(limit=1))
assert str(loved[0].track.artist).lower() == "test artist" assert str(loved[0].track.artist).lower() == "test artist"
assert str(loved[0].track.title).lower() == "test title" assert str(loved[0].track.title).lower() == "test title"
@ -42,7 +42,7 @@ class TestPyLastTrack(TestPyLastWithLastFm):
time.sleep(1) # Delay, for Last.fm latency. TODO Can this be removed later? time.sleep(1) # Delay, for Last.fm latency. TODO Can this be removed later?
# Assert # Assert
loved = lastfm_user.get_loved_tracks(limit=1) loved = list(lastfm_user.get_loved_tracks(limit=1))
if len(loved): # OK to be empty but if not: if len(loved): # OK to be empty but if not:
assert str(loved[0].track.artist) != "Test Artist" assert str(loved[0].track.artist) != "Test Artist"
assert str(loved[0].track.title) != "test title" assert str(loved[0].track.title) != "test title"
@ -80,7 +80,7 @@ class TestPyLastTrack(TestPyLastWithLastFm):
def test_track_is_hashable(self): def test_track_is_hashable(self):
# Arrange # Arrange
artist = self.network.get_artist("Test Artist") artist = self.network.get_artist("Test Artist")
track = artist.get_top_tracks()[0].item track = artist.get_top_tracks(stream=False)[0].item
assert isinstance(track, pylast.Track) assert isinstance(track, pylast.Track)
# Act/Assert # Act/Assert

22
tests/test_user.py
View file

@ -4,6 +4,7 @@ Integration (not unit) tests for pylast.py
""" """
import calendar import calendar
import datetime as dt import datetime as dt
import inspect
import os import os
import re import re
@ -392,6 +393,23 @@ class TestPyLastUser(TestPyLastWithLastFm):
assert str(tracks[0].track.artist) == "Seun Kuti & Egypt 80" assert str(tracks[0].track.artist) == "Seun Kuti & Egypt 80"
assert str(tracks[0].track.title) == "Struggles Sounds" assert str(tracks[0].track.title) == "Struggles Sounds"
def test_get_recent_tracks_is_streamable(self):
# Arrange
lastfm_user = self.network.get_user("bbc6music")
start = dt.datetime(2020, 2, 15, 15, 00)
end = dt.datetime(2020, 2, 15, 15, 40)
utc_start = calendar.timegm(start.utctimetuple())
utc_end = calendar.timegm(end.utctimetuple())
# Act
tracks = lastfm_user.get_recent_tracks(
time_from=utc_start, time_to=utc_end, limit=None, stream=True
)
# Assert
assert inspect.isgenerator(tracks)
def test_get_playcount(self): def test_get_playcount(self):
# Arrange # Arrange
user = self.network.get_user("RJ") user = self.network.get_user("RJ")
@ -469,8 +487,8 @@ class TestPyLastUser(TestPyLastWithLastFm):
# Act # Act
result1 = user.get_track_scrobbles(artist, title, cacheable=False) result1 = user.get_track_scrobbles(artist, title, cacheable=False)
result2 = user.get_track_scrobbles(artist, title, cacheable=True) result2 = list(user.get_track_scrobbles(artist, title, cacheable=True))
result3 = user.get_track_scrobbles(artist, title) result3 = list(user.get_track_scrobbles(artist, title))
# Assert # Assert
self.helper_validate_results(result1, result2, result3) self.helper_validate_results(result1, result2, result3)