mirror of
https://github.com/caronc/apprise.git
synced 2024-11-08 09:14:53 +01:00
Emby Support + testing; refs #2
This commit is contained in:
Chris Caron
parent
5b9be6bcc4
commit
16f1c2b78c
@ -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<br />boxcar://hostname/@tag<br/>boxcar://hostname/device_token<br />boxcar://hostname/device_token1/device_token2/device_tokenN<br />boxcar://hostname/@tag/@tag2/device_token
|
||||
| [Discord](https://github.com/caronc/apprise/wiki/Notify_discord) | discord:// | (TCP) 443 | discord://webhook_id/webhook_token<br />discord://avatar@webhook_id/webhook_token
|
||||
| [Emby](https://github.com/caronc/apprise/wiki/Notify_emby) | emby:// or embys:// | (TCP) 8096 | emby://user@hostname/<br />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<br />growl://hostname:portno<br />growl://password@hostname<br />growl://password@hostname:port</br>**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<br />join://apikey/device1/device2/deviceN/<br />join://apikey/group<br />join://apikey/groupA/groupB/groupN<br />join://apikey/DeviceA/groupA/groupN/DeviceN/
|
||||
|
||||
578
apprise/plugins/NotifyEmby.py
Normal file
578
apprise/plugins/NotifyEmby.py
Normal file
@ -0,0 +1,578 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Emby Notify Wrapper
|
||||
#
|
||||
# Copyright (C) 2017-2018 Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# 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()
|
||||
@ -2,7 +2,7 @@
|
||||
#
|
||||
# Our service wrappers
|
||||
#
|
||||
# Copyright (C) 2017 Chris Caron <lead2gold@gmail.com>
|
||||
# Copyright (C) 2017-2018 Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# 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',
|
||||
|
||||
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user