Token details + internationalization (i18n) support; refs #59

This commit is contained in:
Chris Caron 2019-05-29 20:07:05 -04:00 committed by GitHub
parent 4c375c8a05
commit d5dfbf74fa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
63 changed files with 3988 additions and 247 deletions

1
.gitignore vendored
View File

@ -51,7 +51,6 @@ coverage.xml
# Translations
*.mo
*.pot
# Django stuff:
*.log

View File

@ -25,6 +25,7 @@ matrix:
env: TOXENV=pypy3
install:
- pip install babel
- pip install .
- pip install codecov
- pip install -r dev-requirements.txt

View File

@ -28,7 +28,6 @@ import os
import six
from markdown import markdown
from itertools import chain
from .common import NotifyType
from .common import NotifyFormat
from .utils import is_exclusive_match
@ -39,6 +38,7 @@ from .logger import logger
from .AppriseAsset import AppriseAsset
from .AppriseConfig import AppriseConfig
from .AppriseLocale import AppriseLocale
from .config.ConfigBase import ConfigBase
from .plugins.NotifyBase import NotifyBase
@ -74,48 +74,96 @@ class Apprise(object):
if servers:
self.add(servers)
# Initialize our locale object
self.locale = AppriseLocale()
@staticmethod
def instantiate(url, asset=None, tag=None, suppress_exceptions=True):
"""
Returns the instance of a instantiated plugin based on the provided
Server URL. If the url fails to be parsed, then None is returned.
The specified url can be either a string (the URL itself) or a
dictionary containing all of the components needed to istantiate
the notification service. If identifying a dictionary, at the bare
minimum, one must specify the schema.
An example of a url dictionary object might look like:
{
schema: 'mailto',
host: 'google.com',
user: 'myuser',
password: 'mypassword',
}
Alternatively the string is much easier to specify:
mailto://user:mypassword@google.com
The dictionary works well for people who are calling details() to
extract the components they need to build the URL manually.
"""
# swap hash (#) tag values with their html version
_url = url.replace('/#', '/%23')
# Attempt to acquire the schema at the very least to allow our plugins
# to determine if they can make a better interpretation of a URL
# geared for them
schema = GET_SCHEMA_RE.match(_url)
if schema is None:
logger.error('Unparseable schema:// found in URL {}.'.format(url))
return None
# Initialize our result set
results = None
# Ensure our schema is always in lower case
schema = schema.group('schema').lower()
if isinstance(url, six.string_types):
# swap hash (#) tag values with their html version
_url = url.replace('/#', '/%23')
# Some basic validation
if schema not in plugins.SCHEMA_MAP:
logger.error('Unsupported schema {}.'.format(schema))
return None
# Attempt to acquire the schema at the very least to allow our
# plugins to determine if they can make a better interpretation of
# a URL geared for them
schema = GET_SCHEMA_RE.match(_url)
if schema is None:
logger.error(
'Unparseable schema:// found in URL {}.'.format(url))
return None
# Parse our url details of the server object as dictionary containing
# all of the information parsed from our URL
results = plugins.SCHEMA_MAP[schema].parse_url(_url)
# Ensure our schema is always in lower case
schema = schema.group('schema').lower()
if results is None:
# Failed to parse the server URL
logger.error('Unparseable URL {}.'.format(url))
# Some basic validation
if schema not in plugins.SCHEMA_MAP:
logger.error('Unsupported schema {}.'.format(schema))
return None
# Parse our url details of the server object as dictionary
# containing all of the information parsed from our URL
results = plugins.SCHEMA_MAP[schema].parse_url(_url)
if results is None:
# Failed to parse the server URL
logger.error('Unparseable URL {}.'.format(url))
return None
logger.trace('URL {} unpacked as:{}{}'.format(
url, os.linesep, os.linesep.join(
['{}="{}"'.format(k, v) for k, v in results.items()])))
elif isinstance(url, dict):
# We already have our result set
results = url
if results.get('schema') not in plugins.SCHEMA_MAP:
# schema is a mandatory dictionary item as it is the only way
# we can index into our loaded plugins
logger.error('Dictionary does not include a "schema" entry.')
logger.trace('Invalid dictionary unpacked as:{}{}'.format(
os.linesep, os.linesep.join(
['{}="{}"'.format(k, v) for k, v in results.items()])))
return None
logger.trace('Dictionary unpacked as:{}{}'.format(
os.linesep, os.linesep.join(
['{}="{}"'.format(k, v) for k, v in results.items()])))
else:
logger.error('Invalid URL specified: {}'.format(url))
return None
# Build a list of tags to associate with the newly added notifications
results['tag'] = set(parse_list(tag))
logger.trace('URL {} unpacked as:{}{}'.format(
url, os.linesep, os.linesep.join(
['{}="{}"'.format(k, v) for k, v in results.items()])))
# Prepare our Asset Object
results['asset'] = \
asset if isinstance(asset, AppriseAsset) else AppriseAsset()
@ -166,6 +214,10 @@ class Apprise(object):
if len(servers) == 0:
return False
elif isinstance(servers, dict):
# no problem, we support kwargs, convert it to a list
servers = [servers]
elif isinstance(servers, (ConfigBase, NotifyBase, AppriseConfig)):
# Go ahead and just add our plugin into our list
self.servers.append(servers)
@ -184,7 +236,7 @@ class Apprise(object):
self.servers.append(_server)
continue
elif not isinstance(_server, six.string_types):
elif not isinstance(_server, (six.string_types, dict)):
logger.error(
"An invalid notification (type={}) was specified.".format(
type(_server)))
@ -195,10 +247,9 @@ class Apprise(object):
# returns None if it fails
instance = Apprise.instantiate(_server, asset=asset, tag=tag)
if not isinstance(instance, NotifyBase):
# No logging is requird as instantiate() handles failure
# and/or success reasons for us
return_status = False
logger.error(
"Failed to load notification url: {}".format(_server),
)
continue
# Add our initialized plugin to our server listings
@ -335,7 +386,7 @@ class Apprise(object):
return status
def details(self):
def details(self, lang=None):
"""
Returns the details associated with the Apprise object
@ -352,13 +403,7 @@ class Apprise(object):
}
# to add it's mapping to our hash table
for entry in sorted(dir(plugins)):
# Get our plugin
plugin = getattr(plugins, entry)
if not hasattr(plugin, 'app_id'): # pragma: no branch
# Filter out non-notification modules
continue
for plugin in set(plugins.SCHEMA_MAP.values()):
# Standard protocol(s) should be None or a tuple
protocols = getattr(plugin, 'protocol', None)
@ -370,6 +415,14 @@ class Apprise(object):
if isinstance(secure_protocols, six.string_types):
secure_protocols = (secure_protocols, )
if not lang:
# Simply return our results
details = plugins.details(plugin)
else:
# Emulate the specified language when returning our results
with self.locale.lang_at(lang):
details = plugins.details(plugin)
# Build our response object
response['schemas'].append({
'service_name': getattr(plugin, 'service_name', None),
@ -377,6 +430,7 @@ class Apprise(object):
'setup_url': getattr(plugin, 'setup_url', None),
'protocols': protocols,
'secure_protocols': secure_protocols,
'details': details,
})
return response

207
apprise/AppriseLocale.py Normal file
View File

@ -0,0 +1,207 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import six
import ctypes
import locale
import contextlib
from os.path import join
from os.path import dirname
from os.path import abspath
# Define our translation domain
DOMAIN = 'apprise'
LOCALE_DIR = abspath(join(dirname(__file__), 'i18n'))
# This gets toggled to True if we succeed
GETTEXT_LOADED = False
try:
# Initialize gettext
import gettext
# install() creates a _() in our builtins
gettext.install(DOMAIN, localedir=LOCALE_DIR)
# Toggle our flag
GETTEXT_LOADED = True
except ImportError:
# gettext isn't available; no problem, just fall back to using
# the library features without multi-language support.
try:
# Python v2.7
import __builtin__
__builtin__.__dict__['_'] = lambda x: x # pragma: no branch
except ImportError:
# Python v3.4+
import builtins
builtins.__dict__['_'] = lambda x: x # pragma: no branch
class LazyTranslation(object):
"""
Doesn't translate anything until str() or unicode() references
are made.
"""
def __init__(self, text, *args, **kwargs):
"""
Store our text
"""
self.text = text
super(LazyTranslation, self).__init__(*args, **kwargs)
def __str__(self):
return gettext.gettext(self.text)
# Lazy translation handling
def gettext_lazy(text):
"""
A dummy function that can be referenced
"""
return LazyTranslation(text=text)
class AppriseLocale(object):
"""
A wrapper class to gettext so that we can manipulate multiple lanaguages
on the fly if required.
"""
def __init__(self, language=None):
"""
Initializes our object, if a language is specified, then we
initialize ourselves to that, otherwise we use whatever we detect
from the local operating system. If all else fails, we resort to the
defined default_language.
"""
# Cache previously loaded translations
self._gtobjs = {}
# Get our language
self.lang = AppriseLocale.detect_language(language)
if GETTEXT_LOADED is False:
# We're done
return
if self.lang:
# Load our gettext object and install our language
try:
self._gtobjs[self.lang] = gettext.translation(
DOMAIN, localedir=LOCALE_DIR, languages=[self.lang])
# Install our language
self._gtobjs[self.lang].install()
except IOError:
# This occurs if we can't access/load our translations
pass
@contextlib.contextmanager
def lang_at(self, lang):
"""
The syntax works as:
with at.lang_at('fr'):
# apprise works as though the french language has been
# defined. afterwards, the language falls back to whatever
# it was.
"""
if GETTEXT_LOADED is False:
# yield
yield
# we're done
return
# Tidy the language
lang = AppriseLocale.detect_language(lang, detect_fallback=False)
# Now attempt to load it
try:
if lang in self._gtobjs:
if lang != self.lang:
# Install our language only if we aren't using it
# already
self._gtobjs[lang].install()
else:
self._gtobjs[lang] = gettext.translation(
DOMAIN, localedir=LOCALE_DIR, languages=[self.lang])
# Install our language
self._gtobjs[lang].install()
# Yield
yield
except (IOError, KeyError):
# This occurs if we can't access/load our translations
# Yield reguardless
yield
finally:
# Fall back to our previous language
if lang != self.lang and lang in self._gtobjs:
# Install our language
self._gtobjs[self.lang].install()
return
@staticmethod
def detect_language(lang=None, detect_fallback=True):
"""
returns the language (if it's retrievable)
"""
# We want to only use the 2 character version of this language
# hence en_CA becomes en, en_US becomes en.
if not isinstance(lang, six.string_types):
if detect_fallback is False:
# no detection enabled; we're done
return None
if hasattr(ctypes, 'windll'):
windll = ctypes.windll.kernel32
lang = locale.windows_locale[
windll.GetUserDefaultUILanguage()]
else:
try:
# Detect language
lang = locale.getdefaultlocale()[0]
except TypeError:
# None is returned if the default can't be determined
# we're done in this case
return None
return lang[0:2].lower()

View File

@ -86,6 +86,9 @@ class URLBase(object):
# Maintain a set of tags to associate with this specific notification
tags = set()
# Secure sites should be verified against a Certificate Authority
verify_certificate = True
# Logging
logger = logging.getLogger(__name__)

0
apprise/i18n/__init__.py Normal file
View File

289
apprise/i18n/apprise.pot Normal file
View File

@ -0,0 +1,289 @@
# Translations template for apprise.
# Copyright (C) 2019 Chris Caron
# This file is distributed under the same license as the apprise project.
# FIRST AUTHOR <EMAIL@ADDRESS>, 2019.
#
#, fuzzy
msgid ""
msgstr ""
"Project-Id-Version: apprise 0.7.6\n"
"Report-Msgid-Bugs-To: lead2gold@gmail.com\n"
"POT-Creation-Date: 2019-05-28 16:56-0400\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=utf-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Generated-By: Babel 2.6.0\n"
msgid "API Key"
msgstr ""
msgid "Access Key"
msgstr ""
msgid "Access Key ID"
msgstr ""
msgid "Access Secret"
msgstr ""
msgid "Access Token"
msgstr ""
msgid "Account SID"
msgstr ""
msgid "Add Tokens"
msgstr ""
msgid "Application Key"
msgstr ""
msgid "Application Secret"
msgstr ""
msgid "Auth Token"
msgstr ""
msgid "Authorization Token"
msgstr ""
msgid "Avatar Image"
msgstr ""
msgid "Bot Name"
msgstr ""
msgid "Bot Token"
msgstr ""
msgid "Channels"
msgstr ""
msgid "Consumer Key"
msgstr ""
msgid "Consumer Secret"
msgstr ""
msgid "Detect Bot Owner"
msgstr ""
msgid "Device ID"
msgstr ""
msgid "Display Footer"
msgstr ""
msgid "Domain"
msgstr ""
msgid "Duration"
msgstr ""
msgid "Events"
msgstr ""
msgid "Footer Logo"
msgstr ""
msgid "From Email"
msgstr ""
msgid "From Name"
msgstr ""
msgid "From Phone No"
msgstr ""
msgid "Group"
msgstr ""
msgid "HTTP Header"
msgstr ""
msgid "Hostname"
msgstr ""
msgid "Include Image"
msgstr ""
msgid "Modal"
msgstr ""
msgid "Notify Format"
msgstr ""
msgid "Organization"
msgstr ""
msgid "Overflow Mode"
msgstr ""
msgid "Password"
msgstr ""
msgid "Port"
msgstr ""
msgid "Priority"
msgstr ""
msgid "Provider Key"
msgstr ""
msgid "Region"
msgstr ""
msgid "Region Name"
msgstr ""
msgid "Remove Tokens"
msgstr ""
msgid "Rooms"
msgstr ""
msgid "SMTP Server"
msgstr ""
msgid "Schema"
msgstr ""
msgid "Secret Access Key"
msgstr ""
msgid "Secret Key"
msgstr ""
msgid "Secure Mode"
msgstr ""
msgid "Server Timeout"
msgstr ""
msgid "Sound"
msgstr ""
msgid "Source JID"
msgstr ""
msgid "Target Channel"
msgstr ""
msgid "Target Chat ID"
msgstr ""
msgid "Target Device"
msgstr ""
msgid "Target Device ID"
msgstr ""
msgid "Target Email"
msgstr ""
msgid "Target Emails"
msgstr ""
msgid "Target Encoded ID"
msgstr ""
msgid "Target JID"
msgstr ""
msgid "Target Phone No"
msgstr ""
msgid "Target Room Alias"
msgstr ""
msgid "Target Room ID"
msgstr ""
msgid "Target Short Code"
msgstr ""
msgid "Target Tag ID"
msgstr ""
msgid "Target Topic"
msgstr ""
msgid "Target User"
msgstr ""
msgid "Targets"
msgstr ""
msgid "Text To Speech"
msgstr ""
msgid "To Channel ID"
msgstr ""
msgid "To Email"
msgstr ""
msgid "To User ID"
msgstr ""
msgid "Token"
msgstr ""
msgid "Token A"
msgstr ""
msgid "Token B"
msgstr ""
msgid "Token C"
msgstr ""
msgid "Urgency"
msgstr ""
msgid "Use Avatar"
msgstr ""
msgid "User"
msgstr ""
msgid "User Key"
msgstr ""
msgid "User Name"
msgstr ""
msgid "Username"
msgstr ""
msgid "Verify SSL"
msgstr ""
msgid "Version"
msgstr ""
msgid "Webhook"
msgstr ""
msgid "Webhook ID"
msgstr ""
msgid "Webhook Mode"
msgstr ""
msgid "Webhook Token"
msgstr ""
msgid "X-Axis"
msgstr ""
msgid "XEP"
msgstr ""
msgid "Y-Axis"
msgstr ""

View File

@ -0,0 +1,293 @@
# English translations for apprise.
# Copyright (C) 2019 Chris Caron
# This file is distributed under the same license as the apprise project.
# Chris Caron <lead2gold@gmail.com>, 2019.
#
msgid ""
msgstr ""
"Project-Id-Version: apprise 0.7.6\n"
"Report-Msgid-Bugs-To: lead2gold@gmail.com\n"
"POT-Creation-Date: 2019-05-28 16:56-0400\n"
"PO-Revision-Date: 2019-05-24 20:00-0400\n"
"Last-Translator: Chris Caron <lead2gold@gmail.com>\n"
"Language: en\n"
"Language-Team: en <LL@li.org>\n"
"Plural-Forms: nplurals=2; plural=(n != 1)\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=utf-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Generated-By: Babel 2.6.0\n"
msgid "API Key"
msgstr ""
msgid "Access Key"
msgstr ""
msgid "Access Key ID"
msgstr ""
msgid "Access Secret"
msgstr ""
msgid "Access Token"
msgstr ""
msgid "Account SID"
msgstr ""
msgid "Add Tokens"
msgstr ""
msgid "Application Key"
msgstr ""
msgid "Application Secret"
msgstr ""
msgid "Auth Token"
msgstr ""
msgid "Authorization Token"
msgstr ""
msgid "Avatar Image"
msgstr ""
msgid "Bot Name"
msgstr ""
msgid "Bot Token"
msgstr ""
msgid "Channels"
msgstr ""
msgid "Consumer Key"
msgstr ""
msgid "Consumer Secret"
msgstr ""
msgid "Detect Bot Owner"
msgstr ""
msgid "Device ID"
msgstr ""
msgid "Display Footer"
msgstr ""
msgid "Domain"
msgstr ""
msgid "Duration"
msgstr ""
msgid "Events"
msgstr ""
msgid "Footer Logo"
msgstr ""
msgid "From Email"
msgstr ""
msgid "From Name"
msgstr ""
msgid "From Phone No"
msgstr ""
msgid "Group"
msgstr ""
msgid "HTTP Header"
msgstr ""
msgid "Hostname"
msgstr ""
msgid "Include Image"
msgstr ""
msgid "Modal"
msgstr ""
msgid "Notify Format"
msgstr ""
msgid "Organization"
msgstr ""
msgid "Overflow Mode"
msgstr ""
msgid "Password"
msgstr ""
msgid "Port"
msgstr ""
msgid "Priority"
msgstr ""
msgid "Provider Key"
msgstr ""
msgid "Region"
msgstr ""
msgid "Region Name"
msgstr ""
msgid "Remove Tokens"
msgstr ""
msgid "Rooms"
msgstr ""
msgid "SMTP Server"
msgstr ""
msgid "Schema"
msgstr ""
msgid "Secret Access Key"
msgstr ""
msgid "Secret Key"
msgstr ""
msgid "Secure Mode"
msgstr ""
msgid "Server Timeout"
msgstr ""
msgid "Sound"
msgstr ""
msgid "Source JID"
msgstr ""
msgid "Target Channel"
msgstr ""
msgid "Target Chat ID"
msgstr ""
msgid "Target Device"
msgstr ""
msgid "Target Device ID"
msgstr ""
msgid "Target Email"
msgstr ""
msgid "Target Emails"
msgstr ""
msgid "Target Encoded ID"
msgstr ""
msgid "Target JID"
msgstr ""
msgid "Target Phone No"
msgstr ""
msgid "Target Room Alias"
msgstr ""
msgid "Target Room ID"
msgstr ""
msgid "Target Short Code"
msgstr ""
msgid "Target Tag ID"
msgstr ""
msgid "Target Topic"
msgstr ""
msgid "Target User"
msgstr ""
msgid "Targets"
msgstr ""
msgid "Text To Speech"
msgstr ""
msgid "To Channel ID"
msgstr ""
msgid "To Email"
msgstr ""
msgid "To User ID"
msgstr ""
msgid "Token"
msgstr ""
msgid "Token A"
msgstr ""
msgid "Token B"
msgstr ""
msgid "Token C"
msgstr ""
msgid "Urgency"
msgstr ""
msgid "Use Avatar"
msgstr ""
msgid "User"
msgstr ""
msgid "User Key"
msgstr ""
msgid "User Name"
msgstr ""
msgid "Username"
msgstr ""
msgid "Verify SSL"
msgstr ""
msgid "Version"
msgstr ""
msgid "Webhook"
msgstr ""
msgid "Webhook ID"
msgstr ""
msgid "Webhook Mode"
msgstr ""
msgid "Webhook Token"
msgstr ""
msgid "X-Axis"
msgstr ""
msgid "XEP"
msgstr ""
msgid "Y-Axis"
msgstr ""
#~ msgid "Access Key Secret"
#~ msgstr ""

View File

@ -32,6 +32,7 @@ from ..common import NotifyFormat
from ..common import NOTIFY_FORMATS
from ..common import OverflowMode
from ..common import OVERFLOW_MODES
from ..AppriseLocale import gettext_lazy as _
class NotifyBase(URLBase):
@ -78,6 +79,74 @@ class NotifyBase(URLBase):
# use a <b> tag. The below causes the <b>title</b> to get generated:
default_html_tag_id = 'b'
# Define a default set of template arguments used for dynamically building
# details about our individual plugins for developers.
# Define object templates
templates = ()
# Provides a mapping of tokens, certain entries are fixed and automatically
# configured if found (such as schema, host, user, pass, and port)
template_tokens = {}
# Here is where we define all of the arguments we accept on the url
# such as: schema://whatever/?overflow=upstream&format=text
# These act the same way as tokens except they are optional and/or
# have default values set if mandatory. This rule must be followed
template_args = {
'overflow': {
'name': _('Overflow Mode'),
'type': 'choice:string',
'values': OVERFLOW_MODES,
# Provide a default
'default': overflow_mode,
# look up default using the following parent class value at
# runtime. The variable name identified here (in this case
# overflow_mode) is checked and it's result is placed over-top of
# the 'default'. This is done because once a parent class inherits
# this one, the overflow_mode already set as a default 'could' be
# potentially over-ridden and changed to a different value.
'_lookup_default': 'overflow_mode',
},
'format': {
'name': _('Notify Format'),
'type': 'choice:string',
'values': NOTIFY_FORMATS,
# Provide a default
'default': notify_format,
# look up default using the following parent class value at
# runtime.
'_lookup_default': 'notify_format',
},
'verify': {
'name': _('Verify SSL'),
# SSL Certificate Authority Verification
'type': 'bool',
# Provide a default
'default': URLBase.verify_certificate,
# look up default using the following parent class value at
# runtime.
'_lookup_default': 'verify_certificate',
},
}
# kwargs are dynamically built because a prefix causes us to parse the
# content slightly differently. The prefix is required and can be either
# a (+ or -). Below would handle the +key=value:
# {
# 'headers': {
# 'name': _('HTTP Header'),
# 'prefix': '+',
# 'type': 'string',
# },
# },
#
# In a kwarg situation, the 'key' is always presumed to be treated as
# a string. When the 'type' is defined, it is being defined to respect
# the 'value'.
template_kwargs = {}
def __init__(self, **kwargs):
"""
Initialize some general configuration that will keep things consistent

View File

@ -41,6 +41,7 @@ from .NotifyBase import NotifyBase
from ..utils import parse_bool
from ..common import NotifyType
from ..common import NotifyImageSize
from ..AppriseLocale import gettext_lazy as _
# Default to sending to all devices if nothing is specified
DEFAULT_TAG = '@all'
@ -92,6 +93,62 @@ class NotifyBoxcar(NotifyBase):
# The maximum allowable characters allowed in the body per message
body_maxlen = 10000
# Define object templates
templates = (
'{schema}://{access_key}/{secret_key}/',
'{schema}://{access_key}/{secret_key}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'access_key': {
'name': _('Access Key'),
'type': 'string',
'regex': (r'[A-Z0-9_-]{64}', 'i'),
'private': True,
'required': True,
'map_to': 'access',
},
'secret_key': {
'name': _('Secret Key'),
'type': 'string',
'regex': (r'[A-Z0-9_-]{64}', 'i'),
'private': True,
'required': True,
'map_to': 'secret',
},
'target_tag': {
'name': _('Target Tag ID'),
'type': 'string',
'prefix': '@',
'regex': (r'[A-Z0-9]{1,63}', 'i'),
'map_to': 'targets',
},
'target_device': {
'name': _('Target Device ID'),
'type': 'string',
'regex': (r'[A-Z0-9]{64}', 'i'),
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, access, secret, targets=None, include_image=True,
**kwargs):
"""

View File

@ -31,6 +31,7 @@ from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import GET_SCHEMA_RE
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# Default our global support flag
NOTIFY_DBUS_SUPPORT_ENABLED = False
@ -171,6 +172,39 @@ class NotifyDBus(NotifyBase):
# 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,
'map_to': 'y_axis',
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
})
def __init__(self, urgency=None, x_axis=None, y_axis=None,
include_image=True, **kwargs):
"""

View File

@ -49,6 +49,7 @@ from ..common import NotifyImageSize
from ..common import NotifyFormat
from ..common import NotifyType
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
class NotifyDiscord(NotifyBase):
@ -77,8 +78,66 @@ class NotifyDiscord(NotifyBase):
# The maximum allowable characters allowed in the body per message
body_maxlen = 2000
# Define object templates
templates = (
'{schema}://{webhook_id}/{webhook_token}',
'{schema}://{botname}@{webhook_id}/{webhook_token}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'botname': {
'name': _('Bot Name'),
'type': 'string',
'map_to': 'user',
},
'webhook_id': {
'name': _('Webhook ID'),
'type': 'string',
'private': True,
'required': True,
},
'webhook_token': {
'name': _('Webhook Token'),
'type': 'string',
'private': True,
'required': True,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'tts': {
'name': _('Text To Speech'),
'type': 'bool',
'default': False,
},
'avatar': {
'name': _('Avatar Image'),
'type': 'bool',
'default': True,
},
'footer': {
'name': _('Display Footer'),
'type': 'bool',
'default': False,
},
'footer_logo': {
'name': _('Footer Logo'),
'type': 'bool',
'default': True,
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': False,
'map_to': 'include_image',
},
})
def __init__(self, webhook_id, webhook_token, tts=False, avatar=True,
footer=False, footer_logo=True, include_image=True, **kwargs):
footer=False, footer_logo=True, include_image=False,
**kwargs):
"""
Initialize Discord Object

View File

@ -24,6 +24,7 @@
# THE SOFTWARE.
import re
import six
import smtplib
from email.mime.text import MIMEText
from socket import error as SocketError
@ -33,6 +34,8 @@ from .NotifyBase import NotifyBase
from ..common import NotifyFormat
from ..common import NotifyType
from ..utils import is_email
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
class WebBaseLogin(object):
@ -242,9 +245,86 @@ class NotifyEmail(NotifyBase):
# Default SMTP Timeout (in seconds)
connect_timeout = 15
def __init__(self, **kwargs):
# Define object templates
templates = (
'{schema}://{user}:{password}@{host}',
'{schema}://{user}:{password}@{host}:{port}',
'{schema}://{user}:{password}@{host}/{targets}',
'{schema}://{user}:{password}@{host}:{port}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'user': {
'name': _('User Name'),
'type': 'string',
'required': True,
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
'required': True,
},
'host': {
'name': _('Domain'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'targets': {
'name': _('Target Emails'),
'type': 'list:string',
},
})
template_args = dict(NotifyBase.template_args, **{
'to': {
'name': _('To Email'),
'type': 'string',
'map_to': 'targets',
},
'from': {
'name': _('From Email'),
'type': 'string',
'map_to': 'from_addr',
},
'name': {
'name': _('From Name'),
'type': 'string',
'map_to': 'from_name',
},
'smtp_host': {
'name': _('SMTP Server'),
'type': 'string',
},
'mode': {
'name': _('Secure Mode'),
'type': 'choice:string',
'values': SECURE_MODES,
'default': SecureMailMode.STARTTLS,
'map_to': 'secure_mode',
},
'timeout': {
'name': _('Server Timeout'),
'type': 'int',
'default': 15,
'min': 5,
},
})
def __init__(self, timeout=15, smtp_host=None, from_name=None,
from_addr=None, secure_mode=None, targets=None, **kwargs):
"""
Initialize Email Object
The smtp_host and secure_mode can be automatically detected depending
on how the URL was built
"""
super(NotifyEmail, self).__init__(**kwargs)
@ -258,33 +338,49 @@ class NotifyEmail(NotifyBase):
# Email SMTP Server Timeout
try:
self.timeout = int(kwargs.get('timeout', self.connect_timeout))
self.timeout = int(timeout)
except (ValueError, TypeError):
self.timeout = self.connect_timeout
# Acquire targets
self.targets = parse_list(targets)
# Now we want to construct the To and From email
# addresses from the URL provided
self.from_name = kwargs.get('name', None)
self.from_addr = kwargs.get('from', None)
self.to_addr = kwargs.get('to', self.from_addr)
self.from_name = from_name
self.from_addr = from_addr
if not self.from_addr:
# detect our email address
self.from_addr = '{}@{}'.format(
re.split(r'[\s@]+', self.user)[0],
self.host,
)
if not is_email(self.from_addr):
# Parse Source domain based on from_addr
raise TypeError('Invalid ~From~ email format: %s' % self.from_addr)
msg = 'Invalid ~From~ email specified: {}'.format(self.from_addr)
self.logger.warning(msg)
raise TypeError(msg)
if not is_email(self.to_addr):
raise TypeError('Invalid ~To~ email format: %s' % self.to_addr)
# If our target email list is empty we want to add ourselves to it
if len(self.targets) == 0:
self.targets.append(self.from_addr)
# Now detect the SMTP Server
self.smtp_host = kwargs.get('smtp_host', '')
self.smtp_host = \
smtp_host if isinstance(smtp_host, six.string_types) else ''
# Now detect secure mode
self.secure_mode = kwargs.get('secure_mode', self.default_secure_mode)
self.secure_mode = self.default_secure_mode \
if not isinstance(secure_mode, six.string_types) \
else secure_mode.lower()
if self.secure_mode not in SECURE_MODES:
raise TypeError(
'Invalid secure mode specified: %s.' % self.secure_mode)
msg = 'The secure mode specified ({}) is invalid.'\
.format(secure_mode)
self.logger.warning(msg)
raise TypeError(msg)
# Apply any defaults based on certain known configurations
self.NotifyEmailDefaults()
@ -305,7 +401,7 @@ class NotifyEmail(NotifyBase):
for i in range(len(EMAIL_TEMPLATES)): # pragma: no branch
self.logger.debug('Scanning %s against %s' % (
self.to_addr, EMAIL_TEMPLATES[i][0]
self.from_addr, EMAIL_TEMPLATES[i][0]
))
match = EMAIL_TEMPLATES[i][1].match(self.from_addr)
if match:
@ -345,7 +441,7 @@ class NotifyEmail(NotifyBase):
elif WebBaseLogin.USERID not in login_type:
# user specified but login type
# not supported; switch it to email
self.user = '%s@%s' % (self.user, self.host)
self.user = '{}@{}'.format(self.user, self.host)
break
@ -358,77 +454,94 @@ class NotifyEmail(NotifyBase):
if not from_name:
from_name = self.app_desc
self.logger.debug('Email From: %s <%s>' % (
self.from_addr, from_name))
self.logger.debug('Email To: %s' % (self.to_addr))
self.logger.debug('Login ID: %s' % (self.user))
self.logger.debug('Delivery: %s:%d' % (self.smtp_host, self.port))
# error tracking (used for function return)
has_error = False
# Prepare Email Message
if self.notify_format == NotifyFormat.HTML:
email = MIMEText(body, 'html')
# Create a copy of the targets list
emails = list(self.targets)
while len(emails):
# Get our email to notify
to_addr = emails.pop(0)
else:
email = MIMEText(body, 'plain')
if not is_email(to_addr):
self.logger.warning(
'Invalid ~To~ email specified: {}'.format(to_addr))
has_error = True
continue
email['Subject'] = title
email['From'] = '%s <%s>' % (from_name, self.from_addr)
email['To'] = self.to_addr
email['Date'] = datetime.utcnow()\
.strftime("%a, %d %b %Y %H:%M:%S +0000")
email['X-Application'] = self.app_id
self.logger.debug(
'Email From: {} <{}>'.format(from_name, self.from_addr))
self.logger.debug('Email To: {}'.format(to_addr))
self.logger.debug('Login ID: {}'.format(self.user))
self.logger.debug(
'Delivery: {}:{}'.format(self.smtp_host, self.port))
# bind the socket variable to the current namespace
socket = None
# Prepare Email Message
if self.notify_format == NotifyFormat.HTML:
email = MIMEText(body, 'html')
# Always call throttle before any remote server i/o is made
self.throttle()
else:
email = MIMEText(body, 'plain')
try:
self.logger.debug('Connecting to remote SMTP server...')
socket_func = smtplib.SMTP
if self.secure and self.secure_mode == SecureMailMode.SSL:
self.logger.debug('Securing connection with SSL...')
socket_func = smtplib.SMTP_SSL
email['Subject'] = title
email['From'] = '{} <{}>'.format(from_name, self.from_addr)
email['To'] = to_addr
email['Date'] = \
datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S +0000")
email['X-Application'] = self.app_id
socket = socket_func(
self.smtp_host,
self.port,
None,
timeout=self.timeout,
)
# bind the socket variable to the current namespace
socket = None
if self.secure and self.secure_mode == SecureMailMode.STARTTLS:
# Handle Secure Connections
self.logger.debug('Securing connection with STARTTLS...')
socket.starttls()
# Always call throttle before any remote server i/o is made
self.throttle()
if self.user and self.password:
# Apply Login credetials
self.logger.debug('Applying user credentials...')
socket.login(self.user, self.password)
try:
self.logger.debug('Connecting to remote SMTP server...')
socket_func = smtplib.SMTP
if self.secure and self.secure_mode == SecureMailMode.SSL:
self.logger.debug('Securing connection with SSL...')
socket_func = smtplib.SMTP_SSL
# Send the email
socket.sendmail(self.from_addr, self.to_addr, email.as_string())
socket = socket_func(
self.smtp_host,
self.port,
None,
timeout=self.timeout,
)
self.logger.info('Sent Email notification to "%s".' % (
self.to_addr,
))
if self.secure and self.secure_mode == SecureMailMode.STARTTLS:
# Handle Secure Connections
self.logger.debug('Securing connection with STARTTLS...')
socket.starttls()
except (SocketError, smtplib.SMTPException, RuntimeError) as e:
self.logger.warning(
'A Connection error occured sending Email '
'notification to %s.' % self.smtp_host)
self.logger.debug('Socket Exception: %s' % str(e))
# Return; we're done
return False
if self.user and self.password:
# Apply Login credetials
self.logger.debug('Applying user credentials...')
socket.login(self.user, self.password)
finally:
# Gracefully terminate the connection with the server
if socket is not None: # pragma: no branch
socket.quit()
# Send the email
socket.sendmail(
self.from_addr, to_addr, email.as_string())
return True
self.logger.info(
'Sent Email notification to "{}".'.format(to_addr))
except (SocketError, smtplib.SMTPException, RuntimeError) as e:
self.logger.warning(
'A Connection error occured sending Email '
'notification to {}.'.format(self.smtp_host))
self.logger.debug('Socket Exception: %s' % str(e))
# Mark our failure
has_error = True
finally:
# Gracefully terminate the connection with the server
if socket is not None: # pragma: no branch
socket.quit()
return not has_error
def url(self):
"""
@ -439,7 +552,6 @@ class NotifyEmail(NotifyBase):
args = {
'format': self.notify_format,
'overflow': self.overflow_mode,
'to': self.to_addr,
'from': self.from_addr,
'name': self.from_name,
'mode': self.secure_mode,
@ -469,12 +581,19 @@ class NotifyEmail(NotifyBase):
default_port = \
self.default_secure_port if self.secure else self.default_port
return '{schema}://{auth}{hostname}{port}/?{args}'.format(
# a simple boolean check as to whether we display our target emails
# or not
has_targets = \
not (len(self.targets) == 1 and self.targets[0] == self.from_addr)
return '{schema}://{auth}{hostname}{port}/{targets}?{args}'.format(
schema=self.secure_protocol if self.secure else self.protocol,
auth=auth,
hostname=NotifyEmail.quote(self.host, safe=''),
port='' if self.port is None or self.port == default_port
else ':{}'.format(self.port),
targets='' if has_targets else '/'.join(
[NotifyEmail.quote(x, safe='') for x in self.targets]),
args=NotifyEmail.urlencode(args),
)
@ -491,48 +610,30 @@ class NotifyEmail(NotifyBase):
# We're done early as we couldn't load the results
return results
# The To: address is pre-determined if to= is not otherwise
# specified.
to_addr = ''
# The From address is a must; either through the use of templates
# from= entry and/or merging the user and hostname together, this
# must be calculated or parse_url will fail. The to_addr will
# become the from_addr if it can't be calculated
# must be calculated or parse_url will fail.
from_addr = ''
# The server we connect to to send our mail to
smtp_host = ''
# Get our potential email targets; if none our found we'll just
# add one to ourselves
results['targets'] = NotifyEmail.split_path(results['fullpath'])
# Attempt to detect 'from' email address
if 'from' in results['qsd'] and len(results['qsd']['from']):
from_addr = NotifyEmail.unquote(results['qsd']['from'])
else:
# get 'To' email address
from_addr = '%s@%s' % (
re.split(
r'[\s@]+', NotifyEmail.unquote(results['user']))[0],
results.get('host', '')
)
# Lets be clever and attempt to make the from
# address an email based on the to address
from_addr = '%s@%s' % (
re.split(r'[\s@]+', from_addr)[0],
re.split(r'[\s@]+', from_addr)[-1],
)
# Attempt to detect 'to' email address
if 'to' in results['qsd'] and len(results['qsd']['to']):
to_addr = NotifyEmail.unquote(results['qsd']['to']).strip()
if not to_addr:
# Send to ourselves if not otherwise specified to do so
to_addr = from_addr
results['targets'] += \
NotifyEmail.parse_list(results['qsd']['to'])
if 'name' in results['qsd'] and len(results['qsd']['name']):
# Extract from name to associate with from address
results['name'] = NotifyEmail.unquote(results['qsd']['name'])
results['from_name'] = NotifyEmail.unquote(results['qsd']['name'])
if 'timeout' in results['qsd'] and len(results['qsd']['timeout']):
# Extract the timeout to associate with smtp server
@ -547,8 +648,7 @@ class NotifyEmail(NotifyBase):
# Extract the secure mode to over-ride the default
results['secure_mode'] = results['qsd']['mode'].lower()
results['to'] = to_addr
results['from'] = from_addr
results['from_addr'] = from_addr
results['smtp_host'] = smtp_host
return results

View File

@ -38,6 +38,7 @@ from .NotifyBase import NotifyBase
from ..utils import parse_bool
from ..common import NotifyType
from .. import __version__ as VERSION
from ..AppriseLocale import gettext_lazy as _
class NotifyEmby(NotifyBase):
@ -72,6 +73,46 @@ class NotifyEmby(NotifyBase):
# displayed for. The value is in milli-seconds
emby_message_timeout_ms = 60000
# Define object templates
templates = (
'{schema}://{host}',
'{schema}://{host}:{port}',
'{schema}://{user}:{password}@{host}',
'{schema}://{user}:{password}@{host}:{port}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
})
template_args = dict(NotifyBase.template_args, **{
'modal': {
'name': _('Modal'),
'type': 'bool',
'default': False,
},
})
def __init__(self, modal=False, **kwargs):
"""
Initialize Emby Object

View File

@ -28,6 +28,7 @@ from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
class NotifyFaast(NotifyBase):
@ -53,6 +54,31 @@ class NotifyFaast(NotifyBase):
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_72
# Define object templates
templates = (
'{schema}://{authtoken}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'authtoken': {
'name': _('Authorization Token'),
'type': 'string',
'private': True,
'required': True,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
})
def __init__(self, authtoken, include_image=True, **kwargs):
"""
Initialize Faast Object

View File

@ -47,6 +47,7 @@ from ..common import NotifyFormat
from ..common import NotifyImageSize
from ..utils import parse_list
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# Extend HTTP Error Messages
@ -92,6 +93,60 @@ class NotifyFlock(NotifyBase):
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_72
# Define object templates
templates = (
'{schema}://{token}',
'{schema}://{user}@{token}',
'{schema}://{user}@{token}/{targets}',
'{schema}://{token}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'token': {
'name': _('Access Key'),
'type': 'string',
'regex': (r'[a-z0-9-]{24}', 'i'),
'private': True,
'required': True,
},
'user': {
'name': _('Bot Name'),
'type': 'string',
},
'to_user': {
'name': _('To User ID'),
'type': 'string',
'prefix': '@',
'regex': (r'[A-Z0-9_]{12}', 'i'),
'map_to': 'targets',
},
'to_channel': {
'name': _('To Channel ID'),
'type': 'string',
'prefix': '#',
'regex': (r'[A-Z0-9_]{12}', 'i'),
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, token, targets=None, include_image=True, **kwargs):
"""
Initialize Flock Object

View File

@ -50,7 +50,7 @@ from ..common import NotifyFormat
from ..common import NotifyType
from ..utils import parse_list
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# API Gitter URL
GITTER_API_URL = 'https://api.gitter.im/v1'
@ -102,7 +102,40 @@ class NotifyGitter(NotifyBase):
# Default Notification Format
notify_format = NotifyFormat.MARKDOWN
def __init__(self, token, targets, include_image=True, **kwargs):
# Define object templates
templates = (
'{schema}://{token}:{targets}/',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'token': {
'name': _('Token'),
'type': 'string',
'regex': (r'[a-z0-9]{40}', 'i'),
'private': True,
'required': True,
},
'targets': {
'name': _('Rooms'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': False,
'map_to': 'include_image',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, token, targets, include_image=False, **kwargs):
"""
Initialize Gitter Object
"""

View File

@ -30,6 +30,7 @@ from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# Default our global support flag
NOTIFY_GNOME_SUPPORT_ENABLED = False
@ -110,6 +111,27 @@ class NotifyGnome(NotifyBase):
# let me know! :)
_enabled = NOTIFY_GNOME_SUPPORT_ENABLED
# Define object templates
templates = (
'{schema}://_/',
)
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'urgency': {
'name': _('Urgency'),
'type': 'choice:int',
'values': GNOME_URGENCIES,
'default': GnomeUrgency.NORMAL,
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
})
def __init__(self, urgency=None, include_image=True, **kwargs):
"""
Initialize Gnome Object

View File

@ -37,6 +37,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..AppriseLocale import gettext_lazy as _
# Priorities
@ -76,6 +77,43 @@ class NotifyGotify(NotifyBase):
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_gotify'
# Define object templates
templates = (
'{schema}://{host}/{token}',
'{schema}://{host}:{port}/{token}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'token': {
'name': _('Token'),
'type': 'string',
'private': True,
'required': True,
},
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'priority': {
'name': _('Priority'),
'type': 'choice:int',
'values': GOTIFY_PRIORITIES,
'default': GotifyPriority.NORMAL,
},
})
def __init__(self, token, priority=None, **kwargs):
"""
Initialize Gotify Object

View File

@ -29,6 +29,7 @@ from ..NotifyBase import NotifyBase
from ...common import NotifyImageSize
from ...common import NotifyType
from ...utils import parse_bool
from ...AppriseLocale import gettext_lazy as _
# Priorities
@ -87,6 +88,51 @@ class NotifyGrowl(NotifyBase):
# Default Growl Port
default_port = 23053
# Define object templates
templates = (
'{schema}://{apikey}',
'{schema}://{apikey}/{providerkey}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'apikey': {
'name': _('API Key'),
'type': 'string',
'private': True,
'required': True,
'map_to': 'host',
},
'providerkey': {
'name': _('Provider Key'),
'type': 'string',
'private': True,
'map_to': 'fullpath',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'priority': {
'name': _('Priority'),
'type': 'choice:int',
'values': GROWL_PRIORITIES,
'default': GrowlPriority.NORMAL,
},
'version': {
'name': _('Version'),
'type': 'choice:int',
'values': (1, 2),
'default': 2,
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
})
def __init__(self, priority=None, version=2, include_image=True, **kwargs):
"""
Initialize Growl Object

