mirror of
https://github.com/jellyfin/mopidy-jellyfin.git
synced 2024-11-27 07:30:22 +00:00
major refactoring
This commit is contained in:
Marvin Steadfast
parent
cb134a38c7
commit
a021e4c09b
@ -71,6 +71,11 @@ Credits
|
||||
Changelog
|
||||
=========
|
||||
|
||||
v0.2.0
|
||||
---------------------------------------
|
||||
|
||||
- Alot of splitting and refactoring
|
||||
|
||||
v0.1.3
|
||||
----------------------------------------
|
||||
|
||||
|
||||
@ -6,7 +6,7 @@ import os
|
||||
from mopidy import config, ext
|
||||
|
||||
|
||||
__version__ = '0.1.3'
|
||||
__version__ = '0.2.0'
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@ -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:<artist_id>
|
||||
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:<album_id>
|
||||
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
|
||||
)
|
||||
|
||||
86
mopidy_emby/library.py
Normal file
86
mopidy_emby/library.py
Normal file
@ -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:<artist_id>
|
||||
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:<album_id>
|
||||
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)
|
||||
26
mopidy_emby/playback.py
Normal file
26
mopidy_emby/playback.py
Normal file
@ -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
|
||||
464
mopidy_emby/remote.py
Normal file
464
mopidy_emby/remote.py
Normal file
@ -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
|
||||
)
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
84
tests/test_library.py
Normal file
84
tests/test_library.py
Normal file
@ -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
|
||||
17
tests/test_playback.py
Normal file
17
tests/test_playback.py
Normal file
@ -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
|
||||
382
tests/test_remote.py
Normal file
382
tests/test_remote.py
Normal file
@ -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
|
||||
Loading…
Reference in New Issue
Block a user