Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
ref = cf[0]
assert isinstance(ref, NotifyBase) is True
ref_popped = cf.pop(0)
assert isinstance(ref_popped, NotifyBase) is True
assert ref == ref_popped
assert len(cf) == 0
# reference to calls on initial reference
cf = ConfigFile(path=str(t), format='text')
assert isinstance(cf.pop(0), NotifyBase) is True
cf = ConfigFile(path=str(t), format='text')
assert isinstance(cf[0], NotifyBase) is True
# Second reference actually uses cache
assert isinstance(cf[0], NotifyBase) is True
cf = ConfigFile(path=str(t), format='text')
# Itereator creation (nothing needed to assert here)
iter(cf)
# Second reference actually uses cache
iter(cf)
# Cache Handling; cache each request for 30 seconds
results = ConfigFile.parse_url(
'file://{}?cache=30'.format(str(t)))
assert isinstance(results, dict)
cf = ConfigFile(**results)
assert isinstance(cf.url(), six.string_types) is True
assert isinstance(cf.read(), six.string_types) is True
# over-ride this function. So direct calls to this throws a not
# implemented error intentionally
assert True
try:
nb.send('test message')
assert False
except NotImplementedError:
# Each sub-module is that inherits this as a parent is required to
# over-ride this function. So direct calls to this throws a not
# implemented error intentionally
assert True
# Throttle overrides..
nb = NotifyBase()
nb.request_rate_per_sec = 0.0
start_time = default_timer()
nb.throttle()
elapsed = default_timer() - start_time
# Should be a very fast response time since we set it to zero but we'll
# check for less then 500 to be fair as some testing systems may be slower
# then other
assert elapsed < 0.5
# Concurrent calls should achieve the same response
start_time = default_timer()
nb.throttle()
elapsed = default_timer() - start_time
assert elapsed < 0.5
nb = NotifyBase()
# This entry is a bit hacky, but it allows us to unit-test this library
# in an environment that simply doesn't have the gnome packages
# available to us. It also allows us to handle situations where the
# packages actually are present but we need to test that they aren't.
# If anyone is seeing this had knows a better way of testing this
# outside of what is defined in test/test_glib_plugin.py, please
# let me know! :)
_enabled = NOTIFY_DBUS_SUPPORT_ENABLED
# Define object templates
templates = (
'{schema}://_/',
)
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'urgency': {
'name': _('Urgency'),
'type': 'choice:int',
'values': DBUS_URGENCIES,
'default': DBusUrgency.NORMAL,
},
'x': {
'name': _('X-Axis'),
'type': 'int',
'min': 0,
'map_to': 'x_axis',
},
'y': {
'name': _('Y-Axis'),
'type': 'int',
'min': 0,
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to substantiate this object.
"""
results = NotifyBase.parse_url(url, verify_host=False)
if not results:
# We're done early as we couldn't load the results
return results
# The first token is stored in the hostname
results['token'] = NotifyWebexTeams.unquote(results['host'])
return results
from json import dumps
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import validate_regex
from ..AppriseLocale import gettext_lazy as _
# Extend HTTP Error Messages
KUMULOS_HTTP_ERROR_MAP = {
401: 'Unauthorized - Invalid API and/or Server Key.',
422: 'Unprocessable Entity - The request was unparsable.',
400: 'Bad Request - Targeted users do not exist or have unsubscribed.',
}
class NotifyKumulos(NotifyBase):
"""
A wrapper for Kumulos Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Kumulos'
# The services URL
service_url = 'https://kumulos.com/'
# The default secure protocol
secure_protocol = 'kumulos'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_kumulos'
# You will need this to make this notification work correctly
#
# For each event you create you will assign it a name (this will be known as
# the {event} when building your URL.
import re
import requests
from json import dumps
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..utils import validate_regex
from ..AppriseLocale import gettext_lazy as _
class NotifyIFTTT(NotifyBase):
"""
A wrapper for IFTTT Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'IFTTT'
# The services URL
service_url = 'https://ifttt.com/'
# The default protocol
secure_protocol = 'ifttt'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_ifttt'
# 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 ..URLBase import PrivacyMode
from ..utils import parse_bool
from ..common import NotifyType
from .. import __version__ as VERSION
from ..AppriseLocale import gettext_lazy as _
class NotifyEmby(NotifyBase):
"""
A wrapper for Emby Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'Emby'
# The services URL
service_url = 'https://emby.media/'
# The default protocol
protocol = 'emby'
# The default secure protocol
secure_protocol = 'embys'
# A URL that takes you to the setup/help of the specific protocol
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Apply our settings now
# The first token is stored in the hostname
consumer_key = results['host']
# Now fetch the remaining tokens
try:
consumer_secret, access_token_key, access_token_secret = \
[x for x in filter(bool, NotifyBase.split_path(
results['fullpath']))][0:3]
except (ValueError, AttributeError, IndexError):
# Force some bad values that will get caught
# in parsing later
consumer_secret = None
access_token_key = None
access_token_secret = None
results['ckey'] = consumer_key
results['csecret'] = consumer_secret
results['akey'] = access_token_key
results['asecret'] = access_token_secret
return results