View File

@ -45,6 +45,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
class NotifyIFTTT(NotifyBase):
@ -91,6 +92,45 @@ class NotifyIFTTT(NotifyBase):
notify_url = 'https://maker.ifttt.com/' \
'trigger/{event}/with/key/{webhook_id}'
# Define object templates
templates = (
'{schema}://{webhook_id}/{events}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'webhook_id': {
'name': _('Webhook ID'),
'type': 'string',
'private': True,
'required': True,
},
'events': {
'name': _('Events'),
'type': 'list:string',
'required': True,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'events',
},
})
# Define our token control
template_kwargs = {
'add_tokens': {
'name': _('Add Tokens'),
'prefix': '+',
},
'del_tokens': {
'name': _('Remove Tokens'),
'prefix': '-',
},
}
def __init__(self, webhook_id, events, add_tokens=None, del_tokens=None,
**kwargs):
"""
@ -134,6 +174,10 @@ class NotifyIFTTT(NotifyBase):
if isinstance(del_tokens, (list, tuple, set)):
self.del_tokens = del_tokens
elif isinstance(del_tokens, dict):
# Convert the dictionary into a list
self.del_tokens = set(del_tokens.keys())
else:
msg = 'del_token must be a list; {} was provided'.format(
str(type(del_tokens)))

View File

@ -30,6 +30,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
from ..AppriseLocale import gettext_lazy as _
class NotifyJSON(NotifyBase):
@ -56,6 +57,48 @@ class NotifyJSON(NotifyBase):
# local anyway
request_rate_per_sec = 0
# Define object templates
templates = (
'{schema}://{host}',
'{schema}://{host}:{port}',
'{schema}://{user}@{host}',
'{schema}://{user}@{host}:{port}',
'{schema}://{user}:{password}@{host}',
'{schema}://{user}:{password}@{host}:{port}',
)
# Define our tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
})
# Define any kwargs we're using
template_kwargs = {
'headers': {
'name': _('HTTP Header'),
'prefix': '+',
},
}
def __init__(self, headers=None, **kwargs):
"""
Initialize JSON Object

