From a021e4c09bebd0e49a54789f81dd49421256ad16 Mon Sep 17 00:00:00 2001 From: Marvin Steadfast Date: Tue, 6 Dec 2016 16:43:19 +0100 Subject: [PATCH] major refactoring --- README.rst | 5 + mopidy_emby/__init__.py | 2 +- mopidy_emby/backend.py | 556 +--------------------------------------- mopidy_emby/library.py | 86 +++++++ mopidy_emby/playback.py | 26 ++ mopidy_emby/remote.py | 464 +++++++++++++++++++++++++++++++++ tests/conftest.py | 12 +- tests/test_extension.py | 474 +--------------------------------- tests/test_library.py | 84 ++++++ tests/test_playback.py | 17 ++ tests/test_remote.py | 382 +++++++++++++++++++++++++++ 11 files changed, 1076 insertions(+), 1032 deletions(-) create mode 100644 mopidy_emby/library.py create mode 100644 mopidy_emby/playback.py create mode 100644 mopidy_emby/remote.py create mode 100644 tests/test_library.py create mode 100644 tests/test_playback.py create mode 100644 tests/test_remote.py diff --git a/README.rst b/README.rst index c05a3b4..4695dd3 100644 --- a/README.rst +++ b/README.rst @@ -71,6 +71,11 @@ Credits Changelog ========= +v0.2.0 +--------------------------------------- + +- Alot of splitting and refactoring + v0.1.3 ---------------------------------------- diff --git a/mopidy_emby/__init__.py b/mopidy_emby/__init__.py index d43c432..c8cd168 100644 --- a/mopidy_emby/__init__.py +++ b/mopidy_emby/__init__.py @@ -6,7 +6,7 @@ import os from mopidy import config, ext -__version__ = '0.1.3' +__version__ = '0.2.0' logger = logging.getLogger(__name__) diff --git a/mopidy_emby/backend.py b/mopidy_emby/backend.py index 11be507..4c25be9 100644 --- a/mopidy_emby/backend.py +++ b/mopidy_emby/backend.py @@ -1,20 +1,14 @@ from __future__ import unicode_literals -import hashlib import logging -import time -from urllib import urlencode -from urllib2 import quote -from urlparse import parse_qs, urljoin, urlsplit, urlunsplit - -from mopidy import backend, httpclient, models +from mopidy import backend import pykka -import requests - -import mopidy_emby +from mopidy_emby.library import EmbyLibraryProvider +from mopidy_emby.playback import EmbyPlaybackProvider +from mopidy_emby.remote import EmbyHandler logger = logging.getLogger(__name__) @@ -30,545 +24,3 @@ class EmbyBackend(pykka.ThreadingActor, backend.Backend): self.playback = EmbyPlaybackProvider(audio=audio, backend=self) self.playlist = None self.remote = EmbyHandler(config) - - -class EmbyPlaybackProvider(backend.PlaybackProvider): - - def translate_uri(self, uri): - if uri.startswith('emby:track:') and len(uri.split(':')) == 3: - id = uri.split(':')[-1] - - track_url = self.backend.remote.api_url( - '/Audio/{}/stream.mp3'.format(id) - ) - - logger.debug('Emby track streaming url: {}'.format(track_url)) - - return track_url - - else: - return None - - -class EmbyLibraryProvider(backend.LibraryProvider): - - root_directory = models.Ref.directory(uri='emby:', - name='Emby') - - def browse(self, uri): - # artistlist - if uri == self.root_directory.uri: - logger.debug('Get Emby artist list') - return self.backend.remote.get_artists() - - # split uri - parts = uri.split(':') - - # artists albums - # uri: emby:artist: - if uri.startswith('emby:artist:') and len(parts) == 3: - logger.debug('Get Emby album list') - artist_id = parts[-1] - - return self.backend.remote.get_albums(artist_id) - - # tracklist - # uri: emby:album: - if uri.startswith('emby:album:') and len(parts) == 3: - logger.debug('Get Emby track list') - album_id = parts[-1] - - return self.backend.remote.get_tracks(album_id) - - return [] - - def lookup(self, uri=None, uris=None): - logger.debug('Emby lookup: {}'.format(uri or uris)) - if uri: - parts = uri.split(':') - - if uri.startswith('emby:track:') and len(parts) == 3: - track_id = parts[-1] - tracks = [self.backend.remote.get_track(track_id)] - - elif uri.startswith('emby:album:') and len(parts) == 3: - album_id = parts[-1] - album_data = self.backend.remote.get_directory(album_id) - tracks = [ - self.backend.remote.get_track(i['Id']) - for i in album_data.get('Items', []) - ] - - tracks = sorted(tracks, key=lambda k: k.track_no) - - elif uri.startswith('emby:artist:') and len(parts) == 3: - artist_id = parts[-1] - albums = self.backend.remote.get_directory(artist_id) - tracks = [] - - for album in albums.get('Items', []): - album_data = self.backend.remote.get_directory(album['Id']) - tracklist = [ - self.backend.remote.get_track(i['Id']) - for i in album_data.get('Items', []) - ] - - tracks.extend(sorted(tracklist, key=lambda k: k.track_no)) - - else: - logger.info('Unknown Emby lookup URI: {}'.format(uri)) - tracks = [] - - return [track for track in tracks if track] - - else: - return {uri: self.lookup(uri=uri) for uri in uris} - - def search(self, query=None, uris=None, exact=False): - return self.backend.remote.search(query) - - -class cache(object): - - def __init__(self, ctl=8, ttl=3600): - self.cache = {} - self.ctl = ctl - self.ttl = ttl - self._call_count = 1 - - def __call__(self, func): - def _memoized(*args): - self.func = func - now = time.time() - try: - value, last_update = self.cache[args] - age = now - last_update - if self._call_count >= self.ctl or age > self.ttl: - self._call_count = 1 - raise AttributeError - - self._call_count += 1 - return value - - except (KeyError, AttributeError): - value = self.func(*args) - self.cache[args] = (value, now) - return value - - except TypeError: - return self.func(*args) - - return _memoized - - -class EmbyHandler(object): - def __init__(self, config): - self.hostname = config['emby']['hostname'] - self.port = config['emby']['port'] - self.username = config['emby']['username'] - self.password = config['emby']['password'] - self.proxy = config['proxy'] - - # create authentication headers - self.auth_data = self._password_data() - self.user_id = self._get_user()[0]['Id'] - self.headers = self._create_headers() - self.token = self._get_token() - - self.headers = self._create_headers(token=self.token) - - def _get_user(self): - """Return user dict from server or None if there is no user. - """ - url = self.api_url('/Users/Public') - r = requests.get(url) - user = [i for i in r.json() if i['Name'] == self.username] - - if user: - return user - else: - raise Exception('No Emby user {} found'.format(self.username)) - - def _get_token(self): - """Return token for a user. - """ - url = self.api_url('/Users/AuthenticateByName') - r = requests.post(url, headers=self.headers, data=self.auth_data) - - return r.json().get('AccessToken') - - def _password_data(self): - """Returns a dict with username and its encoded password. - """ - return { - 'username': self.username, - 'password': hashlib.sha1( - self.password.encode('utf-8')).hexdigest(), - 'passwordMd5': hashlib.md5( - self.password.encode('utf-8')).hexdigest() - } - - def _create_headers(self, token=None): - """Return header dict that is needed to talk to the Emby API. - """ - headers = {} - - authorization = ( - 'MediaBrowser UserId="{user_id}", ' - 'Client="other", ' - 'Device="mopidy", ' - 'DeviceId="mopidy", ' - 'Version="0.0.0"' - ).format(user_id=self.user_id) - - headers['x-emby-authorization'] = authorization - - if token: - headers['x-mediabrowser-token'] = self.token - - return headers - - def _get_session(self): - proxy = httpclient.format_proxy(self.proxy) - full_user_agent = httpclient.format_user_agent( - '/'.join( - (mopidy_emby.Extension.dist_name, mopidy_emby.__version__) - ) - ) - - session = requests.Session() - session.proxies.update({'http': proxy, 'https': proxy}) - session.headers.update({'user-agent': full_user_agent}) - - return session - - def r_get(self, url): - counter = 0 - session = self._get_session() - session.headers.update(self.headers) - while counter <= 5: - try: - r = session.get(url) - return r.json() - except Exception as e: - logger.info( - 'Emby connection on try {} with problem: {}'.format( - counter, e - ) - ) - counter += 1 - - # if everything goes wrong return a empty dict - return {} - - def api_url(self, endpoint): - """Returns a joined url. - - Takes host, port and endpoint and generates a valid emby API url. - """ - # check if http or https is defined as host and create hostname - hostname_list = [self.hostname] - if self.hostname.startswith('http://') or \ - self.hostname.startswith('https://'): - hostname = ''.join(hostname_list) - else: - hostname_list.insert(0, 'http://') - hostname = ''.join(hostname_list) - - joined = urljoin( - '{hostname}:{port}'.format( - hostname=hostname, - port=self.port - ), - endpoint - ) - - scheme, netloc, path, query_string, fragment = urlsplit(joined) - query_params = parse_qs(query_string) - - query_params['format'] = ['json'] - new_query_string = urlencode(query_params, doseq=True) - - return urlunsplit((scheme, netloc, path, new_query_string, fragment)) - - def get_music_root(self): - url = self.api_url( - '/Users/{}/Views'.format(self.user_id) - ) - - data = self.r_get(url) - id = [i['Id'] for i in data['Items'] if i['Name'] == 'Music'] - - if id: - logging.debug( - 'Emby: Found music root dir with ID: {}'.format(id[0]) - ) - return id[0] - - else: - logging.debug( - 'Emby: All directories found: {}'.format( - [i['Name'] for i in data['Items']] - ) - ) - raise Exception('Emby: Cant find music root directory') - - def get_artists(self): - music_root = self.get_music_root() - artists = sorted( - self.get_directory(music_root)['Items'], - key=lambda k: k['Name'] - ) - - return [ - models.Ref.artist( - uri='emby:artist:{}'.format(i['Id']), - name=i['Name'] - ) - for i in artists - if i - ] - - def get_albums(self, artist_id): - albums = sorted( - self.get_directory(artist_id)['Items'], - key=lambda k: k['Name'] - ) - return [ - models.Ref.album( - uri='emby:album:{}'.format(i['Id']), - name=i['Name'] - ) - for i in albums - if i - ] - - def get_tracks(self, album_id): - tracks = sorted( - self.get_directory(album_id)['Items'], - key=lambda k: k['IndexNumber'] - ) - - return [ - models.Ref.track( - uri='emby:track:{}'.format( - i['Id'] - ), - name=i['Name'] - ) - for i in tracks - if i - ] - - @cache() - def get_directory(self, id): - """Get directory from Emby API. - - :param id: Directory ID - :type id: int - :returns Directory - :rtype: dict - """ - return self.r_get( - self.api_url( - '/Users/{}/Items?ParentId={}&SortOrder=Ascending'.format( - self.user_id, - id - ) - ) - ) - - @cache() - def get_item(self, id): - """Get item from Emby API. - - :param id: Item ID - :type id: int - :returns: Item - :rtype: dict - """ - data = self.r_get( - self.api_url( - '/Users/{}/Items/{}'.format(self.user_id, id) - ) - ) - - logger.debug('Emby item: {}'.format(data)) - - return data - - def create_track(self, track): - """Create track from Emby API track dict. - - :param track: Track from Emby API - :type track: dict - :returns: Track - :rtype: mopidy.models.Track - """ - # TODO: add more metadata - return models.Track( - uri='emby:track:{}'.format( - track['Id'] - ), - name=track.get('Name'), - track_no=track.get('IndexNumber'), - genre=track.get('Genre'), - artists=self.create_artists(track), - album=self.create_album(track), - length=track['RunTimeTicks'] / 10000 - ) - - def create_album(self, track): - """Create album object from track. - - :param track: Track - :type track: dict - :returns: Album - :rtype: mopidy.models.Album - """ - return models.Album( - name=track.get('Album'), - artists=self.create_artists(track) - ) - - def create_artists(self, track): - """Create artist object from track. - - :param track: Track - :type track: dict - :returns: List of artists - :rtype: list of mopidy.models.Artist - """ - return [ - models.Artist( - name=artist['Name'] - ) - for artist in track['ArtistItems'] - ] - - @cache() - def get_track(self, track_id): - """Get track. - - :param track_id: ID of a Emby track - :type track_id: int - :returns: track - :rtype: mopidy.models.Track - """ - track = self.get_item(track_id) - - return self.create_track(track) - - def _get_search(self, itemtype, term): - """Gets search data from Emby API. - - :param itemtype: Type to search for - :param term: Search term - :type itemtype: str - :type term: str - :returns: List of result dicts - :rtype: list - """ - if itemtype == 'any': - query = 'Audio,MusicAlbum,MusicArtist' - elif itemtype == 'artist': - query = 'MusicArtist' - elif itemtype == 'album': - query = 'MusicAlbum' - elif itemtype == 'track_name': - query = 'Audio' - else: - raise Exception('Emby search: no itemtype {}'.format()) - - data = self.r_get( - self.api_url( - ('/Search/Hints?SearchTerm={}&' - 'IncludeItemTypes={}').format( - quote(term), - query - ) - ) - ) - - return [i for i in data.get('SearchHints', [])] - - @cache() - def search(self, query): - """Search Emby for a term. - - :param query: Search query - :type query: dict - :returns: Search results - :rtype: mopidy.models.SearchResult - """ - logger.debug('Searching in Emby for {}'.format(query)) - - # something to store the results in - data = [] - tracks = [] - albums = [] - artists = [] - - for itemtype, term in query.items(): - - for item in term: - - data.extend( - self._get_search(itemtype, item) - ) - - # walk through all items and create stuff - for item in data: - - if item['Type'] == 'Audio': - - track_artists = [ - models.Artist( - name=artist - ) - for artist in item['Artists'] - ] - - tracks.append( - models.Track( - uri='emby:track:{}'.format(item['ItemId']), - track_no=item.get('IndexNumber'), - name=item.get('Name'), - artists=track_artists, - album=models.Album( - name=item.get('Album'), - artists=track_artists - ) - ) - ) - - elif item['Type'] == 'MusicAlbum': - album_artists = [ - models.Artist( - name=artist - ) - for artist in item['Artists'] - ] - - albums.append( - models.Album( - uri='emby:album:{}'.format(item['ItemId']), - name=item.get('Name'), - artists=album_artists - ) - ) - - elif item['Type'] == 'MusicArtist': - artists.append( - models.Artist( - uri='emby:artist:{}'.format(item['ItemId']), - name=item.get('Name') - ) - ) - - return models.SearchResult( - uri='emby:search', - tracks=tracks, - artists=artists, - albums=albums - ) diff --git a/mopidy_emby/library.py b/mopidy_emby/library.py new file mode 100644 index 0000000..1fb92c9 --- /dev/null +++ b/mopidy_emby/library.py @@ -0,0 +1,86 @@ +from __future__ import unicode_literals + +import logging + +from mopidy import backend, models + + +logger = logging.getLogger(__name__) + + +class EmbyLibraryProvider(backend.LibraryProvider): + + root_directory = models.Ref.directory(uri='emby:', + name='Emby') + + def browse(self, uri): + # artistlist + if uri == self.root_directory.uri: + logger.debug('Get Emby artist list') + return self.backend.remote.get_artists() + + # split uri + parts = uri.split(':') + + # artists albums + # uri: emby:artist: + if uri.startswith('emby:artist:') and len(parts) == 3: + logger.debug('Get Emby album list') + artist_id = parts[-1] + + return self.backend.remote.get_albums(artist_id) + + # tracklist + # uri: emby:album: + if uri.startswith('emby:album:') and len(parts) == 3: + logger.debug('Get Emby track list') + album_id = parts[-1] + + return self.backend.remote.get_tracks(album_id) + + return [] + + def lookup(self, uri=None, uris=None): + logger.debug('Emby lookup: {}'.format(uri or uris)) + if uri: + parts = uri.split(':') + + if uri.startswith('emby:track:') and len(parts) == 3: + track_id = parts[-1] + tracks = [self.backend.remote.get_track(track_id)] + + elif uri.startswith('emby:album:') and len(parts) == 3: + album_id = parts[-1] + album_data = self.backend.remote.get_directory(album_id) + tracks = [ + self.backend.remote.get_track(i['Id']) + for i in album_data.get('Items', []) + ] + + tracks = sorted(tracks, key=lambda k: k.track_no) + + elif uri.startswith('emby:artist:') and len(parts) == 3: + artist_id = parts[-1] + albums = self.backend.remote.get_directory(artist_id) + tracks = [] + + for album in albums.get('Items', []): + album_data = self.backend.remote.get_directory(album['Id']) + tracklist = [ + self.backend.remote.get_track(i['Id']) + for i in album_data.get('Items', []) + ] + + tracks.extend(sorted(tracklist, key=lambda k: k.track_no)) + + else: + logger.info('Unknown Emby lookup URI: {}'.format(uri)) + tracks = [] + + return [track for track in tracks if track] + + else: + return {uri: self.lookup(uri=uri) for uri in uris} + + def search(self, query=None, uris=None, exact=False): + return self.backend.remote.search(query) diff --git a/mopidy_emby/playback.py b/mopidy_emby/playback.py new file mode 100644 index 0000000..4dabc83 --- /dev/null +++ b/mopidy_emby/playback.py @@ -0,0 +1,26 @@ +from __future__ import unicode_literals + +import logging + +from mopidy import backend + + +logger = logging.getLogger(__name__) + + +class EmbyPlaybackProvider(backend.PlaybackProvider): + + def translate_uri(self, uri): + if uri.startswith('emby:track:') and len(uri.split(':')) == 3: + id = uri.split(':')[-1] + + track_url = self.backend.remote.api_url( + '/Audio/{}/stream.mp3'.format(id) + ) + + logger.debug('Emby track streaming url: {}'.format(track_url)) + + return track_url + + else: + return None diff --git a/mopidy_emby/remote.py b/mopidy_emby/remote.py new file mode 100644 index 0000000..717291e --- /dev/null +++ b/mopidy_emby/remote.py @@ -0,0 +1,464 @@ +from __future__ import unicode_literals + +import hashlib +import logging +import time + +from urllib import urlencode +from urllib2 import quote +from urlparse import parse_qs, urljoin, urlsplit, urlunsplit + +from mopidy import httpclient, models + +import requests + +import mopidy_emby + + +logger = logging.getLogger(__name__) + + +class cache(object): + + def __init__(self, ctl=8, ttl=3600): + self.cache = {} + self.ctl = ctl + self.ttl = ttl + self._call_count = 1 + + def __call__(self, func): + def _memoized(*args): + self.func = func + now = time.time() + try: + value, last_update = self.cache[args] + age = now - last_update + if self._call_count >= self.ctl or age > self.ttl: + self._call_count = 1 + raise AttributeError + + self._call_count += 1 + return value + + except (KeyError, AttributeError): + value = self.func(*args) + self.cache[args] = (value, now) + return value + + except TypeError: + return self.func(*args) + + return _memoized + + +class EmbyHandler(object): + def __init__(self, config): + self.hostname = config['emby']['hostname'] + self.port = config['emby']['port'] + self.username = config['emby']['username'] + self.password = config['emby']['password'] + self.proxy = config['proxy'] + + # create authentication headers + self.auth_data = self._password_data() + self.user_id = self._get_user()[0]['Id'] + self.headers = self._create_headers() + self.token = self._get_token() + + self.headers = self._create_headers(token=self.token) + + def _get_user(self): + """Return user dict from server or None if there is no user. + """ + url = self.api_url('/Users/Public') + r = requests.get(url) + user = [i for i in r.json() if i['Name'] == self.username] + + if user: + return user + else: + raise Exception('No Emby user {} found'.format(self.username)) + + def _get_token(self): + """Return token for a user. + """ + url = self.api_url('/Users/AuthenticateByName') + r = requests.post(url, headers=self.headers, data=self.auth_data) + + return r.json().get('AccessToken') + + def _password_data(self): + """Returns a dict with username and its encoded password. + """ + return { + 'username': self.username, + 'password': hashlib.sha1( + self.password.encode('utf-8')).hexdigest(), + 'passwordMd5': hashlib.md5( + self.password.encode('utf-8')).hexdigest() + } + + def _create_headers(self, token=None): + """Return header dict that is needed to talk to the Emby API. + """ + headers = {} + + authorization = ( + 'MediaBrowser UserId="{user_id}", ' + 'Client="other", ' + 'Device="mopidy", ' + 'DeviceId="mopidy", ' + 'Version="0.0.0"' + ).format(user_id=self.user_id) + + headers['x-emby-authorization'] = authorization + + if token: + headers['x-mediabrowser-token'] = self.token + + return headers + + def _get_session(self): + proxy = httpclient.format_proxy(self.proxy) + full_user_agent = httpclient.format_user_agent( + '/'.join( + (mopidy_emby.Extension.dist_name, mopidy_emby.__version__) + ) + ) + + session = requests.Session() + session.proxies.update({'http': proxy, 'https': proxy}) + session.headers.update({'user-agent': full_user_agent}) + + return session + + def r_get(self, url): + counter = 0 + session = self._get_session() + session.headers.update(self.headers) + while counter <= 5: + try: + r = session.get(url) + return r.json() + except Exception as e: + logger.info( + 'Emby connection on try {} with problem: {}'.format( + counter, e + ) + ) + counter += 1 + + # if everything goes wrong return a empty dict + return {} + + def api_url(self, endpoint): + """Returns a joined url. + + Takes host, port and endpoint and generates a valid emby API url. + """ + # check if http or https is defined as host and create hostname + hostname_list = [self.hostname] + if self.hostname.startswith('http://') or \ + self.hostname.startswith('https://'): + hostname = ''.join(hostname_list) + else: + hostname_list.insert(0, 'http://') + hostname = ''.join(hostname_list) + + joined = urljoin( + '{hostname}:{port}'.format( + hostname=hostname, + port=self.port + ), + endpoint + ) + + scheme, netloc, path, query_string, fragment = urlsplit(joined) + query_params = parse_qs(query_string) + + query_params['format'] = ['json'] + new_query_string = urlencode(query_params, doseq=True) + + return urlunsplit((scheme, netloc, path, new_query_string, fragment)) + + def get_music_root(self): + url = self.api_url( + '/Users/{}/Views'.format(self.user_id) + ) + + data = self.r_get(url) + id = [i['Id'] for i in data['Items'] if i['Name'] == 'Music'] + + if id: + logging.debug( + 'Emby: Found music root dir with ID: {}'.format(id[0]) + ) + return id[0] + + else: + logging.debug( + 'Emby: All directories found: {}'.format( + [i['Name'] for i in data['Items']] + ) + ) + raise Exception('Emby: Cant find music root directory') + + def get_artists(self): + music_root = self.get_music_root() + artists = sorted( + self.get_directory(music_root)['Items'], + key=lambda k: k['Name'] + ) + + return [ + models.Ref.artist( + uri='emby:artist:{}'.format(i['Id']), + name=i['Name'] + ) + for i in artists + if i + ] + + def get_albums(self, artist_id): + albums = sorted( + self.get_directory(artist_id)['Items'], + key=lambda k: k['Name'] + ) + return [ + models.Ref.album( + uri='emby:album:{}'.format(i['Id']), + name=i['Name'] + ) + for i in albums + if i + ] + + def get_tracks(self, album_id): + tracks = sorted( + self.get_directory(album_id)['Items'], + key=lambda k: k['IndexNumber'] + ) + + return [ + models.Ref.track( + uri='emby:track:{}'.format( + i['Id'] + ), + name=i['Name'] + ) + for i in tracks + if i + ] + + @cache() + def get_directory(self, id): + """Get directory from Emby API. + + :param id: Directory ID + :type id: int + :returns Directory + :rtype: dict + """ + return self.r_get( + self.api_url( + '/Users/{}/Items?ParentId={}&SortOrder=Ascending'.format( + self.user_id, + id + ) + ) + ) + + @cache() + def get_item(self, id): + """Get item from Emby API. + + :param id: Item ID + :type id: int + :returns: Item + :rtype: dict + """ + data = self.r_get( + self.api_url( + '/Users/{}/Items/{}'.format(self.user_id, id) + ) + ) + + logger.debug('Emby item: {}'.format(data)) + + return data + + def create_track(self, track): + """Create track from Emby API track dict. + + :param track: Track from Emby API + :type track: dict + :returns: Track + :rtype: mopidy.models.Track + """ + # TODO: add more metadata + return models.Track( + uri='emby:track:{}'.format( + track['Id'] + ), + name=track.get('Name'), + track_no=track.get('IndexNumber'), + genre=track.get('Genre'), + artists=self.create_artists(track), + album=self.create_album(track), + length=track['RunTimeTicks'] / 10000 + ) + + def create_album(self, track): + """Create album object from track. + + :param track: Track + :type track: dict + :returns: Album + :rtype: mopidy.models.Album + """ + return models.Album( + name=track.get('Album'), + artists=self.create_artists(track) + ) + + def create_artists(self, track): + """Create artist object from track. + + :param track: Track + :type track: dict + :returns: List of artists + :rtype: list of mopidy.models.Artist + """ + return [ + models.Artist( + name=artist['Name'] + ) + for artist in track['ArtistItems'] + ] + + @cache() + def get_track(self, track_id): + """Get track. + + :param track_id: ID of a Emby track + :type track_id: int + :returns: track + :rtype: mopidy.models.Track + """ + track = self.get_item(track_id) + + return self.create_track(track) + + def _get_search(self, itemtype, term): + """Gets search data from Emby API. + + :param itemtype: Type to search for + :param term: Search term + :type itemtype: str + :type term: str + :returns: List of result dicts + :rtype: list + """ + if itemtype == 'any': + query = 'Audio,MusicAlbum,MusicArtist' + elif itemtype == 'artist': + query = 'MusicArtist' + elif itemtype == 'album': + query = 'MusicAlbum' + elif itemtype == 'track_name': + query = 'Audio' + else: + raise Exception('Emby search: no itemtype {}'.format()) + + data = self.r_get( + self.api_url( + ('/Search/Hints?SearchTerm={}&' + 'IncludeItemTypes={}').format( + quote(term), + query + ) + ) + ) + + return [i for i in data.get('SearchHints', [])] + + @cache() + def search(self, query): + """Search Emby for a term. + + :param query: Search query + :type query: dict + :returns: Search results + :rtype: mopidy.models.SearchResult + """ + logger.debug('Searching in Emby for {}'.format(query)) + + # something to store the results in + data = [] + tracks = [] + albums = [] + artists = [] + + for itemtype, term in query.items(): + + for item in term: + + data.extend( + self._get_search(itemtype, item) + ) + + # walk through all items and create stuff + for item in data: + + if item['Type'] == 'Audio': + + track_artists = [ + models.Artist( + name=artist + ) + for artist in item['Artists'] + ] + + tracks.append( + models.Track( + uri='emby:track:{}'.format(item['ItemId']), + track_no=item.get('IndexNumber'), + name=item.get('Name'), + artists=track_artists, + album=models.Album( + name=item.get('Album'), + artists=track_artists + ) + ) + ) + + elif item['Type'] == 'MusicAlbum': + album_artists = [ + models.Artist( + name=artist + ) + for artist in item['Artists'] + ] + + albums.append( + models.Album( + uri='emby:album:{}'.format(item['ItemId']), + name=item.get('Name'), + artists=album_artists + ) + ) + + elif item['Type'] == 'MusicArtist': + artists.append( + models.Artist( + uri='emby:artist:{}'.format(item['ItemId']), + name=item.get('Name') + ) + ) + + return models.SearchResult( + uri='emby:search', + tracks=tracks, + artists=artists, + albums=albums + ) diff --git a/tests/conftest.py b/tests/conftest.py index 192deb2..b7be155 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -24,14 +24,14 @@ def config(): @pytest.fixture def emby_client(config, mocker): - mocker.patch('mopidy_emby.backend.cache') - mocker.patch('mopidy_emby.backend.EmbyHandler._get_token') - mocker.patch('mopidy_emby.backend.EmbyHandler._create_headers') - mocker.patch('mopidy_emby.backend.EmbyHandler._get_user', + mocker.patch('mopidy_emby.remote.cache') + mocker.patch('mopidy_emby.remote.EmbyHandler._get_token') + mocker.patch('mopidy_emby.remote.EmbyHandler._create_headers') + mocker.patch('mopidy_emby.remote.EmbyHandler._get_user', return_value=[{'Id': 'mock'}]) - mocker.patch('mopidy_emby.backend.EmbyHandler._password_data') + mocker.patch('mopidy_emby.remote.EmbyHandler._password_data') - return mopidy_emby.backend.EmbyHandler(config) + return mopidy_emby.remote.EmbyHandler(config) @pytest.fixture diff --git a/tests/test_extension.py b/tests/test_extension.py index 679ad1f..c43d2e1 100644 --- a/tests/test_extension.py +++ b/tests/test_extension.py @@ -1,14 +1,6 @@ from __future__ import unicode_literals -import json - -import mock - -from mopidy.models import Album, Artist, Ref, SearchResult, Track - -import pytest - -from mopidy_emby import Extension, backend +from mopidy_emby import Extension def test_get_default_config(): @@ -29,467 +21,3 @@ def test_get_config_schema(): assert 'password' in schema assert 'hostname' in schema assert 'port' in schema - - -@pytest.mark.parametrize('hostname,url,expected', [ - ('https://foo.bar', '/Foo', 'https://foo.bar:443/Foo?format=json'), - ('foo.bar', '/Foo', 'http://foo.bar:443/Foo?format=json'), -]) -@mock.patch('mopidy_emby.backend.EmbyHandler._get_token') -@mock.patch('mopidy_emby.backend.EmbyHandler._create_headers') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_user') -@mock.patch('mopidy_emby.backend.EmbyHandler._password_data') -def test_api_url(password_data_mock, get_user_mock, create_header_mock, - get_token_mock, config, hostname, url, expected): - get_user_mock.return_value = [{'Id': 'foo'}] - - config['emby']['hostname'] = hostname - emby = backend.EmbyHandler(config) - - assert emby.api_url(url) == expected - - -@pytest.mark.parametrize('data,expected', [ - ('tests/data/get_music_root0.json', 'eb169f4ba53fc560f549cb0f2a47d577') -]) -@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') -def test_get_music_root(r_get_mock, data, expected, emby_client): - - with open(data, 'r') as f: - r_get_mock.return_value = json.load(f) - - assert emby_client.get_music_root() == expected - - -@pytest.mark.parametrize('data,expected', [ - ( - 'tests/data/get_music_root1.json', - 'Emby: Cant find music root directory' - ) -]) -@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') -def test_get_music_root_cant_find(r_get_mock, data, expected, emby_client): - - with open(data, 'r') as f: - r_get_mock.return_value = json.load(f) - - with pytest.raises(Exception) as execinfo: - print emby_client.get_music_root() - - assert expected in str(execinfo.value) - - -@mock.patch('mopidy_emby.backend.EmbyHandler.get_music_root') -@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') -def test_get_artists(r_get_mock, get_music_root_mock, emby_client): - expected = [ - Ref(name=u'Chairlift', - type='artist', - uri='emby:artist:e0361aff955c30f5a6dcc6fcf0c9d1cf'), - Ref(name=u'Hans Zimmer', - type='artist', - uri='emby:artist:36de3368f493ebca94a55a411cc87862'), - Ref(name=u'The Menzingers', - type='artist', - uri='emby:artist:21c8f78763231ece7defd07b5f3f56be') - ] - - with open('tests/data/get_artists0.json', 'r') as f: - r_get_mock.return_value = json.load(f) - - assert emby_client.get_artists() == expected - - -@mock.patch('mopidy_emby.backend.EmbyHandler.get_music_root') -@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') -def test_get_albums(r_get_mock, get_music_root_mock, emby_client): - expected = [ - Ref(name=u'American Football', - type='album', - uri='emby:album:6e4a2da7df0502650bb9b091312c3dbf'), - Ref(name=u'American Football', - type='album', - uri='emby:album:ca498ea939b28593744c051d9f5e74ed'), - Ref(name=u'American Football', - type='album', - uri='emby:album:0db6395ab76b6edbaba3a51ef23d0aa3') - ] - - with open('tests/data/get_albums0.json', 'r') as f: - r_get_mock.return_value = json.load(f) - - assert emby_client.get_albums(0) == expected - - -@mock.patch('mopidy_emby.backend.EmbyHandler.get_music_root') -@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') -def test_get_tracks(r_get_mock, get_music_root_mock, emby_client): - expected = [ - Ref(name=u'The One With the Tambourine', - type='track', - uri='emby:track:eb6c305bdb1e40d3b46909473c22d906'), - Ref(name=u'Letters and Packages', - type='track', - uri='emby:track:7739d3830818c7aacf6c346172384914'), - Ref(name=u'Five Silent Miles', - type='track', - uri='emby:track:f84df9f70e592a3abda82b1d78026608') - ] - - with open('tests/data/get_tracks0.json', 'r') as f: - r_get_mock.return_value = json.load(f) - - assert emby_client.get_tracks(0) == expected - - -@pytest.mark.parametrize('data,expected', [ - ( - 'tests/data/track0.json', - Track( - album=Album(artists=[Artist(name=u'Chairlift')], name=u'Moth'), - artists=[Artist(name=u'Chairlift')], - length=295915, - name=u'Ottawa to Osaka', - track_no=6, - uri='emby:track:18e5a9871e6a4a2294d5af998457ca16' - ) - ), - ( - 'tests/data/track1.json', - Track( - album=Album(artists=[Artist(name=u'Chairlift')], name=u'Moth'), - artists=[Artist(name=u'Chairlift')], - length=269035, - name=u'Crying in Public', - track_no=5, - uri='emby:track:37f57f0b370274af96de06895a78c2c3' - ) - ), - ( - 'tests/data/track2.json', - Track( - album=Album(artists=[Artist(name=u'Chairlift')], name=u'Moth'), - artists=[Artist(name=u'Chairlift')], - length=283115, - name=u'Polymorphing', - track_no=2, - uri='emby:track:3315cccffe37ab47d50d1dbeefd3537b' - ) - ), -]) -def test_create_track(data, expected, emby_client): - with open(data, 'r') as f: - track = json.load(f) - - assert emby_client.create_track(track) == expected - - -@pytest.mark.parametrize('data,expected', [ - ( - 'tests/data/track0.json', - Album(artists=[Artist(name=u'Chairlift')], name=u'Moth') - ), - ( - 'tests/data/track1.json', - Album(artists=[Artist(name=u'Chairlift')], name=u'Moth') - ), - ( - 'tests/data/track2.json', - Album(artists=[Artist(name=u'Chairlift')], name=u'Moth') - ), -]) -def test_create_album(data, expected, emby_client): - with open(data, 'r') as f: - track = json.load(f) - - assert emby_client.create_album(track) == expected - - -@pytest.mark.parametrize('data,expected', [ - ( - 'tests/data/track0.json', - [Artist(name=u'Chairlift')] - ), - ( - 'tests/data/track1.json', - [Artist(name=u'Chairlift')] - ), - ( - 'tests/data/track2.json', - [Artist(name=u'Chairlift')] - ), -]) -def test_create_artists(data, expected, emby_client): - with open(data, 'r') as f: - track = json.load(f) - - assert emby_client.create_artists(track) == expected - - -@pytest.mark.parametrize('uri,expected', [ - ('emby:', ['Artistlist']), - ('emby:artist:123', ['Albumlist']), - ('emby:album:123', ['Tracklist']), -]) -def test_browse(uri, expected, libraryprovider): - assert libraryprovider.browse(uri) == expected - - -@pytest.mark.parametrize('uri,expected', [ - ('emby:track:123', [ - Track( - album=Album( - artists=[ - Artist(name='American Football') - ], - name='American Football'), - artists=[Artist(name='American Football')], - length=241162, - name='The One With the Tambourine', - track_no=1, - uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' - ) - ]), - ('emby:album:123', [ - Track( - album=Album( - artists=[ - Artist(name='American Football') - ], - name='American Football'), - artists=[Artist(name='American Football')], - length=241162, - name='The One With the Tambourine', - track_no=1, - uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' - ) - ]), - ('emby:artist:123', [ - Track( - album=Album( - artists=[ - Artist(name='American Football') - ], - name='American Football'), - artists=[Artist(name='American Football')], - length=241162, - name='The One With the Tambourine', - track_no=1, - uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' - ) - ]), - ('emby:track', []) -]) -def test_lookup_uri(uri, expected, libraryprovider): - assert libraryprovider.lookup(uri=uri) == expected - - -@pytest.mark.parametrize('uri,expected', [ - (['emby:track:123'], {'emby:track:123': [ - Track( - album=Album( - artists=[ - Artist(name='American Football') - ], - name='American Football'), - artists=[Artist(name='American Football')], - length=241162, - name='The One With the Tambourine', - track_no=1, - uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' - ) - ]}), - (['emby:track'], {u'emby:track': []}) -]) -def test_lookup_uris(uri, expected, libraryprovider): - assert libraryprovider.lookup(uris=uri) == expected - - -@pytest.mark.parametrize('uri,expected', [ - ( - 'emby:track:123', - 'https://foo.bar:443/Audio/123/stream.mp3?format=json' - ), - ( - 'emby:foobar', - None - ) -]) -def test_translate_uri(playbackprovider, uri, expected): - assert playbackprovider.translate_uri(uri) == expected - - -@pytest.mark.parametrize('data,user_id', [ - ('tests/data/get_user0.json', '2ec276a2642e54a19b612b9418a8bd3b') -]) -@mock.patch('mopidy_emby.backend.requests.get') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_token') -@mock.patch('mopidy_emby.backend.EmbyHandler._create_headers') -@mock.patch('mopidy_emby.backend.EmbyHandler._password_data') -def test_get_user(password_mock, create_headers_mock, get_tocken_mock, - get_mock, data, user_id, config): - - mock_response = mock.Mock() - with open(data, 'r') as f: - mock_response.json.return_value = json.load(f) - - get_mock.return_value = mock_response - - emby = backend.EmbyHandler(config) - - assert emby.user_id == user_id - - -@mock.patch('mopidy_emby.backend.requests.get') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_token') -@mock.patch('mopidy_emby.backend.EmbyHandler._create_headers') -@mock.patch('mopidy_emby.backend.EmbyHandler._password_data') -def test_get_user_exception(password_mock, create_headers_mock, - get_tocken_mock, get_mock, config): - - mock_response = mock.Mock() - with open('tests/data/get_user1.json', 'r') as f: - mock_response.json.return_value = json.load(f) - - get_mock.return_value = mock_response - - with pytest.raises(Exception) as execinfo: - backend.EmbyHandler(config) - - assert 'No Emby user embyuser found' in str(execinfo.value) - - -@pytest.mark.parametrize('data,token', [ - ('tests/data/get_token0.json', 'f0d6b372b40b47299ed01b9b2d40489b'), - ('tests/data/get_token1.json', None), -]) -@mock.patch('mopidy_emby.backend.requests.post') -@mock.patch('mopidy_emby.backend.EmbyHandler._create_headers') -@mock.patch('mopidy_emby.backend.EmbyHandler._password_data') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_user') -def test_get_token(get_user_mock, password_data_mock, - create_headers_mock, post_mock, data, - token, config): - - mock_response = mock.Mock() - with open(data, 'r') as f: - mock_response.json.return_value = json.load(f) - - post_mock.return_value = mock_response - - emby = backend.EmbyHandler(config) - - assert emby.token == token - - -@mock.patch('mopidy_emby.backend.requests') -@mock.patch('mopidy_emby.backend.EmbyHandler._create_headers') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_user') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_token') -def test_password_data(get_token_mock, get_user_mock, create_headers_mock, - requests_mock, config): - - emby = backend.EmbyHandler(config) - - assert emby._password_data() == { - 'username': 'embyuser', - 'password': '444b73bcd9dc4331104c5ef960ee240066f8a3e5', - 'passwordMd5': '1d549a7b47c46b7b0a90651360c5574c' - } - - -@pytest.mark.parametrize('token,headers', [ - ( - None, - { - 'x-emby-authorization': ('MediaBrowser UserId="123", ' - 'Client="other", Device="mopidy", ' - 'DeviceId="mopidy", Version="0.0.0"') - } - ), - ( - 'f0d6b372b40b47299ed01b9b2d40489b', - { - 'x-emby-authorization': ('MediaBrowser UserId="123", ' - 'Client="other", Device="mopidy", ' - 'DeviceId="mopidy", Version="0.0.0"'), - 'x-mediabrowser-token': 'f0d6b372b40b47299ed01b9b2d40489b' - } - ) -]) -@mock.patch('mopidy_emby.backend.requests') -@mock.patch('mopidy_emby.backend.EmbyHandler._password_data') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_user') -@mock.patch('mopidy_emby.backend.EmbyHandler._get_token') -def test_create_headers(get_token_mock, get_user_mock, password_data_mock, - requests_mock, token, headers, config): - - get_user_mock.return_value = [{'Id': 123}] - get_token_mock.return_value = token - - emby = backend.EmbyHandler(config) - - assert emby.headers == headers - - -@pytest.mark.parametrize('query,data,expected', [ - ( - {'track_name': ['viva hate']}, - 'tests/data/search_audio0.json', - SearchResult( - tracks=[ - Track( - album=Album( - artists=[Artist(name=u'Rainer Maria')], - name=u'Past Worn Searching' - ), - artists=[Artist(name=u'Rainer Maria')], - name='Viva Anger, Viva Hate', - track_no=3, - uri='emby:track:b5d600663238be5b41da4d8429db85f0' - ) - ], - uri='emby:search' - ) - ), - ( - {'album': ['viva hate']}, - 'tests/data/search_album0.json', - SearchResult( - albums=[ - Album( - artists=[Artist(name=u'Morrissey')], - name=u'Viva Hate', - uri='emby:album:4bf594cb601ec46a0295729c4d0f7f80') - ], - uri='emby:search' - ) - ), - ( - {'artist': ['morrissey']}, - 'tests/data/search_artist0.json', - SearchResult( - artists=[ - Artist( - name=u'Morrissey', - uri='emby:artist:0b74a057d86092f48698be681737c4ed' - ), - Artist( - name=u'Morrissey & Siouxsie Sioux', - uri='emby:artist:32bbd3db105255b24a83d0d955179dc4' - ), - Artist( - name=u'Morrissey & Siouxsie Sioux', - uri='emby:artist:eb69a3f2db13691d24c6a9794926ddb8' - ) - ], - uri='emby:search' - ) - ) -]) -@mock.patch('mopidy_emby.backend.EmbyHandler._get_search') -def test_search(get_search_mock, query, data, expected, emby_client): - with open(data, 'r') as f: - get_search_mock.return_value = json.load(f)['SearchHints'] - - assert emby_client.search(query) == expected diff --git a/tests/test_library.py b/tests/test_library.py new file mode 100644 index 0000000..440730e --- /dev/null +++ b/tests/test_library.py @@ -0,0 +1,84 @@ +from __future__ import unicode_literals + +from mopidy.models import Album, Artist, Track + +import pytest + + +@pytest.mark.parametrize('uri,expected', [ + ('emby:', ['Artistlist']), + ('emby:artist:123', ['Albumlist']), + ('emby:album:123', ['Tracklist']), +]) +def test_browse(uri, expected, libraryprovider): + assert libraryprovider.browse(uri) == expected + + +@pytest.mark.parametrize('uri,expected', [ + ('emby:track:123', [ + Track( + album=Album( + artists=[ + Artist(name='American Football') + ], + name='American Football'), + artists=[Artist(name='American Football')], + length=241162, + name='The One With the Tambourine', + track_no=1, + uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' + ) + ]), + ('emby:album:123', [ + Track( + album=Album( + artists=[ + Artist(name='American Football') + ], + name='American Football'), + artists=[Artist(name='American Football')], + length=241162, + name='The One With the Tambourine', + track_no=1, + uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' + ) + ]), + ('emby:artist:123', [ + Track( + album=Album( + artists=[ + Artist(name='American Football') + ], + name='American Football'), + artists=[Artist(name='American Football')], + length=241162, + name='The One With the Tambourine', + track_no=1, + uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' + ) + ]), + ('emby:track', []) +]) +def test_lookup_uri(uri, expected, libraryprovider): + assert libraryprovider.lookup(uri=uri) == expected + + +@pytest.mark.parametrize('uri,expected', [ + (['emby:track:123'], {'emby:track:123': [ + Track( + album=Album( + artists=[ + Artist(name='American Football') + ], + name='American Football'), + artists=[Artist(name='American Football')], + length=241162, + name='The One With the Tambourine', + track_no=1, + uri='emby:track:eb6c305bdb1e40d3b46909473c22d906' + ) + ]}), + (['emby:track'], {u'emby:track': []}) +]) +def test_lookup_uris(uri, expected, libraryprovider): + assert libraryprovider.lookup(uris=uri) == expected diff --git a/tests/test_playback.py b/tests/test_playback.py new file mode 100644 index 0000000..74bb520 --- /dev/null +++ b/tests/test_playback.py @@ -0,0 +1,17 @@ +from __future__ import unicode_literals + +import pytest + + +@pytest.mark.parametrize('uri,expected', [ + ( + 'emby:track:123', + 'https://foo.bar:443/Audio/123/stream.mp3?format=json' + ), + ( + 'emby:foobar', + None + ) +]) +def test_translate_uri(playbackprovider, uri, expected): + assert playbackprovider.translate_uri(uri) == expected diff --git a/tests/test_remote.py b/tests/test_remote.py new file mode 100644 index 0000000..8d4de90 --- /dev/null +++ b/tests/test_remote.py @@ -0,0 +1,382 @@ +from __future__ import unicode_literals + +import json + +import mock + +from mopidy.models import Album, Artist, Ref, SearchResult, Track + +import pytest + +from mopidy_emby import backend + + +@pytest.mark.parametrize('hostname,url,expected', [ + ('https://foo.bar', '/Foo', 'https://foo.bar:443/Foo?format=json'), + ('foo.bar', '/Foo', 'http://foo.bar:443/Foo?format=json'), +]) +@mock.patch('mopidy_emby.backend.EmbyHandler._get_token') +@mock.patch('mopidy_emby.backend.EmbyHandler._create_headers') +@mock.patch('mopidy_emby.backend.EmbyHandler._get_user') +@mock.patch('mopidy_emby.backend.EmbyHandler._password_data') +def test_api_url(password_data_mock, get_user_mock, create_header_mock, + get_token_mock, config, hostname, url, expected): + get_user_mock.return_value = [{'Id': 'foo'}] + + config['emby']['hostname'] = hostname + emby = backend.EmbyHandler(config) + + assert emby.api_url(url) == expected + + +@pytest.mark.parametrize('data,expected', [ + ('tests/data/get_music_root0.json', 'eb169f4ba53fc560f549cb0f2a47d577') +]) +@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') +def test_get_music_root(r_get_mock, data, expected, emby_client): + + with open(data, 'r') as f: + r_get_mock.return_value = json.load(f) + + assert emby_client.get_music_root() == expected + + +@pytest.mark.parametrize('data,expected', [ + ( + 'tests/data/get_music_root1.json', + 'Emby: Cant find music root directory' + ) +]) +@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') +def test_get_music_root_cant_find(r_get_mock, data, expected, emby_client): + + with open(data, 'r') as f: + r_get_mock.return_value = json.load(f) + + with pytest.raises(Exception) as execinfo: + print emby_client.get_music_root() + + assert expected in str(execinfo.value) + + +@mock.patch('mopidy_emby.backend.EmbyHandler.get_music_root') +@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') +def test_get_artists(r_get_mock, get_music_root_mock, emby_client): + expected = [ + Ref(name=u'Chairlift', + type='artist', + uri='emby:artist:e0361aff955c30f5a6dcc6fcf0c9d1cf'), + Ref(name=u'Hans Zimmer', + type='artist', + uri='emby:artist:36de3368f493ebca94a55a411cc87862'), + Ref(name=u'The Menzingers', + type='artist', + uri='emby:artist:21c8f78763231ece7defd07b5f3f56be') + ] + + with open('tests/data/get_artists0.json', 'r') as f: + r_get_mock.return_value = json.load(f) + + assert emby_client.get_artists() == expected + + +@mock.patch('mopidy_emby.backend.EmbyHandler.get_music_root') +@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') +def test_get_albums(r_get_mock, get_music_root_mock, emby_client): + expected = [ + Ref(name=u'American Football', + type='album', + uri='emby:album:6e4a2da7df0502650bb9b091312c3dbf'), + Ref(name=u'American Football', + type='album', + uri='emby:album:ca498ea939b28593744c051d9f5e74ed'), + Ref(name=u'American Football', + type='album', + uri='emby:album:0db6395ab76b6edbaba3a51ef23d0aa3') + ] + + with open('tests/data/get_albums0.json', 'r') as f: + r_get_mock.return_value = json.load(f) + + assert emby_client.get_albums(0) == expected + + +@mock.patch('mopidy_emby.backend.EmbyHandler.get_music_root') +@mock.patch('mopidy_emby.backend.EmbyHandler.r_get') +def test_get_tracks(r_get_mock, get_music_root_mock, emby_client): + expected = [ + Ref(name=u'The One With the Tambourine', + type='track', + uri='emby:track:eb6c305bdb1e40d3b46909473c22d906'), + Ref(name=u'Letters and Packages', + type='track', + uri='emby:track:7739d3830818c7aacf6c346172384914'), + Ref(name=u'Five Silent Miles', + type='track', + uri='emby:track:f84df9f70e592a3abda82b1d78026608') + ] + + with open('tests/data/get_tracks0.json', 'r') as f: + r_get_mock.return_value = json.load(f) + + assert emby_client.get_tracks(0) == expected + + +@pytest.mark.parametrize('data,expected', [ + ( + 'tests/data/track0.json', + Track( + album=Album(artists=[Artist(name=u'Chairlift')], name=u'Moth'), + artists=[Artist(name=u'Chairlift')], + length=295915, + name=u'Ottawa to Osaka', + track_no=6, + uri='emby:track:18e5a9871e6a4a2294d5af998457ca16' + ) + ), + ( + 'tests/data/track1.json', + Track( + album=Album(artists=[Artist(name=u'Chairlift')], name=u'Moth'), + artists=[Artist(name=u'Chairlift')], + length=269035, + name=u'Crying in Public', + track_no=5, + uri='emby:track:37f57f0b370274af96de06895a78c2c3' + ) + ), + ( + 'tests/data/track2.json', + Track( + album=Album(artists=[Artist(name=u'Chairlift')], name=u'Moth'), + artists=[Artist(name=u'Chairlift')], + length=283115, + name=u'Polymorphing', + track_no=2, + uri='emby:track:3315cccffe37ab47d50d1dbeefd3537b' + ) + ), +]) +def test_create_track(data, expected, emby_client): + with open(data, 'r') as f: + track = json.load(f) + + assert emby_client.create_track(track) == expected + + +@pytest.mark.parametrize('data,expected', [ + ( + 'tests/data/track0.json', + Album(artists=[Artist(name=u'Chairlift')], name=u'Moth') + ), + ( + 'tests/data/track1.json', + Album(artists=[Artist(name=u'Chairlift')], name=u'Moth') + ), + ( + 'tests/data/track2.json', + Album(artists=[Artist(name=u'Chairlift')], name=u'Moth') + ), +]) +def test_create_album(data, expected, emby_client): + with open(data, 'r') as f: + track = json.load(f) + + assert emby_client.create_album(track) == expected + + +@pytest.mark.parametrize('data,expected', [ + ( + 'tests/data/track0.json', + [Artist(name=u'Chairlift')] + ), + ( + 'tests/data/track1.json', + [Artist(name=u'Chairlift')] + ), + ( + 'tests/data/track2.json', + [Artist(name=u'Chairlift')] + ), +]) +def test_create_artists(data, expected, emby_client): + with open(data, 'r') as f: + track = json.load(f) + + assert emby_client.create_artists(track) == expected + + +@pytest.mark.parametrize('data,user_id', [ + ('tests/data/get_user0.json', '2ec276a2642e54a19b612b9418a8bd3b') +]) +@mock.patch('mopidy_emby.remote.requests.get') +@mock.patch('mopidy_emby.remote.EmbyHandler._get_token') +@mock.patch('mopidy_emby.remote.EmbyHandler._create_headers') +@mock.patch('mopidy_emby.remote.EmbyHandler._password_data') +def test_get_user(password_mock, create_headers_mock, get_tocken_mock, + get_mock, data, user_id, config): + + mock_response = mock.Mock() + with open(data, 'r') as f: + mock_response.json.return_value = json.load(f) + + get_mock.return_value = mock_response + + emby = backend.EmbyHandler(config) + + assert emby.user_id == user_id + + +@mock.patch('mopidy_emby.remote.requests.get') +@mock.patch('mopidy_emby.remote.EmbyHandler._get_token') +@mock.patch('mopidy_emby.remote.EmbyHandler._create_headers') +@mock.patch('mopidy_emby.remote.EmbyHandler._password_data') +def test_get_user_exception(password_mock, create_headers_mock, + get_tocken_mock, get_mock, config): + + mock_response = mock.Mock() + with open('tests/data/get_user1.json', 'r') as f: + mock_response.json.return_value = json.load(f) + + get_mock.return_value = mock_response + + with pytest.raises(Exception) as execinfo: + backend.EmbyHandler(config) + + assert 'No Emby user embyuser found' in str(execinfo.value) + + +@pytest.mark.parametrize('data,token', [ + ('tests/data/get_token0.json', 'f0d6b372b40b47299ed01b9b2d40489b'), + ('tests/data/get_token1.json', None), +]) +@mock.patch('mopidy_emby.remote.requests.post') +@mock.patch('mopidy_emby.remote.EmbyHandler._create_headers') +@mock.patch('mopidy_emby.remote.EmbyHandler._password_data') +@mock.patch('mopidy_emby.remote.EmbyHandler._get_user') +def test_get_token(get_user_mock, password_data_mock, + create_headers_mock, post_mock, data, + token, config): + + mock_response = mock.Mock() + with open(data, 'r') as f: + mock_response.json.return_value = json.load(f) + + post_mock.return_value = mock_response + + emby = backend.EmbyHandler(config) + + assert emby.token == token + + +@mock.patch('mopidy_emby.remote.requests') +@mock.patch('mopidy_emby.remote.EmbyHandler._create_headers') +@mock.patch('mopidy_emby.remote.EmbyHandler._get_user') +@mock.patch('mopidy_emby.remote.EmbyHandler._get_token') +def test_password_data(get_token_mock, get_user_mock, create_headers_mock, + requests_mock, config): + + emby = backend.EmbyHandler(config) + + assert emby._password_data() == { + 'username': 'embyuser', + 'password': '444b73bcd9dc4331104c5ef960ee240066f8a3e5', + 'passwordMd5': '1d549a7b47c46b7b0a90651360c5574c' + } + + +@pytest.mark.parametrize('token,headers', [ + ( + None, + { + 'x-emby-authorization': ('MediaBrowser UserId="123", ' + 'Client="other", Device="mopidy", ' + 'DeviceId="mopidy", Version="0.0.0"') + } + ), + ( + 'f0d6b372b40b47299ed01b9b2d40489b', + { + 'x-emby-authorization': ('MediaBrowser UserId="123", ' + 'Client="other", Device="mopidy", ' + 'DeviceId="mopidy", Version="0.0.0"'), + 'x-mediabrowser-token': 'f0d6b372b40b47299ed01b9b2d40489b' + } + ) +]) +@mock.patch('mopidy_emby.remote.requests') +@mock.patch('mopidy_emby.remote.EmbyHandler._password_data') +@mock.patch('mopidy_emby.remote.EmbyHandler._get_user') +@mock.patch('mopidy_emby.remote.EmbyHandler._get_token') +def test_create_headers(get_token_mock, get_user_mock, password_data_mock, + requests_mock, token, headers, config): + + get_user_mock.return_value = [{'Id': 123}] + get_token_mock.return_value = token + + emby = backend.EmbyHandler(config) + + assert emby.headers == headers + + +@pytest.mark.parametrize('query,data,expected', [ + ( + {'track_name': ['viva hate']}, + 'tests/data/search_audio0.json', + SearchResult( + tracks=[ + Track( + album=Album( + artists=[Artist(name=u'Rainer Maria')], + name=u'Past Worn Searching' + ), + artists=[Artist(name=u'Rainer Maria')], + name='Viva Anger, Viva Hate', + track_no=3, + uri='emby:track:b5d600663238be5b41da4d8429db85f0' + ) + ], + uri='emby:search' + ) + ), + ( + {'album': ['viva hate']}, + 'tests/data/search_album0.json', + SearchResult( + albums=[ + Album( + artists=[Artist(name=u'Morrissey')], + name=u'Viva Hate', + uri='emby:album:4bf594cb601ec46a0295729c4d0f7f80') + ], + uri='emby:search' + ) + ), + ( + {'artist': ['morrissey']}, + 'tests/data/search_artist0.json', + SearchResult( + artists=[ + Artist( + name=u'Morrissey', + uri='emby:artist:0b74a057d86092f48698be681737c4ed' + ), + Artist( + name=u'Morrissey & Siouxsie Sioux', + uri='emby:artist:32bbd3db105255b24a83d0d955179dc4' + ), + Artist( + name=u'Morrissey & Siouxsie Sioux', + uri='emby:artist:eb69a3f2db13691d24c6a9794926ddb8' + ) + ], + uri='emby:search' + ) + ) +]) +@mock.patch('mopidy_emby.backend.EmbyHandler._get_search') +def test_search(get_search_mock, query, data, expected, emby_client): + with open(data, 'r') as f: + get_search_mock.return_value = json.load(f)['SearchHints'] + + assert emby_client.search(query) == expected