From 16f1c2b78c47153e9169e3f5bb5d5c66e553a995 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sun, 4 Mar 2018 21:06:41 -0500 Subject: [PATCH] Emby Support + testing; refs #2 --- README.md | 1 + apprise/plugins/NotifyEmby.py | 578 ++++++++++++++++++++++++++++++++++ apprise/plugins/__init__.py | 14 +- test/test_rest_plugins.py | 453 ++++++++++++++++++++++++-- 4 files changed, 1011 insertions(+), 35 deletions(-) create mode 100644 apprise/plugins/NotifyEmby.py diff --git a/README.md b/README.md index 76beab66..bba01f00 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ The table below identifies the services this tool supports and some example serv | -------------------- | ---------- | ------------ | -------------- | | [Boxcar](https://github.com/caronc/apprise/wiki/Notify_boxcar) | boxcar:// | (TCP) 443 | boxcar://hostname
boxcar://hostname/@tag
boxcar://hostname/device_token
boxcar://hostname/device_token1/device_token2/device_tokenN
boxcar://hostname/@tag/@tag2/device_token | [Discord](https://github.com/caronc/apprise/wiki/Notify_discord) | discord:// | (TCP) 443 | discord://webhook_id/webhook_token
discord://avatar@webhook_id/webhook_token +| [Emby](https://github.com/caronc/apprise/wiki/Notify_emby) | emby:// or embys:// | (TCP) 8096 | emby://user@hostname/
emby://user:password@hostname | [Faast](https://github.com/caronc/apprise/wiki/Notify_faast) | faast:// | (TCP) 443 | faast://authorizationtoken | [Growl](https://github.com/caronc/apprise/wiki/Notify_growl) | growl:// | (UDP) 23053 | growl://hostname
growl://hostname:portno
growl://password@hostname
growl://password@hostname:port
**Note**: you can also use the get parameter _version_ which can allow the growl request to behave using the older v1.x protocol. An example would look like: growl://hostname?version=1 | [Join](https://github.com/caronc/apprise/wiki/Notify_join) | join:// | (TCP) 443 | join://apikey/device
join://apikey/device1/device2/deviceN/
join://apikey/group
join://apikey/groupA/groupB/groupN
join://apikey/DeviceA/groupA/groupN/DeviceN/ diff --git a/apprise/plugins/NotifyEmby.py b/apprise/plugins/NotifyEmby.py new file mode 100644 index 00000000..074e67f0 --- /dev/null +++ b/apprise/plugins/NotifyEmby.py @@ -0,0 +1,578 @@ +# -*- coding: utf-8 -*- +# +# Emby Notify Wrapper +# +# Copyright (C) 2017-2018 Chris Caron +# +# This file is part of apprise. +# +# This program is free software; you can redistribute it and/or modify it +# under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation; either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. + +# For this plugin to work correct, the Emby server must be set up to allow +# for remote connections. + +# Emby Docker configuration: https://hub.docker.com/r/emby/embyserver/ +# Authentication: https://github.com/MediaBrowser/Emby/wiki/Authentication +# Notifications: https://github.com/MediaBrowser/Emby/wiki/Remote-control +import requests +import hashlib +from json import dumps +from json import loads + +from .NotifyBase import NotifyBase +from .NotifyBase import HTTP_ERROR_MAP +from ..utils import parse_bool +from .. import __version__ as VERSION + + +class NotifyEmby(NotifyBase): + """ + A wrapper for Emby Notifications + """ + + # The default protocol + protocol = 'emby' + + # The default secure protocol + secure_protocol = 'embys' + + # Emby uses the http protocol with JSON requests + emby_default_port = 8096 + + # By default Emby requires you to provide it a device id + # The following was just a random uuid4 generated one. There + # is no real reason to change this, but hey; that's what open + # source is for right? + emby_device_id = '48df9504-6843-49be-9f2d-a685e25a0bc8' + + # The Emby message timeout; basically it is how long should our message be + # displayed for. The value is in milli-seconds + emby_message_timeout_ms = 60000 + + def __init__(self, modal=False, **kwargs): + """ + Initialize Emby Object + + """ + super(NotifyEmby, self).__init__( + title_maxlen=250, body_maxlen=32768, **kwargs) + + if self.secure: + self.schema = 'https' + + else: + self.schema = 'http' + + # Our access token does not get created until we first + # authenticate with our Emby server. The same goes for the + # user id below. + self.access_token = None + self.user_id = None + + # Whether or not our popup dialog is a timed notification + # or a modal type box (requires an Okay acknowledgement) + self.modal = modal + + if not self.user: + # Token was None + self.logger.warning('No Username was specified.') + raise TypeError('No Username was specified.') + + return + + def login(self, **kwargs): + """ + Creates our authentication token and prepares our header + + """ + + if self.is_authenticated: + # Log out first before we log back in + self.logout() + + # Prepare our login url + url = '%s://%s' % (self.schema, self.host) + if self.port: + url += ':%d' % self.port + + url += '/Users/AuthenticateByName' + + # Initialize our payload + payload = { + 'Username': self.user + } + + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + 'X-Emby-Authorization': self.emby_auth_header, + } + + if self.password: + # Source: https://github.com/MediaBrowser/Emby/wiki/Authentication + # We require the following during our authentication + # pw - password in plain text + # password - password in Sha1 + # passwordMd5 - password in MD5 + payload['pw'] = self.password + + password_md5 = hashlib.md5() + password_md5.update(self.password.encode('utf-8')) + payload['passwordMd5'] = password_md5.hexdigest() + + password_sha1 = hashlib.sha1() + password_sha1.update(self.password.encode('utf-8')) + payload['password'] = password_sha1.hexdigest() + + else: + # Backwards compatibility + payload['password'] = '' + payload['passwordMd5'] = '' + + # April 1st, 2018 and newer requirement: + payload['pw'] = '' + + self.logger.debug( + 'Emby login() POST URL: %s (cert_verify=%r)' % ( + url, self.verify_certificate)) + + try: + r = requests.post( + url, + headers=headers, + data=dumps(payload), + verify=self.verify_certificate, + ) + + if r.status_code != requests.codes.ok: + try: + self.logger.warning( + 'Failed to authenticate user %s details: ' + '%s (error=%s).' % ( + self.user, + HTTP_ERROR_MAP[r.status_code], + r.status_code)) + + except KeyError: + self.logger.warning( + 'Failed to authenticate user %s details: ' + '(error=%s).' % (self.user, r.status_code)) + + self.logger.debug('Emby Response:\r\n%s' % r.text) + + # Return; we're done + return False + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occured authenticating a user with Emby ' + 'at %s.' % self.host) + self.logger.debug('Socket Exception: %s' % str(e)) + + # Return; we're done + return False + + # Load our results + try: + results = loads(r.content) + + except ValueError: + # A string like '' would cause this; basicallly the content + # that was provided was not a JSON string. We can stop here + return False + + # Acquire our Access Token + self.access_token = results.get('AccessToken') + + # Acquire our UserId. It can be in one (or both) of the + # following locations in the response: + # { + # 'User': { + # ... + # 'Id': 'the_user_id_can_be_here', + # ... + # }, + # 'Id': 'the_user_id_can_be_found_here_too', + # } + # + # The below just safely covers both grounds. + self.user_id = results.get('Id') + if not self.user_id: + if 'User' in results: + self.user_id = results['User'].get('Id') + + # No user was found matching the specified + return self.is_authenticated + + def sessions(self, user_controlled=True): + """ + Acquire our Session Identifiers and store them in a dictionary + indexed by the session id itself. + + """ + # A single session might look like this: + # { + # u'AdditionalUsers': [], + # u'ApplicationVersion': u'3.3.1.0', + # u'Client': u'Emby Mobile', + # u'DeviceId': u'00c901e90ae814c00f81c75ae06a1c8a4381f45b', + # u'DeviceName': u'Firefox', + # u'Id': u'e37151ea06d7eb636639fded5a80f223', + # u'LastActivityDate': u'2018-03-04T21:29:02.5590200Z', + # u'PlayState': { + # u'CanSeek': False, + # u'IsMuted': False, + # u'IsPaused': False, + # u'RepeatMode': u'RepeatNone', + # }, + # u'PlayableMediaTypes': [u'Audio', u'Video'], + # u'RemoteEndPoint': u'172.17.0.1', + # u'ServerId': u'4470e977ea704a08b264628c24127d43', + # u'SupportedCommands': [ + # u'MoveUp', + # u'MoveDown', + # u'MoveLeft', + # u'MoveRight', + # u'PageUp', + # u'PageDown', + # u'PreviousLetter', + # u'NextLetter', + # u'ToggleOsd', + # u'ToggleContextMenu', + # u'Select', + # u'Back', + # u'SendKey', + # u'SendString', + # u'GoHome', + # u'GoToSettings', + # u'VolumeUp', + # u'VolumeDown', + # u'Mute', + # u'Unmute', + # u'ToggleMute', + # u'SetVolume', + # u'SetAudioStreamIndex', + # u'SetSubtitleStreamIndex', + # u'DisplayContent', + # u'GoToSearch', + # u'DisplayMessage', + # u'SetRepeatMode', + # u'ChannelUp', + # u'ChannelDown', + # u'PlayMediaSource', + # ], + # u'SupportsRemoteControl': True, + # u'UserId': u'6f98d12cb10f48209ee282787daf7af6', + # u'UserName': u'l2g' + # } + + # Prepare a dict() object to control our sessions; the keys are + # the sessions while the details associated with the session + # are stored inside. + sessions = dict() + + if not self.is_authenticated and not self.login(): + # Authenticate if we aren't already + return sessions + + # Prepare our login url + url = '%s://%s' % (self.schema, self.host) + if self.port: + url += ':%d' % self.port + + url += '/Sessions' + + if user_controlled is True: + # Only return sessions that can be managed by the current Emby + # user. + url += '?ControllableByUserId=%s' % self.user_id + + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + 'X-Emby-Authorization': self.emby_auth_header, + 'X-MediaBrowser-Token': self.access_token, + } + + self.logger.debug( + 'Emby session() GET URL: %s (cert_verify=%r)' % ( + url, self.verify_certificate)) + + try: + r = requests.get( + url, + headers=headers, + verify=self.verify_certificate, + ) + + if r.status_code != requests.codes.ok: + try: + self.logger.warning( + 'Failed to acquire session for user %s details: ' + '%s (error=%s).' % ( + self.user, + HTTP_ERROR_MAP[r.status_code], + r.status_code)) + + except KeyError: + self.logger.warning( + 'Failed to acquire session for user %s details: ' + '(error=%s).' % (self.user, r.status_code)) + + self.logger.debug('Emby Response:\r\n%s' % r.text) + + # Return; we're done + return sessions + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occured querying Emby ' + 'for session information at %s.' % self.host) + self.logger.debug('Socket Exception: %s' % str(e)) + + # Return; we're done + return sessions + + # Load our results + try: + results = loads(r.content) + + except ValueError: + # A string like '' would cause this; basicallly the content + # that was provided was not a JSON string. There is nothing + # more we can do at this point + return sessions + + for entry in results: + session = entry.get('Id') + if session: + sessions[session] = entry + + return sessions + + def logout(self, **kwargs): + """ + Logs out of an already-authenticated session + + """ + if not self.is_authenticated: + # We're not authenticated; there is nothing to do + return True + + # Prepare our login url + url = '%s://%s' % (self.schema, self.host) + if self.port: + url += ':%d' % self.port + + url += '/Sessions/Logout' + + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + 'X-Emby-Authorization': self.emby_auth_header, + 'X-MediaBrowser-Token': self.access_token, + } + + self.logger.debug( + 'Emby logout() POST URL: %s (cert_verify=%r)' % ( + url, self.verify_certificate)) + try: + r = requests.post( + url, + headers=headers, + verify=self.verify_certificate, + ) + + if r.status_code not in ( + # We're already logged out + requests.codes.unauthorized, + # The below show up if we were 'just' logged out + requests.codes.ok, + requests.codes.no_content): + try: + self.logger.warning( + 'Failed to logoff user %s details: ' + '%s (error=%s).' % ( + self.user, + HTTP_ERROR_MAP[r.status_code], + r.status_code)) + + except KeyError: + self.logger.warning( + 'Failed to logoff user %s details: ' + '(error=%s).' % (self.user, r.status_code)) + + self.logger.debug('Emby Response:\r\n%s' % r.text) + + # Return; we're done + return False + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occured querying Emby ' + 'to logoff user %s at %s.' % (self.user, self.host)) + self.logger.debug('Socket Exception: %s' % str(e)) + + # Return; we're done + return False + + # We logged our successfully if we reached here + + # Reset our variables + self.access_token = None + self.user_id = None + return True + + def notify(self, title, body, notify_type, **kwargs): + """ + Perform Emby Notification + """ + if not self.is_authenticated and not self.login(): + # Authenticate if we aren't already + return False + + # Acquire our list of sessions + sessions = self.sessions().keys() + if not sessions: + self.logger.warning('There were no Emby sessions to notify.') + # We don't need to fail; there really is no one to notify + return True + + url = '%s://%s' % (self.schema, self.host) + if self.port: + url += ':%d' % self.port + + # Append our remaining path + url += '/Sessions/%s/Message' + + # Prepare Emby Object + payload = { + 'Header': title, + 'Text': body, + } + + if not self.modal: + payload['TimeoutMs'] = self.emby_message_timeout_ms + + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + 'X-Emby-Authorization': self.emby_auth_header, + 'X-MediaBrowser-Token': self.access_token, + } + + # Track whether or not we had a failure or not. + has_error = False + + for session in sessions: + # Update our session + session_url = url % session + + self.logger.debug('Emby POST URL: %s (cert_verify=%r)' % ( + session_url, self.verify_certificate, + )) + self.logger.debug('Emby Payload: %s' % str(payload)) + try: + r = requests.post( + session_url, + data=dumps(payload), + headers=headers, + verify=self.verify_certificate, + ) + if r.status_code not in ( + requests.codes.ok, + requests.codes.no_content): + try: + self.logger.warning( + 'Failed to send Emby notification: ' + '%s (error=%s).' % ( + HTTP_ERROR_MAP[r.status_code], + r.status_code)) + + except KeyError: + self.logger.warning( + 'Failed to send Emby notification ' + '(error=%s).' % (r.status_code)) + + # Mark our failure + has_error = True + continue + + else: + self.logger.info('Sent Emby notification.') + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occured sending Emby ' + 'notification to %s.' % self.host) + self.logger.debug('Socket Exception: %s' % str(e)) + + # Mark our failure + has_error = True + continue + + return not has_error + + @property + def is_authenticated(self): + """ + Returns True if we're authenticated and False if not. + + """ + return True if self.access_token and self.user_id else False + + @property + def emby_auth_header(self): + """ + Generates the X-Emby-Authorization header response based on whether + we're authenticated or not. + + """ + # Specific to Emby + header_args = [ + ('MediaBrowser Client', self.app_id), + ('Device', self.app_id), + ('DeviceId', self.emby_device_id), + ('Version', str(VERSION)), + ] + + if self.user_id: + # Append UserId variable if we're authenticated + header_args.append(('UserId', self.user)) + + return ', '.join(['%s="%s"' % (k, v) for k, v in header_args]) + + @staticmethod + def parse_url(url): + """ + Parses the URL and returns enough arguments that can allow + us to substantiate this object. + + """ + results = NotifyBase.parse_url(url) + if not results: + # We're done early + return results + + # Assign Default Emby Port + if not results['port']: + results['port'] = NotifyEmby.emby_default_port + + # Modal type popup (default False) + results['modal'] = parse_bool(results['qsd'].get('modal', False)) + + return results + + def __del__(self): + """ + Deconstructor + """ + self.logout() diff --git a/apprise/plugins/__init__.py b/apprise/plugins/__init__.py index 19699182..9fe62d83 100644 --- a/apprise/plugins/__init__.py +++ b/apprise/plugins/__init__.py @@ -2,7 +2,7 @@ # # Our service wrappers # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -23,6 +23,7 @@ from . import NotifyEmail as NotifyEmailBase from .NotifyBoxcar import NotifyBoxcar from .NotifyDiscord import NotifyDiscord from .NotifyEmail import NotifyEmail +from .NotifyEmby import NotifyEmby from .NotifyFaast import NotifyFaast from .NotifyGrowl.NotifyGrowl import NotifyGrowl from .NotifyGrowl import gntp @@ -52,11 +53,12 @@ from ..common import NOTIFY_TYPES __all__ = [ # Notification Services - 'NotifyBoxcar', 'NotifyEmail', 'NotifyFaast', 'NotifyGrowl', 'NotifyJSON', - 'NotifyMyAndroid', 'NotifyProwl', 'NotifyPushalot', 'NotifyPushBullet', - 'NotifyPushover', 'NotifyRocketChat', 'NotifyToasty', 'NotifyTwitter', - 'NotifyXBMC', 'NotifyXML', 'NotifySlack', 'NotifyJoin', 'NotifyTelegram', - 'NotifyMatterMost', 'NotifyPushjet', 'NotifyDiscord', + 'NotifyBoxcar', 'NotifyEmail', 'NotifyEmby', 'NotifyDiscord', + 'NotifyFaast', 'NotifyGrowl', 'NotifyJoin', 'NotifyJSON', + 'NotifyMatterMost', 'NotifyMyAndroid', 'NotifyProwl', 'NotifyPushalot', + 'NotifyPushBullet', 'NotifyPushjet', 'NotifyPushover', 'NotifyRocketChat', + 'NotifySlack', 'NotifyToasty', 'NotifyTwitter', 'NotifyTelegram', + 'NotifyXBMC', 'NotifyXML', # Reference 'NotifyImageSize', 'NOTIFY_IMAGE_SIZES', 'NotifyType', 'NOTIFY_TYPES', diff --git a/test/test_rest_plugins.py b/test/test_rest_plugins.py index 1eceeff7..ca4c360a 100644 --- a/test/test_rest_plugins.py +++ b/test/test_rest_plugins.py @@ -25,6 +25,20 @@ from json import dumps import requests import mock +# Some exception handling we'll use +REQUEST_EXCEPTIONS = ( + requests.ConnectionError( + 0, 'requests.ConnectionError() not handled'), + requests.RequestException( + 0, 'requests.RequestException() not handled'), + requests.HTTPError( + 0, 'requests.HTTPError() not handled'), + requests.ReadTimeout( + 0, 'requests.ReadTimeout() not handled'), + requests.TooManyRedirects( + 0, 'requests.TooManyRedirects() not handled'), +) + TEST_URLS = ( ################################## # NotifyBoxcar @@ -146,6 +160,42 @@ TEST_URLS = ( 'test_requests_exceptions': True, }), + ################################## + # NotifyEmby + ################################## + # Insecure Request; no hostname specified + ('emby://', { + 'instance': None, + }), + # Secure Emby Request; no hostname specified + ('embys://', { + 'instance': None, + }), + # No user specified + ('emby://localhost', { + # Missing a username + 'instance': TypeError, + }), + ('emby://:@/', { + 'instance': None, + }), + # Valid Authentication + ('emby://l2g@localhost', { + 'instance': plugins.NotifyEmby, + # our response will be False because our authentication can't be + # tested very well using this matrix. It will resume in + # in test_notify_emby_plugin() + 'response': False, + }), + ('embys://l2g:password@localhost', { + 'instance': plugins.NotifyEmby, + # our response will be False because our authentication can't be + # tested very well using this matrix. It will resume in + # in test_notify_emby_plugin() + 'response': False, + }), + # The rest of the emby tests are in test_notify_emby_plugin() + ################################## # NotifyFaast ################################## @@ -1239,7 +1289,6 @@ def test_rest_plugins(mock_post, mock_get): # iterate over our dictionary and test it out for (url, meta) in TEST_URLS: - # Our expected instance instance = meta.get('instance', None) @@ -1285,6 +1334,8 @@ def test_rest_plugins(mock_post, mock_get): setattr(robj, 'raw', mock.Mock()) # Allow raw.read() calls robj.raw.read.return_value = '' + robj.text = '' + robj.content = '' mock_get.return_value = robj mock_post.return_value = robj @@ -1304,18 +1355,7 @@ def test_rest_plugins(mock_post, mock_get): else: # Handle exception testing; first we turn the boolean flag ito # a list of exceptions - test_requests_exceptions = ( - requests.ConnectionError( - 0, 'requests.ConnectionError() not handled'), - requests.RequestException( - 0, 'requests.RequestException() not handled'), - requests.HTTPError( - 0, 'requests.HTTPError() not handled'), - requests.ReadTimeout( - 0, 'requests.ReadTimeout() not handled'), - requests.TooManyRedirects( - 0, 'requests.TooManyRedirects() not handled'), - ) + test_requests_exceptions = REQUEST_EXCEPTIONS try: obj = Apprise.instantiate( @@ -1346,7 +1386,7 @@ def test_rest_plugins(mock_post, mock_get): notify_type=notify_type) == response else: - for _exception in test_requests_exceptions: + for _exception in REQUEST_EXCEPTIONS: mock_post.side_effect = _exception mock_get.side_effect = _exception @@ -1493,6 +1533,375 @@ def test_notify_discord_plugin(mock_post, mock_get): notify_type=NotifyType.INFO) is True +@mock.patch('requests.get') +@mock.patch('requests.post') +def test_notify_emby_plugin_login(mock_post, mock_get): + """ + API: NotifyEmby.login() + + """ + + # Prepare Mock + mock_get.return_value = requests.Request() + mock_post.return_value = requests.Request() + + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost') + assert isinstance(obj, plugins.NotifyEmby) + + # Test our exception handling + for _exception in REQUEST_EXCEPTIONS: + mock_post.side_effect = _exception + mock_get.side_effect = _exception + # We'll fail to log in each time + assert obj.login() is False + + # Disable Exceptions + mock_post.side_effect = None + mock_get.side_effect = None + + # Our login flat out fails if we don't have proper parseable content + mock_post.return_value.content = u'' + mock_post.return_value.text = '' + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # KeyError handling + mock_post.return_value.status_code = 999 + mock_get.return_value.status_code = 999 + assert obj.login() is False + + # General Internal Server Error + mock_post.return_value.status_code = requests.codes.internal_server_error + mock_get.return_value.status_code = requests.codes.internal_server_error + assert obj.login() is False + + mock_post.return_value.status_code = requests.codes.ok + mock_get.return_value.status_code = requests.codes.ok + + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost:%d' % ( + # Increment our port so it will always be something different than + # the default + plugins.NotifyEmby.emby_default_port + 1)) + assert isinstance(obj, plugins.NotifyEmby) + assert obj.port == (plugins.NotifyEmby.emby_default_port + 1) + + # The login will fail because '' is not a parseable JSON response + assert obj.login() is False + + # Disable the port completely + obj.port = None + assert obj.login() is False + + # Default port assigments + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost') + assert isinstance(obj, plugins.NotifyEmby) + assert obj.port == plugins.NotifyEmby.emby_default_port + + # The login will (still) fail because '' is not a parseable JSON response + assert obj.login() is False + + # Our login flat out fails if we don't have proper parseable content + mock_post.return_value.content = dumps({ + u'AccessToken': u'0000-0000-0000-0000', + }) + mock_post.return_value.text = str(mock_post.return_value.content) + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost') + assert isinstance(obj, plugins.NotifyEmby) + + # The login will fail because the 'User' or 'Id' field wasn't parsed + assert obj.login() is False + + # Our text content (we intentionally reverse the 2 locations + # that store the same thing; we do this so we can test which + # one it defaults to if both are present + mock_post.return_value.content = dumps({ + u'User': { + u'Id': u'abcd123', + }, + u'Id': u'123abc', + u'AccessToken': u'0000-0000-0000-0000', + }) + mock_post.return_value.text = str(mock_post.return_value.content) + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost') + assert isinstance(obj, plugins.NotifyEmby) + + # Login + assert obj.login() is True + assert obj.user_id == '123abc' + assert obj.access_token == '0000-0000-0000-0000' + + # We're going to log in a second time which checks that we logout + # first before logging in again. But this time we'll scrap the + # 'Id' area and use the one found in the User area if detected + mock_post.return_value.content = dumps({ + u'User': { + u'Id': u'abcd123', + }, + u'AccessToken': u'0000-0000-0000-0000', + }) + mock_post.return_value.text = str(mock_post.return_value.content) + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # Login + assert obj.login() is True + assert obj.user_id == 'abcd123' + assert obj.access_token == '0000-0000-0000-0000' + + +@mock.patch('apprise.plugins.NotifyEmby.login') +@mock.patch('apprise.plugins.NotifyEmby.logout') +@mock.patch('requests.get') +@mock.patch('requests.post') +def test_notify_emby_plugin_sessions(mock_post, mock_get, mock_logout, + mock_login): + """ + API: NotifyEmby.sessions() + + """ + + # Prepare Mock + mock_get.return_value = requests.Request() + mock_post.return_value = requests.Request() + + # This is done so we don't obstruct our access_token and user_id values + mock_login.return_value = True + mock_logout.return_value = True + + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost') + assert isinstance(obj, plugins.NotifyEmby) + obj.access_token = 'abc' + obj.user_id = '123' + + # Test our exception handling + for _exception in REQUEST_EXCEPTIONS: + mock_post.side_effect = _exception + mock_get.side_effect = _exception + # We'll fail to log in each time + sessions = obj.sessions() + assert isinstance(sessions, dict) is True + assert len(sessions) == 0 + + # Disable Exceptions + mock_post.side_effect = None + mock_get.side_effect = None + + # Our login flat out fails if we don't have proper parseable content + mock_post.return_value.content = u'' + mock_post.return_value.text = '' + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # KeyError handling + mock_post.return_value.status_code = 999 + mock_get.return_value.status_code = 999 + sessions = obj.sessions() + assert isinstance(sessions, dict) is True + assert len(sessions) == 0 + + # General Internal Server Error + mock_post.return_value.status_code = requests.codes.internal_server_error + mock_get.return_value.status_code = requests.codes.internal_server_error + sessions = obj.sessions() + assert isinstance(sessions, dict) is True + assert len(sessions) == 0 + + mock_post.return_value.status_code = requests.codes.ok + mock_get.return_value.status_code = requests.codes.ok + mock_post.return_value.text = str(mock_post.return_value.content) + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # Disable the port completely + obj.port = None + + sessions = obj.sessions() + assert isinstance(sessions, dict) is True + assert len(sessions) == 0 + + # Let's get some results + mock_post.return_value.content = dumps([ + { + u'Id': u'abc123', + }, + { + u'Id': u'def456', + }, + { + u'InvalidEntry': None, + }, + ]) + mock_post.return_value.text = str(mock_post.return_value.content) + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + sessions = obj.sessions(user_controlled=True) + assert isinstance(sessions, dict) is True + assert len(sessions) == 2 + + # Test it without setting user-controlled sessions + sessions = obj.sessions(user_controlled=False) + assert isinstance(sessions, dict) is True + assert len(sessions) == 2 + + # Triggers an authentication failure + obj.user_id = None + mock_login.return_value = False + sessions = obj.sessions() + assert isinstance(sessions, dict) is True + assert len(sessions) == 0 + + +@mock.patch('apprise.plugins.NotifyEmby.login') +@mock.patch('requests.get') +@mock.patch('requests.post') +def test_notify_emby_plugin_logout(mock_post, mock_get, mock_login): + """ + API: NotifyEmby.sessions() + + """ + + # Prepare Mock + mock_get.return_value = requests.Request() + mock_post.return_value = requests.Request() + + # This is done so we don't obstruct our access_token and user_id values + mock_login.return_value = True + + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost') + assert isinstance(obj, plugins.NotifyEmby) + obj.access_token = 'abc' + obj.user_id = '123' + + # Test our exception handling + for _exception in REQUEST_EXCEPTIONS: + mock_post.side_effect = _exception + mock_get.side_effect = _exception + # We'll fail to log in each time + obj.logout() + obj.access_token = 'abc' + obj.user_id = '123' + + # Disable Exceptions + mock_post.side_effect = None + mock_get.side_effect = None + + # Our login flat out fails if we don't have proper parseable content + mock_post.return_value.content = u'' + mock_post.return_value.text = '' + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # KeyError handling + mock_post.return_value.status_code = 999 + mock_get.return_value.status_code = 999 + obj.logout() + obj.access_token = 'abc' + obj.user_id = '123' + + # General Internal Server Error + mock_post.return_value.status_code = requests.codes.internal_server_error + mock_get.return_value.status_code = requests.codes.internal_server_error + obj.logout() + obj.access_token = 'abc' + obj.user_id = '123' + + mock_post.return_value.status_code = requests.codes.ok + mock_get.return_value.status_code = requests.codes.ok + mock_post.return_value.text = str(mock_post.return_value.content) + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # Disable the port completely + obj.port = None + obj.logout() + + +@mock.patch('apprise.plugins.NotifyEmby.sessions') +@mock.patch('apprise.plugins.NotifyEmby.login') +@mock.patch('apprise.plugins.NotifyEmby.logout') +@mock.patch('requests.get') +@mock.patch('requests.post') +def test_notify_emby_plugin_notify(mock_post, mock_get, mock_logout, + mock_login, mock_sessions): + """ + API: NotifyEmby.notify() + + """ + + # Prepare Mock + mock_get.return_value = requests.Request() + mock_post.return_value = requests.Request() + mock_post.return_value.status_code = requests.codes.ok + mock_get.return_value.status_code = requests.codes.ok + + # This is done so we don't obstruct our access_token and user_id values + mock_login.return_value = True + mock_logout.return_value = True + mock_sessions.return_value = {'abcd': {}} + + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost?modal=False') + assert isinstance(obj, plugins.NotifyEmby) + assert obj.notify('title', 'body', 'info') is True + obj.access_token = 'abc' + obj.user_id = '123' + + # Test Modal support + obj = Apprise.instantiate('emby://l2g:l2gpass@localhost?modal=True') + assert isinstance(obj, plugins.NotifyEmby) + assert obj.notify('title', 'body', 'info') is True + obj.access_token = 'abc' + obj.user_id = '123' + + # Test our exception handling + for _exception in REQUEST_EXCEPTIONS: + mock_post.side_effect = _exception + mock_get.side_effect = _exception + # We'll fail to log in each time + assert obj.notify('title', 'body', 'info') is False + + # Disable Exceptions + mock_post.side_effect = None + mock_get.side_effect = None + + # Our login flat out fails if we don't have proper parseable content + mock_post.return_value.content = u'' + mock_post.return_value.text = '' + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # KeyError handling + mock_post.return_value.status_code = 999 + mock_get.return_value.status_code = 999 + assert obj.notify('title', 'body', 'info') is False + + # General Internal Server Error + mock_post.return_value.status_code = requests.codes.internal_server_error + mock_get.return_value.status_code = requests.codes.internal_server_error + assert obj.notify('title', 'body', 'info') is False + + mock_post.return_value.status_code = requests.codes.ok + mock_get.return_value.status_code = requests.codes.ok + mock_post.return_value.text = str(mock_post.return_value.content) + mock_get.return_value.content = mock_post.return_value.content + mock_get.return_value.text = mock_post.return_value.text + + # Disable the port completely + obj.port = None + assert obj.notify('title', 'body', 'info') is True + + # An Empty return set (no query is made, but notification will still + # succeed + mock_sessions.return_value = {} + assert obj.notify('title', 'body', 'info') is True + + @mock.patch('requests.get') @mock.patch('requests.post') def test_notify_join_plugin(mock_post, mock_get): @@ -2066,22 +2475,8 @@ def test_notify_telegram_plugin(mock_post, mock_get): assert nimg_obj.notify( title='title', body='body', notify_type=NotifyType.INFO) is False - # Test our exception handling with bot detection - test_requests_exceptions = ( - requests.ConnectionError( - 0, 'requests.ConnectionError() not handled'), - requests.RequestException( - 0, 'requests.RequestException() not handled'), - requests.HTTPError( - 0, 'requests.HTTPError() not handled'), - requests.ReadTimeout( - 0, 'requests.ReadTimeout() not handled'), - requests.TooManyRedirects( - 0, 'requests.TooManyRedirects() not handled'), - ) - # iterate over our exceptions and test them - for _exception in test_requests_exceptions: + for _exception in REQUEST_EXCEPTIONS: mock_post.side_effect = _exception try: obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=None)