View File

@ -41,9 +41,10 @@ from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import parse_list
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# Token required as part of the API request
VALIDATE_APIKEY = re.compile(r'[A-Za-z0-9]{32}')
VALIDATE_APIKEY = re.compile(r'[a-z0-9]{32}', re.I)
# Extend HTTP Error Messages
JOIN_HTTP_ERROR_MAP = {
@ -51,7 +52,7 @@ JOIN_HTTP_ERROR_MAP = {
}
# Used to detect a device
IS_DEVICE_RE = re.compile(r'([A-Za-z0-9]{32})')
IS_DEVICE_RE = re.compile(r'([a-z0-9]{32})', re.I)
# Used to detect a device
IS_GROUP_RE = re.compile(
@ -97,6 +98,53 @@ class NotifyJoin(NotifyBase):
# The default group to use if none is specified
default_join_group = 'group.all'
# Define object templates
templates = (
'{schema}://{apikey}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'apikey': {
'name': _('API Key'),
'type': 'string',
'regex': (r'[a-z0-9]{32}', 'i'),
'private': True,
'required': True,
},
'device': {
'name': _('Device ID'),
'type': 'string',
'regex': (r'[a-z0-9]{32}', 'i'),
'map_to': 'targets',
},
'group': {
'name': _('Group'),
'type': 'choice:string',
'values': (
'all', 'android', 'chrome', 'windows10', 'phone', 'tablet',
'pc'),
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
'required': True,
},
})
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': False,
'map_to': 'include_image',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, apikey, targets, include_image=True, **kwargs):
"""
Initialize Join Object

View File

@ -69,6 +69,7 @@ from ..common import NotifyImageSize
from ..common import NotifyType
from ..common import NotifyFormat
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# Used to prepare our UUID regex matching
UUID4_RE = \
@ -114,8 +115,49 @@ class NotifyMSTeams(NotifyBase):
# The maximum allowable characters allowed in the body per message
body_maxlen = 1000
# Default Notification Format
notify_format = NotifyFormat.MARKDOWN
# Define object templates
templates = (
'{schema}://{token_a}/{token_b}{token_c}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'token_a': {
'name': _('Token A'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'{}@{}'.format(UUID4_RE, UUID4_RE), 'i'),
},
'token_b': {
'name': _('Token B'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[a-z0-9]{32}', 'i'),
},
'token_c': {
'name': _('Token C'),
'type': 'string',
'private': True,
'required': True,
'regex': (UUID4_RE, 'i'),
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': False,
'map_to': 'include_image',
},
})
def __init__(self, token_a, token_b, token_c, include_image=True,
**kwargs):
"""

