Merge branch 'streaming'

This commit is contained in:
Koen van Zuijlen 2020-12-29 20:59:29 +01:00
commit 421c80a617
5 changed files with 33 additions and 58 deletions

View file

@ -1147,12 +1147,12 @@ class _BaseObject:
return first_child.wholeText.strip() return first_child.wholeText.strip()
def _get_things( def _get_things(
self, method, thing, thing_type, params=None, cacheable=True, stream=False self, method, thing_type, params=None, cacheable=True, stream=False
): ):
"""Returns a list of the most played thing_types by this thing.""" """Returns a list of the most played thing_types by this thing."""
def _stream_get_things(): def _stream_get_things():
limit = params.get("limit", 1) limit = params.get("limit", 50)
nodes = _collect_nodes( nodes = _collect_nodes(
limit, limit,
self, self,
@ -1818,9 +1818,7 @@ class Artist(_Taggable):
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things( return self._get_things("getTopAlbums", Album, params, cacheable, stream=stream)
"getTopAlbums", "album", Album, params, cacheable, stream=stream
)
def get_top_tracks(self, limit=None, cacheable=True, stream=False): 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."""
@ -1828,9 +1826,7 @@ class Artist(_Taggable):
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things( return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
"getTopTracks", "track", 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.
@ -1904,9 +1900,7 @@ class Country(_BaseObject):
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things( return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
"getTopTracks", "track", 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.
@ -2035,9 +2029,7 @@ class Tag(_Chartable):
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things( return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
"getTopTracks", "track", 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."""
@ -2244,7 +2236,7 @@ class User(_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. """
def _get_friends(): def _get_friends():
@ -2261,7 +2253,7 @@ class User(_Chartable):
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=False, it will yield tracks as soon as a page has been retrieved. 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.
@ -2530,9 +2522,7 @@ class User(_Chartable):
if limit: if limit:
params["limit"] = limit params["limit"] = limit
return self._get_things( return self._get_things("getTopTracks", Track, params, cacheable, stream=stream)
"getTopTracks", "track", Track, params, cacheable, stream=stream
)
def get_track_scrobbles(self, artist, track, cacheable=False, stream=False): def get_track_scrobbles(self, artist, track, cacheable=False, stream=False):
""" """

View file

@ -79,7 +79,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_tracks(limit=2, stream=False) things = artist.get_top_tracks(limit=2)
# Assert # Assert
self.helper_two_different_things_in_top_list(things, pylast.Track) self.helper_two_different_things_in_top_list(things, pylast.Track)
@ -95,42 +95,29 @@ class TestPyLastArtist(TestPyLastWithLastFm):
# Assert # Assert
self.helper_two_different_things_in_top_list(things, pylast.Album) self.helper_two_different_things_in_top_list(things, pylast.Album)
def test_artist_top_albums_limit_1(self): @pytest.mark.parametrize("test_limit", [1, 50, 100])
def test_artist_top_albums_limit(self, test_limit: int) -> None:
# Arrange # Arrange
limit = 1
# Pick an artist with plenty of plays # Pick an artist with plenty of plays
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=limit, stream=False) things = artist.get_top_albums(limit=test_limit)
# Assert # Assert
assert len(things) == 1 assert len(things) == test_limit
def test_artist_top_albums_limit_50(self): def test_artist_top_albums_limit_default(self):
# Arrange # Arrange
limit = 50
# Pick an artist with plenty of plays # Pick an artist with plenty of plays
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=limit, stream=False) things = artist.get_top_albums()
# Assert # Assert
assert len(things) == 50 assert len(things) == 50
def test_artist_top_albums_limit_100(self):
# Arrange
limit = 100
# Pick an artist with plenty of plays
artist = self.network.get_top_artists(limit=1)[0].item
# Act
things = list(artist.get_top_albums(limit=limit))
# Assert
assert len(things) == 100
def test_artist_listener_count(self): def test_artist_listener_count(self):
# Arrange # Arrange
artist = self.network.get_artist("Test Artist") artist = self.network.get_artist("Test Artist")

View file

@ -154,7 +154,7 @@ class TestPyLastNetwork(TestPyLastWithLastFm):
country = self.network.get_country("Croatia") country = self.network.get_country("Croatia")
# Act # Act
things = country.get_top_tracks(limit=2, stream=False) things = country.get_top_tracks(limit=2)
# Assert # Assert
self.helper_two_different_things_in_top_list(things, pylast.Track) self.helper_two_different_things_in_top_list(things, pylast.Track)
@ -172,7 +172,7 @@ class TestPyLastNetwork(TestPyLastWithLastFm):
tag = self.network.get_tag("blues") tag = self.network.get_tag("blues")
# Act # Act
things = tag.get_top_tracks(limit=2, stream=False) things = tag.get_top_tracks(limit=2)
# Assert # Assert
self.helper_two_different_things_in_top_list(things, pylast.Track) self.helper_two_different_things_in_top_list(things, pylast.Track)

View file

@ -34,11 +34,11 @@ def load_secrets(): # pragma: no cover
class PyLastTestCase: class PyLastTestCase:
def assert_startswith(self, str, prefix, start=None, end=None): def assert_startswith(self, s, prefix, start=None, end=None):
assert str.startswith(prefix, start, end) assert s.startswith(prefix, start, end)
def assert_endswith(self, str, suffix, start=None, end=None): def assert_endswith(self, s, suffix, start=None, end=None):
assert str.endswith(suffix, start, end) assert s.endswith(suffix, start, end)
def _no_xfail_rerun_filter(err, name, test, plugin): def _no_xfail_rerun_filter(err, name, test, plugin):
@ -100,8 +100,8 @@ class TestPyLastWithLastFm(PyLastTestCase):
func = getattr(thing, function_name, None) func = getattr(thing, function_name, None)
# Act # Act
result1 = func(limit=1, cacheable=False, stream=False) result1 = func(limit=1, cacheable=False)
result2 = func(limit=1, cacheable=True, stream=False) result2 = func(limit=1, cacheable=True)
result3 = list(func(limit=1)) result3 = list(func(limit=1))
# Assert # Assert

View file

@ -143,10 +143,10 @@ class TestPyLastUser(TestPyLastWithLastFm):
user = self.network.get_user("test-user") user = self.network.get_user("test-user")
# Act/Assert # Act/Assert
assert len(user.get_loved_tracks(limit=20, stream=False)) == 20 assert len(user.get_loved_tracks(limit=20)) == 20
assert len(user.get_loved_tracks(limit=100, stream=False)) <= 100 assert len(user.get_loved_tracks(limit=100)) <= 100
assert len(user.get_loved_tracks(limit=None, stream=False)) >= 23 assert len(user.get_loved_tracks(limit=None)) >= 23
assert len(user.get_loved_tracks(limit=0, stream=False)) >= 23 assert len(user.get_loved_tracks(limit=0)) >= 23
def test_user_is_hashable(self): def test_user_is_hashable(self):
# Arrange # Arrange
@ -216,7 +216,7 @@ class TestPyLastUser(TestPyLastWithLastFm):
lastfm_user = self.network.get_user("RJ") lastfm_user = self.network.get_user("RJ")
# Act # Act
things = lastfm_user.get_top_tracks(limit=2, stream=False) things = lastfm_user.get_top_tracks(limit=2)
# Assert # Assert
self.helper_two_different_things_in_top_list(things, pylast.Track) self.helper_two_different_things_in_top_list(things, pylast.Track)
@ -367,9 +367,7 @@ class TestPyLastUser(TestPyLastWithLastFm):
utc_end = calendar.timegm(end.utctimetuple()) utc_end = calendar.timegm(end.utctimetuple())
# Act # Act
tracks = lastfm_user.get_recent_tracks( tracks = lastfm_user.get_recent_tracks(time_from=utc_start, time_to=utc_end)
time_from=utc_start, time_to=utc_end, stream=False
)
# Assert # Assert
assert len(tracks) == 1 assert len(tracks) == 1
@ -387,7 +385,7 @@ class TestPyLastUser(TestPyLastWithLastFm):
# Act # Act
tracks = lastfm_user.get_recent_tracks( tracks = lastfm_user.get_recent_tracks(
time_from=utc_start, time_to=utc_end, limit=None, stream=False time_from=utc_start, time_to=utc_end, limit=None
) )
# Assert # Assert
@ -474,7 +472,7 @@ class TestPyLastUser(TestPyLastWithLastFm):
user = self.network.get_user("bbc6music") user = self.network.get_user("bbc6music")
# Act # Act
scrobbles = user.get_track_scrobbles(artist, title, stream=False) scrobbles = user.get_track_scrobbles(artist, title)
# Assert # Assert
assert len(scrobbles) > 0 assert len(scrobbles) > 0
@ -488,7 +486,7 @@ class TestPyLastUser(TestPyLastWithLastFm):
user = self.network.get_user("bbc6music") user = self.network.get_user("bbc6music")
# Act # Act
result1 = user.get_track_scrobbles(artist, title, cacheable=False, stream=False) result1 = user.get_track_scrobbles(artist, title, cacheable=False)
result2 = list(user.get_track_scrobbles(artist, title, cacheable=True)) result2 = list(user.get_track_scrobbles(artist, title, cacheable=True))
result3 = list(user.get_track_scrobbles(artist, title)) result3 = list(user.get_track_scrobbles(artist, title))