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)