View File

@ -58,7 +58,7 @@ from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..utils import is_email
from ..AppriseLocale import gettext_lazy as _
# Used to validate your personal access apikey
VALIDATE_API_KEY = re.compile(r'^[a-z0-9]{32}-[a-z0-9]{8}-[a-z0-9]{8}$', re.I)
@ -117,6 +117,56 @@ class NotifyMailgun(NotifyBase):
# The default region to use if one isn't otherwise specified
mailgun_default_region = MailgunRegion.US
# Define object templates
templates = (
'{schema}://{user}@{host}:{apikey}/',
'{schema}://{user}@{host}:{apikey}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'user': {
'name': _('User Name'),
'type': 'string',
'required': True,
},
'host': {
'name': _('Domain'),
'type': 'string',
'required': True,
},
'apikey': {
'name': _('API Key'),
'type': 'string',
'regex': (r'[a-z0-9]{32}-[a-z0-9]{8}-[a-z0-9]{8}', 'i'),
'private': True,
'required': True,
},
'targets': {
'name': _('Target Emails'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'name': {
'name': _('From Name'),
'type': 'string',
'map_to': 'from_name',
},
'region': {
'name': _('Region Name'),
'type': 'choice:string',
'values': MAILGUN_REGIONS,
'default': MailgunRegion.US,
'map_to': 'region_name',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, apikey, targets, from_name=None, region_name=None,
**kwargs):
"""

View File

@ -40,6 +40,7 @@ from ..common import NotifyImageSize
from ..common import NotifyFormat
from ..utils import parse_bool
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Define default path
MATRIX_V2_API_PATH = '/_matrix/client/r0'
@ -66,6 +67,9 @@ SLACK_DEFAULT_USER = 'apprise'
class MatrixWebhookMode(object):
# Webhook Mode is disabled
DISABLED = "off"
# The default webhook mode is to just be set to Matrix
MATRIX = "matrix"
@ -75,6 +79,7 @@ class MatrixWebhookMode(object):
# webhook modes are placed ito this list for validation purposes
MATRIX_WEBHOOK_MODES = (
MatrixWebhookMode.DISABLED,
MatrixWebhookMode.MATRIX,
MatrixWebhookMode.SLACK,
)
@ -117,7 +122,86 @@ class NotifyMatrix(NotifyBase):
# the server doesn't remind us how long we shoul wait for
default_wait_ms = 1000
def __init__(self, targets=None, mode=None, include_image=True,
# Define object templates
templates = (
'{schema}://{user}:{password}@{host}/{targets}',
'{schema}://{user}:{password}@{host}:{port}/{targets}',
'{schema}://{token}:{password}@{host}/{targets}',
'{schema}://{token}:{password}@{host}:{port}/{targets}',
'{schema}://{user}:{token}:{password}@{host}/{targets}',
'{schema}://{user}:{token}:{password}@{host}:{port}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
'token': {
'name': _('Access Token'),
'map_to': 'password',
},
'target_user': {
'name': _('Target User'),
'type': 'string',
'prefix': '@',
'map_to': 'targets',
},
'target_room_id': {
'name': _('Target Room ID'),
'type': 'string',
'prefix': '!',
'map_to': 'targets',
},
'target_room_alias': {
'name': _('Target Room Alias'),
'type': 'string',
'prefix': '!',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': False,
'map_to': 'include_image',
},
'mode': {
'name': _('Webhook Mode'),
'type': 'choice:string',
'values': MATRIX_WEBHOOK_MODES,
'default': MatrixWebhookMode.DISABLED,
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, targets=None, mode=None, include_image=False,
**kwargs):
"""
Initialize Matrix Object
@ -144,7 +228,7 @@ class NotifyMatrix(NotifyBase):
self._room_cache = {}
# Setup our mode
self.mode = None \
self.mode = MatrixWebhookMode.DISABLED \
if not isinstance(mode, six.string_types) else mode.lower()
if self.mode and self.mode not in MATRIX_WEBHOOK_MODES:
msg = 'The mode specified ({}) is invalid.'.format(mode)
@ -160,7 +244,8 @@ class NotifyMatrix(NotifyBase):
# - calls _send_webhook_notification if the mode variable is set
# - calls _send_server_notification if the mode variable is not set
return getattr(self, '_send_{}_notification'.format(
'webhook' if self.mode else 'server'))(
'webhook' if self.mode != MatrixWebhookMode.DISABLED
else 'server'))(
body=body, title=title, notify_type=notify_type, **kwargs)
def _send_webhook_notification(self, body, title='',
@ -875,11 +960,9 @@ class NotifyMatrix(NotifyBase):
'overflow': self.overflow_mode,
'image': 'yes' if self.include_image else 'no',
'verify': 'yes' if self.verify_certificate else 'no',
'mode': self.mode,
}
if self.mode:
args['mode'] = self.mode
# Determine Authentication
auth = ''
if self.user and self.password:

View File

@ -32,13 +32,14 @@ from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import parse_bool
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Some Reference Locations:
# - https://docs.mattermost.com/developer/webhooks-incoming.html
# - https://docs.mattermost.com/administration/config-settings.html
# Used to validate Authorization Token
VALIDATE_AUTHTOKEN = re.compile(r'[A-Za-z0-9]{24,32}')
VALIDATE_AUTHTOKEN = re.compile(r'[a-z0-9]{24,32}', re.I)
class NotifyMatterMost(NotifyBase):
@ -73,7 +74,59 @@ class NotifyMatterMost(NotifyBase):
# Mattermost does not have a title
title_maxlen = 0
def __init__(self, authtoken, channels=None, include_image=True,
# Define object templates
templates = (
'{schema}://{host}/{authtoken}',
'{schema}://{host}/{authtoken}:{port}',
'{schema}://{botname}@{host}/{authtoken}',
'{schema}://{botname}@{host}/{authtoken}:{port}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'authtoken': {
'name': _('Access Key'),
'type': 'string',
'regex': (r'[a-z0-9]{24,32}', 'i'),
'private': True,
'required': True,
},
'botname': {
'name': _('Bot Name'),
'type': 'string',
'map_to': 'user',
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'channels': {
'name': _('Channels'),
'type': 'list:string',
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
'to': {
'alias_of': 'channels',
},
})
def __init__(self, authtoken, channels=None, include_image=False,
**kwargs):
"""
Initialize MatterMost Object
@ -86,7 +139,7 @@ class NotifyMatterMost(NotifyBase):
else:
self.schema = 'http'
# Our API Key
# Our Authorization Token
self.authtoken = authtoken
# Validate authtoken

View File

@ -28,6 +28,7 @@ import requests
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..AppriseLocale import gettext_lazy as _
# Used to validate API Key
VALIDATE_APIKEY = re.compile(r'[A-Za-z0-9]{40}')
@ -90,6 +91,37 @@ class NotifyProwl(NotifyBase):
# Defines the maximum allowable characters in the title
title_maxlen = 1024
# Define object templates
templates = (
'{schema}://{apikey}',
'{schema}://{apikey}/{providerkey}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'apikey': {
'name': _('API Key'),
'type': 'string',
'private': True,
'required': True,
},
'providerkey': {
'name': _('Provider Key'),
'type': 'string',
'private': True,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'priority': {
'name': _('Priority'),
'type': 'choice:int',
'values': PROWL_PRIORITIES,
'default': ProwlPriority.NORMAL,
},
})
def __init__(self, apikey, providerkey=None, priority=None, **kwargs):
"""
Initialize Prowl Object

View File

@ -30,6 +30,7 @@ from .NotifyBase import NotifyBase
from ..utils import GET_EMAIL_RE
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Flag used as a placeholder to sending to all devices
PUSHBULLET_SEND_TO_ALL = 'ALL_DEVICES'
@ -60,6 +61,49 @@ class NotifyPushBullet(NotifyBase):
# PushBullet uses the http protocol with JSON requests
notify_url = 'https://api.pushbullet.com/v2/pushes'
# Define object templates
templates = (
'{schema}://{accesstoken}',
'{schema}://{accesstoken}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'accesstoken': {
'name': _('Access Token'),
'type': 'string',
'private': True,
'required': True,
},
'target_device': {
'name': _('Target Device'),
'type': 'string',
'map_to': 'targets',
},
'target_channel': {
'name': _('Target Channel'),
'type': 'string',
'prefix': '#',
'map_to': 'targets',
},
'target_email': {
'name': _('Target Email'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
})
def __init__(self, accesstoken, targets=None, **kwargs):
"""
Initialize PushBullet Object

View File

@ -31,6 +31,7 @@ from itertools import chain
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Used to detect and parse channels
IS_CHANNEL = re.compile(r'^#(?P<name>[A-Za-z0-9]+)$')
@ -67,6 +68,51 @@ class NotifyPushed(NotifyBase):
# The maximum allowable characters allowed in the body per message
body_maxlen = 140
# Define object templates
templates = (
'{schema}://{app_key}/{app_secret}',
'{schema}://{app_key}/{app_secret}@{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'app_key': {
'name': _('Application Key'),
'type': 'string',
'private': True,
'required': True,
},
'app_secret': {
'name': _('Application Secret'),
'type': 'string',
'private': True,
'required': True,
},
'target_user': {
'name': _('Target User'),
'prefix': '@',
'type': 'string',
'map_to': 'targets',
},
'target_channel': {
'name': _('Target Channel'),
'type': 'string',
'prefix': '#',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
})
def __init__(self, app_key, app_secret, targets=None, **kwargs):
"""
Initialize Pushed Object

View File

@ -28,6 +28,7 @@ from . import pushjet
from ..NotifyBase import NotifyBase
from ...common import NotifyType
from ...AppriseLocale import gettext_lazy as _
PUBLIC_KEY_RE = re.compile(
r'^[a-z0-9]{4}-[a-z0-9]{6}-[a-z0-9]{12}-[a-z0-9]{5}-[a-z0-9]{9}$', re.I)
@ -56,6 +57,33 @@ class NotifyPushjet(NotifyBase):
# local anyway (the remote/online service is no more)
request_rate_per_sec = 0
# Define object templates
templates = (
'{schema}://{secret_key}@{host}',
'{schema}://{secret_key}@{host}:{port}',
)
# Define our tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'secret_key': {
'name': _('Secret Key'),
'type': 'string',
'required': True,
'private': True,
},
})
def __init__(self, secret_key, **kwargs):
"""
Initialize Pushjet Object

View File

@ -30,6 +30,7 @@ import requests
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Flag used as a placeholder to sending to all devices
PUSHOVER_SEND_TO_ALL = 'ALL_DEVICES'
@ -38,7 +39,7 @@ PUSHOVER_SEND_TO_ALL = 'ALL_DEVICES'
VALIDATE_TOKEN = re.compile(r'^[a-z0-9]{30}$', re.I)
# Used to detect a User and/or Group
VALIDATE_USERGROUP = re.compile(r'^[a-z0-9]{30}$', re.I)
VALIDATE_USER_KEY = re.compile(r'^[a-z0-9]{30}$', re.I)
# Used to detect a User and/or Group
VALIDATE_DEVICE = re.compile(r'^[a-z0-9_]{1,25}$', re.I)
@ -144,6 +145,60 @@ class NotifyPushover(NotifyBase):
# Default Pushover sound
default_pushover_sound = PushoverSound.PUSHOVER
# Define object templates
templates = (
'{schema}://{user_key}@{token}',
'{schema}://{user_key}@{token}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'user_key': {
'name': _('User Key'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[a-z0-9]{30}', 'i'),
'map_to': 'user',
},
'token': {
'name': _('Access Token'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[a-z0-9]{30}', 'i'),
},
'target_device': {
'name': _('Target Device'),
'type': 'string',
'regex': (r'[a-z0-9_]{1,25}', 'i'),
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'priority': {
'name': _('Priority'),
'type': 'choice:int',
'values': PUSHOVER_PRIORITIES,
'default': PushoverPriority.NORMAL,
},
'sound': {
'name': _('Sound'),
'type': 'string',
'regex': (r'[a-z]{1,12}', 'i'),
'default': PushoverSound.PUSHOVER,
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, token, targets=None, priority=None, sound=None,
**kwargs):
"""
@ -186,12 +241,12 @@ class NotifyPushover(NotifyBase):
self.priority = priority
if not self.user:
msg = 'No user was specified.'
msg = 'No user key was specified.'
self.logger.warning(msg)
raise TypeError(msg)
if not VALIDATE_USERGROUP.match(self.user):
msg = 'The user/group specified (%s) is invalid.' % self.user
if not VALIDATE_USER_KEY.match(self.user):
msg = 'The user key specified (%s) is invalid.' % self.user
self.logger.warning(msg)
raise TypeError(msg)

View File

@ -36,6 +36,7 @@ from ..common import NotifyFormat
from ..common import NotifyType
from ..utils import parse_list
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
IS_CHANNEL = re.compile(r'^#(?P<name>[A-Za-z0-9_-]+)$')
IS_USER = re.compile(r'^@(?P<name>[A-Za-z0-9._-]+)$')
@ -103,8 +104,84 @@ class NotifyRocketChat(NotifyBase):
# Default to markdown
notify_format = NotifyFormat.MARKDOWN
def __init__(self, webhook=None, targets=None, mode=None,
include_avatar=True, **kwargs):
# Define object templates
templates = (
'{schema}://{user}:{password}@{host}:{port}/{targets}',
'{schema}://{user}:{password}@{host}/{targets}',
'{schema}://{webhook}@{host}',
'{schema}://{webhook}@{host}:{port}',
'{schema}://{webhook}@{host}/{targets}',
'{schema}://{webhook}@{host}:{port}/{targets}',
)
# Define our template arguments
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
'webhook': {
'name': _('Webhook'),
'type': 'string',
},
'target_channel': {
'name': _('Target Channel'),
'type': 'string',
'prefix': '#',
'map_to': 'targets',
},
'target_user': {
'name': _('Target User'),
'type': 'string',
'prefix': '@',
'map_to': 'targets',
},
'target_room': {
'name': _('Target Room ID'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'mode': {
'name': _('Webhook Mode'),
'type': 'choice:string',
'values': ROCKETCHAT_AUTH_MODES,
},
'avatar': {
'name': _('Use Avatar'),
'type': 'bool',
'default': True,
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, webhook=None, targets=None, mode=None, avatar=True,
**kwargs):
"""
Initialize Notify Rocket.Chat Object
"""
@ -132,7 +209,7 @@ class NotifyRocketChat(NotifyBase):
self.webhook = webhook
# Place an avatar image to associate with our content
self.include_avatar = include_avatar
self.avatar = avatar
# Used to track token headers upon authentication (if successful)
# This is only used if not on webhook mode
@ -212,7 +289,7 @@ class NotifyRocketChat(NotifyBase):
'format': self.notify_format,
'overflow': self.overflow_mode,
'verify': 'yes' if self.verify_certificate else 'no',
'avatar': 'yes' if self.include_avatar else 'no',
'avatar': 'yes' if self.avatar else 'no',
'mode': self.mode,
}
@ -371,7 +448,7 @@ class NotifyRocketChat(NotifyBase):
# apply our images if they're set to be displayed
image_url = self.image_url(notify_type)
if self.include_avatar:
if self.avatar:
payload['avatar'] = image_url
return payload
@ -599,7 +676,7 @@ class NotifyRocketChat(NotifyBase):
NotifyRocketChat.unquote(results['qsd']['mode'])
# avatar icon
results['include_avatar'] = \
results['avatar'] = \
parse_bool(results['qsd'].get('avatar', True))
# The 'to' makes it easier to use yaml configuration

View File

@ -40,6 +40,7 @@ from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# Token required as part of the API request
VALIDATE_TOKEN = re.compile(r'[A-Za-z0-9]{15}')
@ -86,6 +87,47 @@ class NotifyRyver(NotifyBase):
# The maximum allowable characters allowed in the body per message
body_maxlen = 1000
# Define object templates
templates = (
'{schema}://{organization}/{token}',
'{schema}://{user}@{organization}/{token}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'organization': {
'name': _('Organization'),
'type': 'string',
'required': True,
},
'token': {
'name': _('Token'),
'type': 'string',
'required': True,
'private': True,
},
'user': {
'name': _('Bot Name'),
'type': 'string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'mode': {
'name': _('Webhook Mode'),
'type': 'choice:string',
'values': RYVER_WEBHOOK_MODES,
'default': RyverWebhookMode.RYVER,
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
})
def __init__(self, organization, token, mode=RyverWebhookMode.RYVER,
include_image=True, **kwargs):
"""

View File

@ -35,6 +35,7 @@ from itertools import chain
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Some Phone Number Detection
IS_PHONE_NO = re.compile(r'^\+?(?P<phone>[0-9\s)(+-]+)\s*$')
@ -92,6 +93,58 @@ class NotifySNS(NotifyBase):
# cause any title (if defined) to get placed into the message body.
title_maxlen = 0
# Define object templates
templates = (
'{schema}://{access_key_id}/{secret_access_key}{region}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'access_key_id': {
'name': _('Access Key ID'),
'type': 'string',
'private': True,
'required': True,
},
'secret_access_key': {
'name': _('Secret Access Key'),
'type': 'string',
'private': True,
'required': True,
},
'region': {
'name': _('Region'),
'type': 'string',
'required': True,
'regex': (r'[a-z]{2}-[a-z]+-[0-9]+', 'i'),
'map_to': 'region_name',
},
'target_phone_no': {
'name': _('Target Phone No'),
'type': 'string',
'map_to': 'targets',
'regex': (r'[0-9\s)(+-]+', 'i')
},
'target_topic': {
'name': _('Target Topic'),
'type': 'string',
'map_to': 'targets',
'prefix': '#',
'regex': (r'[A-Za-z0-9_-]+', 'i'),
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
})
def __init__(self, access_key_id, secret_access_key, region_name,
targets=None, **kwargs):
"""

View File

@ -46,6 +46,7 @@ from ..common import NotifyType
from ..common import NotifyFormat
from ..utils import parse_bool
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Token required as part of the API request
# /AAAAAAAAA/........./........................
@ -71,7 +72,7 @@ SLACK_HTTP_ERROR_MAP = {
CHANNEL_LIST_DELIM = re.compile(r'[ \t\r\n,#\\/]+')
# Used to detect a channel
IS_CHANNEL_RE = re.compile(r'[+#@]?([A-Z0-9_]{1,32})', re.I)
IS_VALID_TARGET_RE = re.compile(r'[+#@]?([A-Z0-9_]{1,32})', re.I)
class NotifySlack(NotifyBase):
@ -100,8 +101,82 @@ class NotifySlack(NotifyBase):
# The maximum allowable characters allowed in the body per message
body_maxlen = 1000
# Default Notification Format
notify_format = NotifyFormat.MARKDOWN
# Define object templates
templates = (
'{schema}://{token_a}/{token_b}{token_c}',
'{schema}://{botname}@{token_a}/{token_b}{token_c}',
'{schema}://{token_a}/{token_b}{token_c}/{targets}',
'{schema}://{botname}@{token_a}/{token_b}{token_c}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'botname': {
'name': _('Bot Name'),
'type': 'string',
'map_to': 'user',
},
'token_a': {
'name': _('Token A'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[A-Z0-9]{9}', 'i'),
},
'token_b': {
'name': _('Token B'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[A-Z0-9]{9}', 'i'),
},
'token_c': {
'name': _('Token C'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[A-Za-z0-9]{24}', 'i'),
},
'target_encoded_id': {
'name': _('Target Encoded ID'),
'type': 'string',
'prefix': '+',
'map_to': 'targets',
},
'target_user': {
'name': _('Target User'),
'type': 'string',
'prefix': '@',
'map_to': 'targets',
},
'target_channels': {
'name': _('Target Channel'),
'type': 'string',
'prefix': '#',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, token_a, token_b, token_c, targets,
include_image=True, **kwargs):
"""
@ -232,7 +307,7 @@ class NotifySlack(NotifyBase):
if channel is not None:
# Channel over-ride was specified
if not IS_CHANNEL_RE.match(channel):
if not IS_VALID_TARGET_RE.match(channel):
self.logger.warning(
"The specified target {} is invalid;"
"skipping.".format(channel))

View File

@ -63,6 +63,7 @@ from ..common import NotifyImageSize
from ..common import NotifyFormat
from ..utils import parse_bool
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
TELEGRAM_IMAGE_XY = NotifyImageSize.XY_256
@ -107,8 +108,55 @@ class NotifyTelegram(NotifyBase):
# The maximum allowable characters allowed in the body per message
body_maxlen = 4096
def __init__(self, bot_token, targets, detect_bot_owner=True,
include_image=True, **kwargs):
# Define object templates
templates = (
'{schema}://{bot_token}',
'{schema}://{bot_token}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'bot_token': {
'name': _('Bot Token'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'(bot)?[0-9]+:[a-z0-9_-]+', 'i'),
},
'target_user': {
'name': _('Target Chat ID'),
'type': 'string',
'map_to': 'targets',
'map_to': 'targets',
'regex': (r'((-?[0-9]{1,32})|([a-z_-][a-z0-9_-]+))', 'i'),
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': False,
'map_to': 'include_image',
},
'detect': {
'name': _('Detect Bot Owner'),
'type': 'bool',
'default': True,
'map_to': 'detect_owner',
},
'to': {
'alias_of': 'targets',
},
})
def __init__(self, bot_token, targets, detect_owner=True,
include_image=False, **kwargs):
"""
Initialize Telegram Object
"""
@ -135,11 +183,13 @@ class NotifyTelegram(NotifyBase):
# Parse our list
self.targets = parse_list(targets)
self.detect_owner = detect_owner
if self.user:
# Treat this as a channel too
self.targets.append(self.user)
if len(self.targets) == 0 and detect_bot_owner:
if len(self.targets) == 0 and self.detect_owner:
_id = self.detect_bot_owner()
if _id:
# Store our id
@ -502,6 +552,7 @@ class NotifyTelegram(NotifyBase):
'overflow': self.overflow_mode,
'image': self.include_image,
'verify': 'yes' if self.verify_certificate else 'no',
'detect': 'yes' if self.detect_owner else 'no',
}
# No need to check the user token because the user automatically gets
@ -589,4 +640,8 @@ class NotifyTelegram(NotifyBase):
results['include_image'] = \
parse_bool(results['qsd'].get('image', False))
# Include images with our message
results['detect_owner'] = \
parse_bool(results['qsd'].get('detect', True))
return results

View File

@ -47,6 +47,7 @@ from json import loads
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# Used to validate your personal access apikey
@ -93,6 +94,70 @@ class NotifyTwilio(NotifyBase):
# cause any title (if defined) to get placed into the message body.
title_maxlen = 0
# Define object templates
templates = (
'{schema}://{account_sid}:{auth_token}@{from_phone}',
'{schema}://{account_sid}:{auth_token}@{from_phone}/{targets}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'account_sid': {
'name': _('Account SID'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'AC[a-f0-9]{32}', 'i'),
},
'auth_token': {
'name': _('Auth Token'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[a-f0-9]{32}', 'i'),
},
'from_phone': {
'name': _('From Phone No'),
'type': 'string',
'required': True,
'regex': (r'\+?[0-9\s)(+-]+', 'i'),
'map_to': 'source',
},
'target_phone': {
'name': _('Target Phone No'),
'type': 'string',
'prefix': '+',
'regex': (r'[0-9\s)(+-]+', 'i'),
'map_to': 'targets',
},
'short_code': {
'name': _('Target Short Code'),
'type': 'string',
'regex': (r'[0-9]{5,6}', 'i'),
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
'from': {
'alias_of': 'from_phone',
},
'sid': {
'alias_of': 'account_sid',
},
'token': {
'alias_of': 'auth_token',
},
})
def __init__(self, account_sid, auth_token, source, targets=None,
**kwargs):
"""

View File

@ -27,6 +27,7 @@ from . import tweepy
from ..NotifyBase import NotifyBase
from ...common import NotifyType
from ...utils import parse_list
from ...AppriseLocale import gettext_lazy as _
class NotifyTwitter(NotifyBase):
@ -55,6 +56,54 @@ class NotifyTwitter(NotifyBase):
# Twitter does have titles when creating a message
title_maxlen = 0
templates = (
'{schema}://{user}@{ckey}{csecret}/{akey}/{asecret}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'ckey': {
'name': _('Consumer Key'),
'type': 'string',
'private': True,
'required': True,
},
'csecret': {
'name': _('Consumer Secret'),
'type': 'string',
'private': True,
'required': True,
},
'akey': {
'name': _('Access Key'),
'type': 'string',
'private': True,
'required': True,
},
'asecret': {
'name': _('Access Secret'),
'type': 'string',
'private': True,
'required': True,
},
'user': {
'name': _('User'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
})
def __init__(self, ckey, csecret, akey, asecret, targets=None, **kwargs):
"""
Initialize Twitter Object

View File

@ -63,6 +63,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..common import NotifyFormat
from ..AppriseLocale import gettext_lazy as _
# Token required as part of the API request
VALIDATE_TOKEN = re.compile(r'[a-z0-9]{80}', re.I)
@ -106,6 +107,22 @@ class NotifyWebexTeams(NotifyBase):
# Default to markdown; fall back to text
notify_format = NotifyFormat.MARKDOWN
# Define object templates
templates = (
'{schema}://{token}',
)
# Define our template tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'token': {
'name': _('Token'),
'type': 'string',
'private': True,
'required': True,
'regex': (r'[a-z0-9]{80}', 'i'),
},
})
def __init__(self, token, **kwargs):
"""
Initialize Webex Teams Object

View File

@ -32,6 +32,7 @@ from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
# Default our global support flag
NOTIFY_WINDOWS_SUPPORT_ENABLED = False
@ -88,6 +89,27 @@ class NotifyWindows(NotifyBase):
# let me know! :)
_enabled = NOTIFY_WINDOWS_SUPPORT_ENABLED
# Define object templates
templates = (
'{schema}://_/',
)
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'duration': {
'name': _('Duration'),
'type': 'int',
'min': 1,
'default': 12,
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
})
def __init__(self, include_image=True, duration=None, **kwargs):
"""
Initialize Windows Object

View File

@ -30,6 +30,7 @@ from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..common import NotifyImageSize
from ..utils import parse_bool
from ..AppriseLocale import gettext_lazy as _
class NotifyXBMC(NotifyBase):
@ -80,6 +81,54 @@ class NotifyXBMC(NotifyBase):
# KODI default protocol version (v6)
kodi_remote_protocol = 6
# Define object templates
templates = (
'{schema}://{host}',
'{schema}://{host}:{port}',
'{schema}://{user}:{password}@{host}',
'{schema}://{user}:{password}@{host}:{port}',
)
# Define our tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'duration': {
'name': _('Duration'),
'type': 'int',
'min': 1,
'default': 12,
},
'image': {
'name': _('Include Image'),
'type': 'bool',
'default': True,
'map_to': 'include_image',
},
})
def __init__(self, include_image=True, duration=None, **kwargs):
"""
Initialize XBMC/KODI Object

View File

@ -30,6 +30,7 @@ import requests
from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..common import NotifyType
from ..AppriseLocale import gettext_lazy as _
class NotifyXML(NotifyBase):
@ -56,6 +57,51 @@ class NotifyXML(NotifyBase):
# local anyway
request_rate_per_sec = 0
# Define object templates
# Define object templates
templates = (
'{schema}://{host}',
'{schema}://{host}:{port}',
'{schema}://{user}@{host}',
'{schema}://{user}@{host}:{port}',
'{schema}://{user}:{password}@{host}',
'{schema}://{user}:{password}@{host}:{port}',
)
# Define our tokens; these are the minimum tokens required required to
# be passed into this function (as arguments). The syntax appends any
# previously defined in the base package and builds onto them
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
},
})
# Define any kwargs we're using
template_kwargs = {
'headers': {
'name': _('HTTP Header'),
'prefix': '+',
},
}
def __init__(self, headers=None, **kwargs):
"""
Initialize XML Object

View File

@ -30,6 +30,7 @@ from os.path import isfile
from .NotifyBase import NotifyBase
from ..common import NotifyType
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
# xep string parser
XEP_PARSE_RE = re.compile('^[^1-9]*(?P<xep>[1-9][0-9]{0,3})$')
@ -98,6 +99,71 @@ class NotifyXMPP(NotifyBase):
# let me know! :)
_enabled = NOTIFY_XMPP_SUPPORT_ENABLED
# Define object templates
templates = (
'{schema}://{host}',
'{schema}://{password}@{host}',
'{schema}://{password}@{host}:{port}',
'{schema}://{user}:{password}@{host}',
'{schema}://{user}:{password}@{host}:{port}',
'{schema}://{host}/{targets}',
'{schema}://{password}@{host}/{targets}',
'{schema}://{password}@{host}:{port}/{targets}',
'{schema}://{user}:{password}@{host}/{targets}',
'{schema}://{user}:{password}@{host}:{port}/{targets}',
)
# Define our tokens
template_tokens = dict(NotifyBase.template_tokens, **{
'host': {
'name': _('Hostname'),
'type': 'string',
'required': True,
},
'port': {
'name': _('Port'),
'type': 'int',
'min': 1,
'max': 65535,
},
'user': {
'name': _('Username'),
'type': 'string',
},
'password': {
'name': _('Password'),
'type': 'string',
'private': True,
'required': True,
},
'target_jid': {
'name': _('Target JID'),
'type': 'string',
'map_to': 'targets',
},
'targets': {
'name': _('Targets'),
'type': 'list:string',
},
})
# Define our template arguments
template_args = dict(NotifyBase.template_args, **{
'to': {
'alias_of': 'targets',
},
'xep': {
'name': _('XEP'),
'type': 'list:string',
'prefix': 'xep-',
'regex': (r'[1-9][0-9]{0,3}', 'i'),
},
'jid': {
'name': _('Source JID'),
'type': 'string',
},
})
def __init__(self, targets=None, jid=None, xep=None, **kwargs):
"""
Initialize XMPP Object

View File

@ -25,6 +25,7 @@
import six
import re
import copy
from os import listdir
from os.path import dirname
@ -45,6 +46,9 @@ from ..common import NotifyImageSize
from ..common import NOTIFY_IMAGE_SIZES
from ..common import NotifyType
from ..common import NOTIFY_TYPES
from ..utils import parse_list
from ..AppriseLocale import gettext_lazy as _
from ..AppriseLocale import LazyTranslation
# Maintains a mapping of all of the Notification services
SCHEMA_MAP = {}
@ -67,6 +71,10 @@ __all__ = [
'tweepy',
]
# we mirror our base purely for the ability to reset everything; this
# is generally only used in testing and should not be used by developers
__MODULE_MAP = {}
# Load our Lookup Matrix
def __load_matrix(path=abspath(dirname(__file__)), name='apprise.plugins'):
@ -109,15 +117,20 @@ def __load_matrix(path=abspath(dirname(__file__)), name='apprise.plugins'):
# Filter out non-notification modules
continue
elif plugin_name in __all__:
elif plugin_name in __MODULE_MAP:
# we're already handling this object
continue
# Add our plugin name to our module map
__MODULE_MAP[plugin_name] = {
'plugin': plugin,
'module': module,
}
# Add our module name to our __all__
__all__.append(plugin_name)
# Ensure we provide the class as the reference to this directory and
# not the module:
# Load our module into memory so it's accessible to all
globals()[plugin_name] = plugin
# Load protocol(s) if defined
@ -147,5 +160,257 @@ def __load_matrix(path=abspath(dirname(__file__)), name='apprise.plugins'):
return SCHEMA_MAP
# Reset our Lookup Matrix
def __reset_matrix():
"""
Restores the Lookup matrix to it's base setting. This is only used through
testing and should not be directly called.
"""
# Reset our schema map
SCHEMA_MAP.clear()
# Iterate over our module map so we can clear out our __all__ and globals
for plugin_name in __MODULE_MAP.keys():
# Clear out globals
del globals()[plugin_name]
# Remove element from plugins
__all__.remove(plugin_name)
# Clear out our module map
__MODULE_MAP.clear()
# Dynamically build our schema base
__load_matrix()
def _sanitize_token(tokens, default_delimiter):
"""
This is called by the details() function and santizes the output by
populating expected and consistent arguments if they weren't otherwise
specified.
"""
# Iterate over our tokens
for key in tokens.keys():
for element in tokens[key].keys():
# Perform translations (if detected to do so)
if isinstance(tokens[key][element], LazyTranslation):
tokens[key][element] = str(tokens[key][element])
if 'alias_of' in tokens[key]:
# Do not touch this field
continue
if 'map_to' not in tokens[key]:
# Default type to key
tokens[key]['map_to'] = key
if 'type' not in tokens[key]:
# Default type to string
tokens[key]['type'] = 'string'
elif tokens[key]['type'].startswith('list') \
and 'delim' not in tokens[key]:
# Default list delimiter (if not otherwise specified)
tokens[key]['delim'] = default_delimiter
elif tokens[key]['type'].startswith('choice') \
and 'default' not in tokens[key] \
and 'values' in tokens[key] \
and len(tokens[key]['values']) == 1:
# If there is only one choice; then make it the default
tokens[key]['default'] = \
tokens[key]['values'][0]
if 'regex' in tokens[key]:
# Verify that we are a tuple; convert strings to tuples
if isinstance(tokens[key]['regex'], six.string_types):
# Default tuple setup
tokens[key]['regex'] = \
(tokens[key]['regex'], None)
elif not isinstance(tokens[key]['regex'], (list, tuple)):
# Invalid regex
del tokens[key]['regex']
if 'required' not in tokens[key]:
# Default required is False
tokens[key]['required'] = False
if 'private' not in tokens[key]:
# Private flag defaults to False if not set
tokens[key]['private'] = False
return
def details(plugin):
"""
Provides templates that can be used by developers to build URLs
dynamically.
If a list of templates is provided, then they will be used over
the default value.
If a list of tokens are provided, then they will over-ride any
additional settings built from this script and/or will be appended
to them afterwards.
"""
# Our unique list of parsing will be based on the provided templates
# if none are provided we will use our own
templates = tuple(plugin.templates)
# The syntax is simple
# {
# # The token_name must tie back to an entry found in the
# # templates list.
# 'token_name': {
#
# # types can be 'string', 'int', 'choice', 'list, 'float'
# # both choice and list may additionally have a : identify
# # what the list/choice type is comprised of; the default
# # is string.
# 'type': 'choice:string',
#
# # values will only exist the type must be a fixed
# # list of inputs (generated from type choice for example)
#
# # If this is a choice:bool then you should ALWAYS define
# # this list as a (True, False) such as ('Yes, 'No') or
# # ('Enabled', 'Disabled'), etc
# 'values': [ 'http', 'https' ],
#
# # Identifies if the entry specified is required or not
# 'required': True,
#
# # Identify a default value
# 'default': 'http',
#
# # Optional Verification Entries min and max are for floats
# # and/or integers
# 'min': 4,
# 'max': 5,
#
# # A list will always identify a delimiter. If this is
# # part of a path, this may be a '/', or it could be a
# # comma and/or space. delimiters are always in a list
# # eg (if space and/or comma is a delimiter the entry
# # would look like: 'delim': [',' , ' ' ]
# 'delim': None,
#
# # Use regex if you want to share the regular expression
# # required to validate the field. The regex will never
# # accomodate the prefix (if one is specified). That is
# # up to the user building the URLs to include the prefix
# # on the URL when constructing it.
# # The format is ('regex', 'reg options')
# 'regex': (r'[A-Z0-9]+', 'i'),
#
# # A Prefix is always a string, to differentiate between
# # multiple arguments, sometimes content is prefixed.
# 'prefix': '@',
#
# # By default the key of this object is to be interpreted
# # as the argument to the notification in question. However
# # To accomodate cases where there are multiple types that
# # all map to the same entry, one can find a map_to value.
# 'map_to': 'function_arg',
#
# # Some arguments act as an alias_of an already defined object
# # This plays a role more with configuration file generation
# # since yaml files allow you to define different argumuments
# # in line to simplify things. If this directive is set, then
# # it should be treated exactly the same as the object it is
# # an alias of
# 'alias_of': 'function_arg',
#
# # Advise developers to consider the potential sensitivity
# # of this field owned by the user. This is for passwords,
# # and api keys, etc...
# 'private': False,
# },
# }
# Template tokens identify the arguments required to initialize the
# plugin itself. It identifies all of the tokens and provides some
# details on their use. Each token defined should in some way map
# back to at least one URL {token} defined in the templates
# Since we nest a dictionary within a dictionary, a simple copy isn't
# enough. a deepcopy allows us to manipulate this object in this
# funtion without obstructing the original.
template_tokens = copy.deepcopy(plugin.template_tokens)
# Arguments and/or Options either have a default value and/or are
# optional to be set.
#
# Since we nest a dictionary within a dictionary, a simple copy isn't
# enough. a deepcopy allows us to manipulate this object in this
# funtion without obstructing the original.
template_args = copy.deepcopy(plugin.template_args)
# Our template keyword arguments ?+key=value&-key=value
# Basically the user provides both the key and the value. this is only
# possibly by identifying the key prefix required for them to be
# interpreted hence the +/- keys are built into apprise by default for easy
# reference. In these cases, entry might look like '+' being the prefix:
# {
# 'arg_name': {
# 'name': 'label',
# 'prefix': '+',
# }
# }
#
# Since we nest a dictionary within a dictionary, a simple copy isn't
# enough. a deepcopy allows us to manipulate this object in this
# funtion without obstructing the original.
template_kwargs = copy.deepcopy(plugin.template_kwargs)
# We automatically create a schema entry
template_tokens['schema'] = {
'name': _('Schema'),
'type': 'choice:string',
'required': True,
'values': parse_list(plugin.secure_protocol, plugin.protocol)
}
# Sanitize our tokens
_sanitize_token(template_tokens, default_delimiter=('/', ))
# Delimiter(s) are space and/or comma
_sanitize_token(template_args, default_delimiter=(',', ' '))
_sanitize_token(template_kwargs, default_delimiter=(',', ' '))
# Argument/Option Handling
for key in list(template_args.keys()):
# _lookup_default looks up what the default value
if '_lookup_default' in template_args[key]:
template_args[key]['default'] = getattr(
plugin, template_args[key]['_lookup_default'])
# Tidy as we don't want to pass this along in response
del template_args[key]['_lookup_default']
# _exists_if causes the argument to only exist IF after checking
# the return of an internal variable requiring a check
if '_exists_if' in template_args[key]:
if not getattr(plugin,
template_args[key]['_exists_if']):
# Remove entire object
del template_args[key]
else:
# We only nee to remove this key
del template_args[key]['_exists_if']
return {
'templates': templates,
'tokens': template_tokens,
'args': template_args,
'kwargs': template_kwargs,
}

View File

@ -25,6 +25,8 @@
import re
import six
import contextlib
import os
from os.path import expanduser
try:
@ -589,3 +591,29 @@ def is_exclusive_match(logic, data):
# Return True if we matched against our logic (or simply none was
# specified).
return matched
@contextlib.contextmanager
def environ(*remove, **update):
"""
Temporarily updates the ``os.environ`` dictionary in-place.
The ``os.environ`` dictionary is updated in-place so that the modification
is sure to work in all situations.
:param remove: Environment variable(s) to remove.
:param update: Dictionary of environment variables and values to
add/update.
"""
# Create a backup of our environment for restoration purposes
env_orig = os.environ.copy()
try:
os.environ.update(update)
[os.environ.pop(k, None) for k in remove]
yield
finally:
# Restore our snapshot
os.environ = env_orig.copy()

View File

@ -4,3 +4,4 @@ mock
pytest
pytest-cov
tox
babel

View File

@ -95,8 +95,10 @@ Requires: python-six
Requires: python-markdown
%if 0%{?rhel} && 0%{?rhel} <= 7
BuildRequires: python-yaml
BuildRequires: python-babel
%else
Requires: python2-yaml
Requires: python2-babel
%endif # using rhel7
%if %{with tests}
@ -141,6 +143,7 @@ BuildRequires: python%{python3_pkgversion}-six
BuildRequires: python%{python3_pkgversion}-click >= 5.0
BuildRequires: python%{python3_pkgversion}-markdown
BuildRequires: python%{python3_pkgversion}-yaml
BuildRequires: python%{python3_pkgversion}-babel
Requires: python%{python3_pkgversion}-decorator
Requires: python%{python3_pkgversion}-requests
Requires: python%{python3_pkgversion}-requests-oauthlib
@ -167,9 +170,11 @@ BuildRequires: python%{python3_pkgversion}-pytest-runner
%build
%if 0%{?with_python2}
%{__python2} setup.py compile_catalog
%py2_build
%endif # with_python2
%if 0%{?with_python3}
%{__python3} setup.py compile_catalog
%py3_build
%endif # with_python3

View File

@ -5,17 +5,12 @@ universal = 1
# ensure LICENSE is included in wheel metadata
license_file = LICENSE
[pycodestyle]
# We exclude packages we don't maintain
exclude = .eggs,.tox,gntp,tweepy,pushjet
ignore = E722,W503,W504
statistics = true
[flake8]
# We exclude packages we don't maintain
exclude = .eggs,.tox,gntp,tweepy,pushjet
ignore = E722,W503,W504
statistics = true
builtins = _
[aliases]
test=pytest
@ -26,3 +21,28 @@ python_files = test/test_*.py
filterwarnings =
once::Warning
strict = true
[extract_messages]
output-file = apprise/i18n/apprise.pot
sort-output = true
copyright-holder = Chris Caron
msgid-bugs-address = lead2gold@gmail.com
charset = utf-8
no-location = true
add-comments = false
[compile_catalog]
domain = apprise
directory = apprise/i18n
statistics = true
use-fuzzy = true
[init_catalog]
domain = apprise
input-file = apprise/i18n/apprise.pot
output-dir = apprise/i18n
[update_catalog]
domain = apprise
input-file = apprise/i18n/apprise.pot
output-dir = apprise/i18n

View File

@ -33,6 +33,7 @@ except ImportError:
from distutils.core import setup
from setuptools import find_packages
from babel.messages import frontend as babel
install_options = os.environ.get("APPRISE_INSTALL", "").split(",")
install_requires = open('requirements.txt').readlines()
@ -55,6 +56,12 @@ setup(
license='MIT',
long_description=open('README.md').read(),
long_description_content_type='text/markdown',
cmdclass={
'compile_catalog': babel.compile_catalog,
'extract_messages': babel.extract_messages,
'init_catalog': babel.init_catalog,
'update_catalog': babel.update_catalog,
},
url='https://github.com/caronc/apprise',
keywords='Push Notifications Alerts Email AWS SNS Boxcar Discord Dbus '
'Emby Faast Flock Gitter Gnome Gotify Growl IFTTT Join KODI Mailgun '
@ -69,6 +76,8 @@ setup(
'assets/NotifyXML-1.0.xsd',
'assets/themes/default/*.png',
'assets/themes/default/*.ico',
'i18n/*.py',
'i18n/*/LC_MESSAGES/*.mo',
],
},
install_requires=install_requires,
@ -87,6 +96,6 @@ setup(
),
entry_points={'console_scripts': console_scripts},
python_requires='>=2.7',
setup_requires=['pytest-runner', ],
setup_requires=['pytest-runner', 'babel', ],
tests_require=open('dev-requirements.txt').readlines(),
)

View File

@ -24,6 +24,7 @@
# THE SOFTWARE.
from __future__ import print_function
import re
import sys
import six
import pytest
@ -43,6 +44,9 @@ from apprise import __version__
from apprise.plugins import SCHEMA_MAP
from apprise.plugins import __load_matrix
from apprise.plugins import __reset_matrix
from apprise.utils import parse_list
import inspect
# Disable logging for a cleaner testing output
import logging
@ -255,6 +259,10 @@ def test_apprise():
a.clear()
assert(len(a) == 0)
# Instantiate a bad object
plugin = a.instantiate(object, tag="bad_object")
assert plugin is None
# Instantiate a good object
plugin = a.instantiate('good://localhost', tag="good")
assert(isinstance(plugin, NotifyBase))
@ -292,6 +300,57 @@ def test_apprise():
'throw://localhost', suppress_exceptions=True) is None)
assert(len(a) == 0)
#
# We rince and repeat the same tests as above, however we do them
# using the dict version
#
# Reset our object
a.clear()
assert(len(a) == 0)
# Instantiate a good object
plugin = a.instantiate({
'schema': 'good',
'host': 'localhost'}, tag="good")
assert(isinstance(plugin, NotifyBase))
# Test simple tagging inside of the object
assert("good" in plugin)
assert("bad" not in plugin)
# the in (__contains__ override) is based on or'ed content; so although
# 'bad' isn't tagged as being in the plugin, 'good' is, so the return
# value of this is True
assert(["bad", "good"] in plugin)
assert(set(["bad", "good"]) in plugin)
assert(("bad", "good") in plugin)
# We an add already substatiated instances into our Apprise object
a.add(plugin)
assert(len(a) == 1)
# We can add entries as a list too (to add more then one)
a.add([plugin, plugin, plugin])
assert(len(a) == 4)
# Reset our object again
a.clear()
try:
a.instantiate({
'schema': 'throw',
'host': 'localhost'}, suppress_exceptions=False)
assert(False)
except TypeError:
assert(True)
assert(len(a) == 0)
assert(a.instantiate({
'schema': 'throw',
'host': 'localhost'}, suppress_exceptions=True) is None)
assert(len(a) == 0)
@mock.patch('requests.get')
@mock.patch('requests.post')
@ -320,9 +379,16 @@ def test_apprise_tagging(mock_post, mock_get):
# An invalid addition can't add the tag
assert(a.add('averyinvalidschema://localhost', tag='uhoh') is False)
assert(a.add({
'schema': 'averyinvalidschema',
'host': 'localhost'}, tag='uhoh') is False)
# Add entry and assign it to a tag called 'awesome'
assert(a.add('json://localhost/path1/', tag='awesome') is True)
assert(a.add({
'schema': 'json',
'host': 'localhost',
'fullpath': '/path1/'}, tag='awesome') is True)
# Add another notification and assign it to a tag called 'awesome'
# and another tag called 'local'
@ -674,10 +740,131 @@ def test_apprise_details():
API: Apprise() Details
"""
# Reset our matrix
__reset_matrix()
# Caling load matix a second time which is an internal function causes it
# to skip over content already loaded into our matrix and thefore accesses
# other if/else parts of the code that aren't otherwise called
# This is a made up class that is just used to verify
class TestDetailNotification(NotifyBase):
"""
This class is used to test various configurations supported
"""
# Minimum requirements for a plugin to produce details
service_name = 'Detail Testing'
# The default simple (insecure) protocol (used by NotifyMail)
protocol = 'details'
# Set test_bool flag
always_true = True
always_false = False
# Define object templates
templates = (
'{schema}://{host}',
'{schema}://{host}:{port}',
'{schema}://{user}@{host}:{port}',
'{schema}://{user}:{pass}@{host}:{port}',
)
# Define our tokens; these are the minimum tokens required required to
# be passed into this function (as arguments). The syntax appends any
# previously defined in the base package and builds onto them
template_tokens = dict(NotifyBase.template_tokens, **{
'notype': {
# Nothing defined is still valid
},
'regex_test01': {
'name': _('RegexTest'),
'type': 'string',
'regex': r'[A-Z0-9]',
},
'regex_test02': {
'name': _('RegexTest'),
# Support regex options too
'regex': (r'[A-Z0-9]', 'i'),
},
'regex_test03': {
'name': _('RegexTest'),
# Support regex option without a second option
'regex': (r'[A-Z0-9]'),
},
'regex_test04': {
# this entry would just end up getting removed
'regex': None,
},
# List without delimiters (causes defaults to kick in)
'mylistA': {
'name': 'fruit',
'type': 'list:string',
},
# A list with a delimiter list
'mylistB': {
'name': 'softdrinks',
'type': 'list:string',
'delim': ['|', '-'],
},
})
template_args = dict(NotifyBase.template_args, **{
# Test _exist_if logic
'test_exists_if_01': {
'name': 'Always False',
'type': 'bool',
# Provide a default
'default': False,
# Base the existance of this key/value entry on the lookup
# of this class value at runtime. Hence:
# if not NotifyObject.always_false
# del this_entry
#
'_exists_if': 'always_false',
},
# Test _exist_if logic
'test_exists_if_02': {
'name': 'Always True',
'type': 'bool',
# Provide a default
'default': False,
# Base the existance of this key/value entry on the lookup
# of this class value at runtime. Hence:
# if not NotifyObject.always_true
# del this_entry
#
'_exists_if': 'always_true',
},
})
def url(self):
# Support URL
return ''
def notify(self, **kwargs):
# Pretend everything is okay (so we don't break other tests)
return True
# Store our good detail notification in our schema map
SCHEMA_MAP['details'] = TestDetailNotification
# Create our Apprise instance
a = Apprise()
# Dictionary response
assert isinstance(a.details(), dict)
# Reset our matrix
__reset_matrix()
__load_matrix()
def test_apprise_details_plugin_verification():
"""
API: Apprise() Details Plugin Verification
"""
# Reset our matrix
__reset_matrix()
__load_matrix()
a = Apprise()
@ -688,6 +875,18 @@ def test_apprise_details():
# Dictionary response
assert isinstance(details, dict)
# Details object with language defined:
details = a.details(lang='en')
# Dictionary response
assert isinstance(details, dict)
# Details object with unsupported language:
details = a.details(lang='xx')
# Dictionary response
assert isinstance(details, dict)
# Apprise version
assert 'version' in details
assert details.get('version') == __version__
@ -707,10 +906,240 @@ def test_apprise_details():
assert 'image_url_mask' in details['asset']
assert 'image_url_logo' in details['asset']
# All plugins must have a name defined; the below generates
# a list of entrys that do not have a string defined.
assert(not len([x['service_name'] for x in details['schemas']
if not isinstance(x['service_name'], six.string_types)]))
# Valid Type Regular Expression Checker
# Case Sensitive and MUST match the following:
is_valid_type_re = re.compile(
r'((choice|list):)?(string|bool|int|float)')
# match tokens found in templates so we can cross reference them back
# to see if they have a matching argument
template_token_re = re.compile(r'{([^}]+)}[^{]*?(?=$|{)')
# Define acceptable map_to arguments that can be tied in with the
# kwargs function definitions.
valid_kwargs = set([
# URL prepared kwargs
'user', 'password', 'port', 'host', 'schema', 'fullpath',
# URLBase and NotifyBase args:
'verify', 'format', 'overflow',
])
# Valid Schema Entries:
valid_schema_keys = (
'name', 'private', 'required', 'type', 'values', 'min', 'max',
'regex', 'default', 'list', 'delim', 'prefix', 'map_to', 'alias_of',
)
for entry in details['schemas']:
# Track the map_to entries (if specified); We need to make sure that
# these properly map back
map_to_entries = set()
# Track the alias_of entries
map_to_aliases = set()
# A Service Name MUST be defined
assert 'service_name' in entry
assert isinstance(entry['service_name'], six.string_types)
# Acquire our protocols
protocols = parse_list(
entry['protocols'], entry['secure_protocols'])
# At least one schema/protocol MUST be defined
assert len(protocols) > 0
# our details
assert 'details' in entry
assert isinstance(entry['details'], dict)
# All schema details should include args
for section in ['kwargs', 'args', 'tokens']:
assert section in entry['details']
assert isinstance(entry['details'][section], dict)
for key, arg in entry['details'][section].items():
# Validate keys (case-sensitive)
assert len([k for k in arg.keys()
if k not in valid_schema_keys]) == 0
# Test our argument
assert isinstance(arg, dict)
if 'alias_of' not in arg:
# Minimum requirement of an argument
assert 'name' in arg
assert isinstance(arg['name'], six.string_types)
assert 'type' in arg
assert isinstance(arg['type'], six.string_types)
assert is_valid_type_re.match(arg['type']) is not None
if 'min' in arg:
assert arg['type'].endswith('float') \
or arg['type'].endswith('int')
assert isinstance(arg['min'], (int, float))
if 'max' in arg:
# If a min and max was specified, at least check
# to confirm the min is less then the max
assert arg['min'] < arg['max']
if 'max' in arg:
assert arg['type'].endswith('float') \
or arg['type'].endswith('int')
assert isinstance(arg['max'], (int, float))
if 'private' in arg:
assert isinstance(arg['private'], bool)
if 'required' in arg:
assert isinstance(arg['required'], bool)
if 'prefix' in arg:
assert isinstance(arg['prefix'], six.string_types)
if section == 'kwargs':
# The only acceptable prefix types for kwargs
assert arg['prefix'] in ('+', '-')
else:
# kwargs requires that the 'prefix' is defined
assert section != 'kwargs'
if 'map_to' in arg:
# must be a string
assert isinstance(arg['map_to'], six.string_types)
# Track our map_to object
map_to_entries.add(arg['map_to'])
else:
map_to_entries.add(key)
# Some verification
if arg['type'].startswith('choice'):
# choice:bool is redundant and should be swapped to
# just bool
assert not arg['type'].endswith('bool')
# Choices require that a values list is provided
assert 'values' in arg
assert isinstance(arg['values'], (list, tuple))
assert len(arg['values']) > 0
# Test default
if 'default' in arg:
# if a default is provided on a choice object,
# it better be in the list of values
assert arg['default'] in arg['values']
if arg['type'].startswith('bool'):
# Boolean choices are less restrictive but require a
# default value
assert 'default' in arg
assert isinstance(arg['default'], bool)
if 'regex' in arg:
# Regex must ALWAYS be in the format (regex, option)
assert isinstance(arg['regex'], (tuple, list))
assert len(arg['regex']) == 2
assert isinstance(arg['regex'][0], six.string_types)
assert arg['regex'][1] is None or isinstance(
arg['regex'][1], six.string_types)
# Compile the regular expression to verify that it is
# valid
try:
re.compile(arg['regex'][0])
except:
assert '{} is an invalid regex'\
.format(arg['regex'][0])
# Regex should never start and/or end with ^/$; leave
# that up to the user making use of the regex instead
assert re.match(r'^[()\s]*\^', arg['regex'][0]) is None
assert re.match(r'[()\s$]*\$', arg['regex'][0]) is None
if arg['type'].startswith('list'):
# Delimiters MUST be defined
assert 'delim' in arg
assert isinstance(arg['delim'], (list, tuple))
assert len(arg['delim']) > 0
else: # alias_of is in the object
# must be a string
assert isinstance(arg['alias_of'], six.string_types)
# Track our alias_of object
map_to_aliases.add(arg['alias_of'])
# We should never map to ourselves
assert arg['alias_of'] != key
# 2 entries (name, and alias_of only!)
assert len(entry['details'][section][key]) == 1
# inspect our object
spec = inspect.getargspec(SCHEMA_MAP[protocols[0]].__init__)
function_args = \
(set(parse_list(spec.keywords)) - set(['kwargs'])) \
| (set(spec.args) - set(['self'])) | valid_kwargs
# Iterate over our map_to_entries and make sure that everything
# maps to a function argument
for arg in map_to_entries:
if arg not in function_args:
# This print statement just makes the error easier to
# troubleshoot
print('{}:// template/arg/func reference missing error.'
.format(protocols[0]))
assert arg in function_args
# Iterate over all of the function arguments and make sure that
# it maps back to a key
function_args -= valid_kwargs
for arg in function_args:
if arg not in map_to_entries:
# This print statement just makes the error easier to
# troubleshoot
print('{}:// template/func/arg reference missing error.'
.format(protocols[0]))
assert arg in map_to_entries
# Iterate over our map_to_aliases and make sure they were defined in
# either the as a token or arg
for arg in map_to_aliases:
assert arg in set(entry['details']['args'].keys()) \
| set(entry['details']['tokens'].keys())
# Template verification
assert 'templates' in entry['details']
assert isinstance(entry['details']['templates'], (set, tuple, list))
# Iterate over our templates and parse our arguments
for template in entry['details']['templates']:
# Ensure we've properly opened and closed all of our tokens
assert template.count('{') == template.count('}')
expected_tokens = template.count('}')
args = template_token_re.findall(template)
assert expected_tokens == len(args)
# Build a cross reference set of our current defined objects
defined_tokens = set()
for key, arg in entry['details']['tokens'].items():
defined_tokens.add(key)
if 'alias_of' in arg:
defined_tokens.add(arg['alias_of'])
# We want to make sure all of our defined tokens have been
# accounted for in at least one defined template
for arg in args:
assert arg in set(entry['details']['args'].keys()) \
| set(entry['details']['tokens'].keys())
# The reverse of the above; make sure that each entry defined
# in the template_tokens is accounted for in at least one of
# the defined templates
assert arg in defined_tokens
def test_notify_matrix_dynamic_importing(tmpdir):

View File

@ -358,14 +358,9 @@ urls:
- tag: my-custom-tag, my-other-tag
# How to stack multiple entries:
- mailto://:
- user: jeff
pass: 123abc
from: jeff@yahoo.ca
- user: jack
pass: pass123
from: jack@hotmail.com
- mailto://user:123abc@yahoo.ca:
- to: test@examle.com
- to: test2@examle.com
# This is an illegal entry; the schema can not be changed
schema: json

View File

@ -90,6 +90,9 @@ TEST_URLS = (
('mailtos://user:pass@nuxref.com:567?to=l2g@nuxref.com', {
'instance': plugins.NotifyEmail,
}),
('mailtos://user:pass@nuxref.com:567/l2g@nuxref.com', {
'instance': plugins.NotifyEmail,
}),
(
'mailtos://user:pass@example.com?smtp=smtp.example.com&timeout=5'
'&name=l2g&from=noreply@example.com', {
@ -126,9 +129,11 @@ TEST_URLS = (
('mailtos://nuxref.com?user=&pass=.', {
'instance': TypeError,
}),
# Invalid To Address
# Invalid To Address is accepted, but we won't be able to properly email
# using the notify() call
('mailtos://user:pass@nuxref.com?to=@', {
'instance': TypeError,
'instance': plugins.NotifyEmail,
'response': False,
}),
# Valid URL, but can't structure a proper email
('mailtos://nuxref.com?user=%20!&pass=.', {
@ -171,7 +176,7 @@ def test_email_plugin(mock_smtp, mock_smtpssl):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# iterate over our dictionary and test it out
for (url, meta) in TEST_URLS:
@ -234,7 +239,7 @@ def test_email_plugin(mock_smtp, mock_smtpssl):
assert(isinstance(obj, instance))
if isinstance(obj, plugins.NotifyBase.NotifyBase):
if isinstance(obj, plugins.NotifyBase):
# We loaded okay; now lets make sure we can reverse this url
assert(isinstance(obj.url(), six.string_types) is True)
@ -244,7 +249,7 @@ def test_email_plugin(mock_smtp, mock_smtpssl):
# Our object should be the same instance as what we had
# originally expected above.
if not isinstance(obj_cmp, plugins.NotifyBase.NotifyBase):
if not isinstance(obj_cmp, plugins.NotifyBase):
# Assert messages are hard to trace back with the way
# these tests work. Just printing before throwing our
# assertion failure makes things easier to debug later on
@ -333,7 +338,8 @@ def test_webbase_lookup(mock_smtp, mock_smtpssl):
'mailto://user:pass@l2g.com', suppress_exceptions=True)
assert(isinstance(obj, plugins.NotifyEmail))
assert obj.to_addr == 'user@l2g.com'
assert len(obj.targets) == 1
assert 'user@l2g.com' in obj.targets
assert obj.from_addr == 'user@l2g.com'
assert obj.password == 'pass'
assert obj.user == 'user'
@ -355,7 +361,7 @@ def test_smtplib_init_fail(mock_smtplib):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
obj = Apprise.instantiate(
'mailto://user:pass@gmail.com', suppress_exceptions=False)
@ -380,7 +386,7 @@ def test_smtplib_send_okay(mock_smtplib):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Defaults to HTML
obj = Apprise.instantiate(
@ -476,8 +482,9 @@ def test_email_url_variations():
assert obj.password == 'abcd123'
assert obj.user == 'apprise@example21.ca'
assert obj.to_addr == 'apprise@example.com'
assert obj.to_addr == obj.from_addr
assert len(obj.targets) == 1
assert 'apprise@example.com' in obj.targets
assert obj.targets[0] == obj.from_addr
# test user and password specified in the url body (as an argument)
# this always over-rides the entries at the front of the url
@ -492,8 +499,9 @@ def test_email_url_variations():
assert obj.password == 'abcd123'
assert obj.user == 'apprise@example21.ca'
assert obj.to_addr == 'apprise@example.com'
assert obj.to_addr == obj.from_addr
assert len(obj.targets) == 1
assert 'apprise@example.com' in obj.targets
assert obj.targets[0] == obj.from_addr
assert obj.smtp_host == 'example.com'
# test a complicated example
@ -515,5 +523,21 @@ def test_email_url_variations():
assert obj.host == 'example.com'
assert obj.port == 1234
assert obj.smtp_host == 'smtp.example.edu'
assert obj.to_addr == 'to@example.jp'
assert len(obj.targets) == 1
assert 'to@example.jp' in obj.targets
assert obj.from_addr == 'from@example.jp'
def test_email_dict_variations():
"""
API: Test email dictionary variations to ensure parsing is correct
"""
# Test variations of username required to be an email address
# user@example.com
obj = Apprise.instantiate({
'schema': 'mailto',
'user': 'apprise@example.com',
'password': 'abd123',
'host': 'example.com'}, suppress_exceptions=False)
assert isinstance(obj, plugins.NotifyEmail) is True

View File

@ -43,7 +43,7 @@ def test_notify_gitter_plugin_general(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Generate a valid token (40 characters)
token = 'a' * 40

View File

@ -223,7 +223,7 @@ def test_growl_plugin(mock_gntp):
assert(isinstance(obj, instance) is True)
if isinstance(obj, plugins.NotifyBase.NotifyBase):
if isinstance(obj, plugins.NotifyBase):
# We loaded okay; now lets make sure we can reverse this url
assert(isinstance(obj.url(), six.string_types) is True)
@ -233,7 +233,7 @@ def test_growl_plugin(mock_gntp):
# Our object should be the same instance as what we had
# originally expected above.
if not isinstance(obj_cmp, plugins.NotifyBase.NotifyBase):
if not isinstance(obj_cmp, plugins.NotifyBase):
# Assert messages are hard to trace back with the way
# these tests work. Just printing before throwing our
# assertion failure makes things easier to debug later on

162
test/test_locale.py Normal file
View File

@ -0,0 +1,162 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import mock
import ctypes
from apprise import AppriseLocale
try:
# Python v3.4+
from importlib import reload
except ImportError:
try:
# Python v3.0-v3.3
from imp import reload
except ImportError:
# Python v2.7
pass
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
@mock.patch('gettext.install')
def test_apprise_locale(mock_gettext_install):
"""
API: Test apprise locale object
"""
lazytrans = AppriseLocale.LazyTranslation('Token')
assert str(lazytrans) == 'Token'
@mock.patch('gettext.install')
def test_gettext_init(mock_gettext_install):
"""
API: Mock Gettext init
"""
mock_gettext_install.side_effect = ImportError()
# Test our fall back to not supporting translations
reload(AppriseLocale)
# Objects can still be created
al = AppriseLocale.AppriseLocale()
with al.lang_at('en'):
# functions still behave as normal
pass
# restore the object
mock_gettext_install.side_effect = None
reload(AppriseLocale)
@mock.patch('gettext.translation')
def test_gettext_translations(mock_gettext_trans):
"""
API: Apprise() Gettext translations
"""
mock_gettext_trans.side_effect = IOError()
# This throws internally but we handle it gracefully
al = AppriseLocale.AppriseLocale()
with al.lang_at('en'):
# functions still behave as normal
pass
# This throws internally but we handle it gracefully
AppriseLocale.AppriseLocale(language="fr")
@mock.patch('gettext.translation')
def test_gettext_installs(mock_gettext_trans):
"""
API: Apprise() Gettext install
"""
mock_lang = mock.Mock()
mock_lang.install.return_value = True
mock_gettext_trans.return_value = mock_lang
# This throws internally but we handle it gracefully
al = AppriseLocale.AppriseLocale()
with al.lang_at('en'):
# functions still behave as normal
pass
# This throws internally but we handle it gracefully
AppriseLocale.AppriseLocale(language="fr")
# Force a few different languages
al._gtobjs['en'] = mock_lang
al._gtobjs['es'] = mock_lang
al.lang = 'en'
with al.lang_at('en'):
# functions still behave as normal
pass
with al.lang_at('es'):
# functions still behave as normal
pass
with al.lang_at('fr'):
# functions still behave as normal
pass
@mock.patch('locale.getdefaultlocale')
def test_detect_language(mock_getlocale):
"""
API: Apprise() Detect language
"""
if not hasattr(ctypes, 'windll'):
windll = mock.Mock()
# 4105 = en_CA
windll.kernel32.GetUserDefaultUILanguage.return_value = 4105
setattr(ctypes, 'windll', windll)
# The below accesses the windows fallback code
assert AppriseLocale.AppriseLocale.detect_language() == 'en'
assert AppriseLocale.AppriseLocale\
.detect_language(detect_fallback=False) is None
# Handle case where getdefaultlocale() can't be detected
mock_getlocale.return_value = None
delattr(ctypes, 'windll')
assert AppriseLocale.AppriseLocale.detect_language() is None
# if detect_language and windows env fail us, then we don't
# set up a default language on first load
AppriseLocale.AppriseLocale()

View File

@ -43,7 +43,7 @@ def test_notify_matrix_plugin_general(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
response_obj = {
'room_id': '!abc123:localhost',
@ -158,7 +158,7 @@ def test_notify_matrix_plugin_fetch(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
response_obj = {
'room_id': '!abc123:localhost',
@ -205,7 +205,7 @@ def test_notify_matrix_plugin_fetch(mock_post, mock_get):
assert obj.send(user='test', password='passwd', body="test") is False
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
response_obj = {
# Registration
@ -227,7 +227,7 @@ def test_notify_matrix_plugin_fetch(mock_post, mock_get):
mock_post.return_value = request
mock_get.return_value = request
obj = plugins.NotifyMatrix()
obj = plugins.NotifyMatrix(include_image=True)
assert isinstance(obj, plugins.NotifyMatrix) is True
assert obj.access_token is None
assert obj._register() is True
@ -264,7 +264,7 @@ def test_notify_matrix_plugin_auth(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
response_obj = {
# Registration
@ -360,7 +360,7 @@ def test_notify_matrix_plugin_rooms(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
response_obj = {
# Registration
@ -539,3 +539,87 @@ def test_notify_matrix_url_parsing():
assert '#room1' in result['targets']
assert '#room2' in result['targets']
assert '#room3' in result['targets']
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_notify_matrix_plugin_image_errors(mock_post, mock_get):
"""
API: NotifyMatrix() Image Error Handling
"""
def mock_function_handing(url, data, **kwargs):
"""
dummy function for handling image posts (as a failure)
"""
response_obj = {
'room_id': '!abc123:localhost',
'room_alias': '#abc123:localhost',
'joined_rooms': ['!abc123:localhost', '!def456:localhost'],
'access_token': 'abcd1234',
'home_server': 'localhost',
}
request = mock.Mock()
request.content = dumps(response_obj)
request.status_code = requests.codes.ok
if 'm.image' in data:
# Fail for images
request.status_code = 400
return request
# Prepare Mock
mock_get.side_effect = mock_function_handing
mock_post.side_effect = mock_function_handing
obj = plugins.NotifyMatrix(include_image=True)
assert isinstance(obj, plugins.NotifyMatrix) is True
assert obj.access_token is None
# Notification was successful, however we could not post image and since
# we had post errors (of any kind) we still report a failure.
assert obj.notify('test', 'test') is False
obj = plugins.NotifyMatrix(include_image=False)
assert isinstance(obj, plugins.NotifyMatrix) is True
assert obj.access_token is None
# We didn't post an image (which was set to fail) and therefore our
# post was okay
assert obj.notify('test', 'test') is True
def mock_function_handing(url, data, **kwargs):
"""
dummy function for handling image posts (successfully)
"""
response_obj = {
'room_id': '!abc123:localhost',
'room_alias': '#abc123:localhost',
'joined_rooms': ['!abc123:localhost', '!def456:localhost'],
'access_token': 'abcd1234',
'home_server': 'localhost',
}
request = mock.Mock()
request.content = dumps(response_obj)
request.status_code = requests.codes.ok
return request
# Prepare Mock
mock_get.side_effect = mock_function_handing
mock_post.side_effect = mock_function_handing
obj = plugins.NotifyMatrix(include_image=True)
assert isinstance(obj, plugins.NotifyMatrix) is True
assert obj.access_token is None
assert obj.notify('test', 'test') is True
obj = plugins.NotifyMatrix(include_image=False)
assert isinstance(obj, plugins.NotifyMatrix) is True
assert obj.access_token is None
assert obj.notify('test', 'test') is True

View File

@ -127,7 +127,7 @@ def test_plugin(mock_refresh, mock_send):
assert(isinstance(obj, instance))
if isinstance(obj, plugins.NotifyBase.NotifyBase):
if isinstance(obj, plugins.NotifyBase):
# We loaded okay; now lets make sure we can reverse this url
assert(isinstance(obj.url(), six.string_types) is True)
@ -137,7 +137,7 @@ def test_plugin(mock_refresh, mock_send):
# Our object should be the same instance as what we had
# originally expected above.
if not isinstance(obj_cmp, plugins.NotifyBase.NotifyBase):
if not isinstance(obj_cmp, plugins.NotifyBase):
# Assert messages are hard to trace back with the way
# these tests work. Just printing before throwing our
# assertion failure makes things easier to debug later on

View File

@ -2195,7 +2195,7 @@ def test_rest_plugins(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Define how many characters exist per line
row = 80
@ -2298,7 +2298,7 @@ def test_rest_plugins(mock_post, mock_get):
assert isinstance(obj, instance) is True
if isinstance(obj, plugins.NotifyBase.NotifyBase):
if isinstance(obj, plugins.NotifyBase):
# We loaded okay; now lets make sure we can reverse this url
assert isinstance(obj.url(), six.string_types) is True
@ -2308,7 +2308,7 @@ def test_rest_plugins(mock_post, mock_get):
# Our object should be the same instance as what we had
# originally expected above.
if not isinstance(obj_cmp, plugins.NotifyBase.NotifyBase):
if not isinstance(obj_cmp, plugins.NotifyBase):
# Assert messages are hard to trace back with the way
# these tests work. Just printing before throwing our
# assertion failure makes things easier to debug later on
@ -2446,7 +2446,7 @@ def test_notify_boxcar_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Generate some generic message types
device = 'A' * 64
@ -2517,7 +2517,7 @@ def test_notify_discord_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Initialize some generic (but valid) tokens
webhook_id = 'A' * 24
@ -2600,7 +2600,7 @@ def test_notify_emby_plugin_login(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Prepare Mock
mock_get.return_value = requests.Request()
@ -2719,7 +2719,7 @@ def test_notify_emby_plugin_sessions(mock_post, mock_get, mock_logout,
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Prepare Mock
mock_get.return_value = requests.Request()
@ -2814,7 +2814,7 @@ def test_notify_twilio_plugin(mock_post):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Prepare our response
response = requests.Request()
@ -2875,7 +2875,7 @@ def test_notify_emby_plugin_logout(mock_post, mock_get, mock_login):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Prepare Mock
mock_get.return_value = requests.Request()
@ -2941,7 +2941,7 @@ def test_notify_emby_plugin_notify(mock_post, mock_get, mock_logout,
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
req = requests.Request()
req.status_code = requests.codes.ok
@ -3014,7 +3014,7 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Initialize some generic (but valid) tokens
webhook_id = 'webhook_id'
@ -3097,6 +3097,19 @@ def test_notify_ifttt_plugin(mock_post, mock_get):
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
# Test removal of tokens as dict
obj = plugins.NotifyIFTTT(
webhook_id=webhook_id, events=events,
add_tokens={
'MyKey': 'MyValue'
},
del_tokens={
plugins.NotifyIFTTT.ifttt_default_title_key: None,
plugins.NotifyIFTTT.ifttt_default_body_key: None,
plugins.NotifyIFTTT.ifttt_default_type_key: None})
assert isinstance(obj, plugins.NotifyIFTTT) is True
@mock.patch('requests.get')
@mock.patch('requests.post')
@ -3106,7 +3119,7 @@ def test_notify_join_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Generate some generic message types
device = 'A' * 32
@ -3140,7 +3153,7 @@ def test_notify_pover_plugin():
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# No token
try:
@ -3158,7 +3171,7 @@ def test_notify_ryver_plugin():
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# must be 15 characters long
token = 'a' * 15
@ -3181,7 +3194,7 @@ def test_notify_slack_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Initialize some generic (but valid) tokens
token_a = 'A' * 9
@ -3230,7 +3243,7 @@ def test_notify_pushbullet_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Initialize some generic (but valid) tokens
accesstoken = 'a' * 32
@ -3275,7 +3288,7 @@ def test_notify_pushed_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Chat ID
recipients = '@ABCDEFG, @DEFGHIJ, #channel, #channel2'
@ -3344,7 +3357,7 @@ def test_notify_pushover_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Initialize some generic (but valid) tokens
token = 'a' * 30
@ -3408,7 +3421,7 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Chat ID
recipients = 'AbcD1245, @l2g, @lead2gold, #channel, #channel2'
@ -3513,7 +3526,7 @@ def test_notify_telegram_plugin(mock_post, mock_get):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Bot Token
bot_token = '123456789:abcdefg_hijklmnop'
@ -3587,7 +3600,7 @@ def test_notify_telegram_plugin(mock_post, mock_get):
# We don't override the title maxlen so we should be set to the same
# as our parent class in this case
assert obj.title_maxlen == plugins.NotifyBase.NotifyBase.title_maxlen
assert obj.title_maxlen == plugins.NotifyBase.title_maxlen
# This tests erroneous messages involving multiple chat ids
assert obj.notify(
@ -3732,7 +3745,7 @@ def test_notify_overflow_truncate():
#
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Number of characters per line
row = 24
@ -3912,7 +3925,7 @@ def test_notify_overflow_split():
#
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Number of characters per line
row = 24

View File

@ -85,7 +85,7 @@ def test_plugin(mock_oauth, mock_api):
"""
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
plugins.NotifyBase.request_rate_per_sec = 0
# Define how many characters exist per line
row = 80
@ -140,7 +140,7 @@ def test_plugin(mock_oauth, mock_api):
assert isinstance(obj, instance) is True
if isinstance(obj, plugins.NotifyBase.NotifyBase):
if isinstance(obj, plugins.NotifyBase):
# We loaded okay; now lets make sure we can reverse this url
assert isinstance(obj.url(), six.string_types) is True
@ -150,7 +150,7 @@ def test_plugin(mock_oauth, mock_api):
# Our object should be the same instance as what we had
# originally expected above.
if not isinstance(obj_cmp, plugins.NotifyBase.NotifyBase):
if not isinstance(obj_cmp, plugins.NotifyBase):
# Assert messages are hard to trace back with the way
# these tests work. Just printing before throwing our
# assertion failure makes things easier to debug later on

View File

@ -25,6 +25,7 @@
from __future__ import print_function
import re
import os
try:
# Python 2.7
from urllib import unquote
@ -616,3 +617,90 @@ def test_exclusive_match():
# www or zzz or abc and jjj
assert utils.is_exclusive_match(
logic=['www', 'zzz', ('abc', 'jjj')], data=data) is False
def test_environ_temporary_change():
"""utils: environ() testing
"""
e_key1 = 'APPRISE_TEMP1'
e_key2 = 'APPRISE_TEMP2'
e_key3 = 'APPRISE_TEMP3'
e_val1 = 'ABCD'
e_val2 = 'DEFG'
e_val3 = 'HIJK'
os.environ[e_key1] = e_val1
os.environ[e_key2] = e_val2
os.environ[e_key3] = e_val3
# Ensure our environment variable stuck
assert e_key1 in os.environ
assert e_val1 in os.environ[e_key1]
assert e_key2 in os.environ
assert e_val2 in os.environ[e_key2]
assert e_key3 in os.environ
assert e_val3 in os.environ[e_key3]
with utils.environ(e_key1, e_key3):
# Eliminates Environment Variable 1 and 3
assert e_key1 not in os.environ
assert e_key2 in os.environ
assert e_val2 in os.environ[e_key2]
assert e_key3 not in os.environ
# after with is over, environment is restored to normal
assert e_key1 in os.environ
assert e_val1 in os.environ[e_key1]
assert e_key2 in os.environ
assert e_val2 in os.environ[e_key2]
assert e_key3 in os.environ
assert e_val3 in os.environ[e_key3]
d_key = 'APPRISE_NOT_SET'
n_key = 'APPRISE_NEW_KEY'
n_val = 'NEW_VAL'
# Verify that our temporary variables (defined above) are not pre-existing
# environemnt variables as we'll be setting them below
assert n_key not in os.environ
assert d_key not in os.environ
# makes it easier to pass in the arguments
updates = {
e_key1: e_val3,
e_key2: e_val1,
n_key: n_val,
}
with utils.environ(d_key, e_key3, **updates):
# Attempt to eliminate an undefined key (silently ignored)
# Eliminates Environment Variable 3
# Environment Variable 1 takes on the value of Env 3
# Environment Variable 2 takes on the value of Env 1
# Set a brand new variable that previously didn't exist
assert e_key1 in os.environ
assert e_val3 in os.environ[e_key1]
assert e_key2 in os.environ
assert e_val1 in os.environ[e_key2]
assert e_key3 not in os.environ
# Can't delete a variable that doesn't exist; so we're in the same
# state here.
assert d_key not in os.environ
# Our temporary variables will be found now
assert n_key in os.environ
assert n_val in os.environ[n_key]
# after with is over, environment is restored to normal
assert e_key1 in os.environ
assert e_val1 in os.environ[e_key1]
assert e_key2 in os.environ
assert e_val2 in os.environ[e_key2]
assert e_key3 in os.environ
assert e_val3 in os.environ[e_key3]
# Even our temporary variables are now missing
assert n_key not in os.environ
assert d_key not in os.environ

View File

@ -108,7 +108,7 @@ def test_xmpp_plugin(tmpdir):
.CA_CERTIFICATE_FILE_LOCATIONS = []
# Disable Throttling to speed testing
apprise.plugins.NotifyBase.NotifyBase.request_rate_per_sec = 0
apprise.plugins.NotifyBase.request_rate_per_sec = 0
# Create our instance
obj = apprise.Apprise.instantiate('xmpp://', suppress_exceptions=False)

View File

@ -11,6 +11,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics
@ -20,6 +21,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics
@ -29,6 +31,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics
@ -38,6 +41,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics
@ -47,6 +51,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics
@ -56,6 +61,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics
@ -64,6 +70,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics
@ -72,6 +79,7 @@ deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
python setup.py compile_catalog
coverage run --parallel -m pytest {posargs}
flake8 . --count --show-source --statistics