extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-11-21 23:53:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

Load Dynamic Libraries and Emoji Engine on Demand (#1020)

Browse Source
This commit is contained in:
Chris Caron 2023-12-27 14:23:13 -05:00 committed by GitHub
parent 34a26da4f4
commit 9dcf769397
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
35 changed files with 1943 additions and 1064 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.github/workflows/tests.yml vendored
View File

@ -31,7 +31,7 @@ jobs:
# expanded using subsequent `include` items.
matrix:
os: ["ubuntu-latest"]
python-version: ["3.10"]
python-version: ["3.11"]
bare: [false]
include:
@ -39,14 +39,14 @@ jobs:
# Within the `bare` environment, `all-plugin-requirements.txt` will NOT be
# installed, to verify the application also works without those dependencies.
- os: "ubuntu-latest"
python-version: "3.10"
python-version: "3.11"
bare: true
# Let's save resources and only build a single slot on macOS- and Windows.
- os: "macos-latest"
python-version: "3.10"
python-version: "3.11"
- os: "windows-latest"
python-version: "3.10"
python-version: "3.11"
# Test more available versions of CPython on Linux.
- os: "ubuntu-20.04"

21
apprise/Apprise.py
View File

@ -33,6 +33,7 @@ from itertools import chain
from . import common
from .conversion import convert_between
from .utils import is_exclusive_match
from .NotificationManager import NotificationManager
from .utils import parse_list
from .utils import parse_urls
from .utils import cwe312_url
@ -45,10 +46,12 @@ from .AppriseLocale import AppriseLocale
from .config.ConfigBase import ConfigBase
from .plugins.NotifyBase import NotifyBase
from . import plugins
from . import __version__
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
class Apprise:
"""
@ -138,7 +141,7 @@ class Apprise:
# We already have our result set
results = url
if results.get('schema') not in common.NOTIFY_SCHEMA_MAP:
if results.get('schema') not in N_MGR:
# 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.')
@ -161,7 +164,7 @@ class Apprise:
type(url))
return None
if not common.NOTIFY_SCHEMA_MAP[results['schema']].enabled:
if not N_MGR[results['schema']].enabled:
#
# First Plugin Enable Check (Pre Initialization)
#
@ -181,13 +184,12 @@ class Apprise:
try:
# Attempt to create an instance of our plugin using the parsed
# URL information
plugin = common.NOTIFY_SCHEMA_MAP[results['schema']](**results)
plugin = N_MGR[results['schema']](**results)
# Create log entry of loaded URL
logger.debug(
'Loaded {} URL: {}'.format(
common.
NOTIFY_SCHEMA_MAP[results['schema']].service_name,
N_MGR[results['schema']].service_name,
plugin.url(privacy=asset.secure_logging)))
except Exception:
@ -198,15 +200,14 @@ class Apprise:
# the arguments are invalid or can not be used.
logger.error(
'Could not load {} URL: {}'.format(
common.
NOTIFY_SCHEMA_MAP[results['schema']].service_name,
N_MGR[results['schema']].service_name,
loggable_url))
return None
else:
# Attempt to create an instance of our plugin using the parsed
# URL information but don't wrap it in a try catch
plugin = common.NOTIFY_SCHEMA_MAP[results['schema']](**results)
plugin = N_MGR[results['schema']](**results)
if not plugin.enabled:
#
@ -690,7 +691,7 @@ class Apprise:
'asset': self.asset.details(),
}
for plugin in set(common.NOTIFY_SCHEMA_MAP.values()):
for plugin in N_MGR.plugins():
# Iterate over our hashed plugins and dynamically build details on
# their status:

8
apprise/AppriseAsset.py
View File

@ -33,7 +33,11 @@ from os.path import dirname
from os.path import isfile
from os.path import abspath
from .common import NotifyType
from .utils import module_detection
from .NotificationManager import NotificationManager
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
class AppriseAsset:
@ -180,7 +184,7 @@ class AppriseAsset:
if plugin_paths:
# Load any decorated modules if defined
module_detection(plugin_paths)
N_MGR.module_detection(plugin_paths)
def color(self, notify_type, color_type=None):
"""

24
apprise/AppriseAttachment.py
View File

@ -26,15 +26,18 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from . import attachment
from . import URLBase
from .attachment.AttachBase import AttachBase
from .AppriseAsset import AppriseAsset
from .AttachmentManager import AttachmentManager
from .logger import logger
from .common import ContentLocation
from .common import CONTENT_LOCATIONS
from .common import ATTACHMENT_SCHEMA_MAP
from .utils import GET_SCHEMA_RE
# Grant access to our Notification Manager Singleton
A_MGR = AttachmentManager()
class AppriseAttachment:
"""
@ -139,7 +142,7 @@ class AppriseAttachment:
# prepare default asset
asset = self.asset
if isinstance(attachments, attachment.AttachBase):
if isinstance(attachments, AttachBase):
# Go ahead and just add our attachments into our list
self.attachments.append(attachments)
return True
@ -169,7 +172,7 @@ class AppriseAttachment:
# returns None if it fails
instance = AppriseAttachment.instantiate(
_attachment, asset=asset, cache=cache)
if not isinstance(instance, attachment.AttachBase):
if not isinstance(instance, AttachBase):
return_status = False
continue
@ -178,7 +181,7 @@ class AppriseAttachment:
# append our content together
instance = _attachment.attachments
elif not isinstance(_attachment, attachment.AttachBase):
elif not isinstance(_attachment, AttachBase):
logger.warning(
"An invalid attachment (type={}) was specified.".format(
type(_attachment)))
@ -228,7 +231,7 @@ class AppriseAttachment:
schema = GET_SCHEMA_RE.match(url)
if schema is None:
# Plan B is to assume we're dealing with a file
schema = attachment.AttachFile.protocol
schema = 'file'
url = '{}://{}'.format(schema, URLBase.quote(url))
else:
@ -236,13 +239,13 @@ class AppriseAttachment:
schema = schema.group('schema').lower()
# Some basic validation
if schema not in ATTACHMENT_SCHEMA_MAP:
if schema not in A_MGR:
logger.warning('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 = ATTACHMENT_SCHEMA_MAP[schema].parse_url(url)
results = A_MGR[schema].parse_url(url)
if not results:
# Failed to parse the server URL
@ -261,8 +264,7 @@ class AppriseAttachment:
try:
# Attempt to create an instance of our plugin using the parsed
# URL information
attach_plugin = \
ATTACHMENT_SCHEMA_MAP[results['schema']](**results)
attach_plugin = A_MGR[results['schema']](**results)
except Exception:
# the arguments are invalid or can not be used.
@ -272,7 +274,7 @@ class AppriseAttachment:
else:
# Attempt to create an instance of our plugin using the parsed
# URL information but don't wrap it in a try catch
attach_plugin = ATTACHMENT_SCHEMA_MAP[results['schema']](**results)
attach_plugin = A_MGR[results['schema']](**results)
return attach_plugin

17
apprise/AppriseConfig.py
View File

@ -29,6 +29,7 @@
from . import config
from . import ConfigBase
from . import CONFIG_FORMATS
from .ConfigurationManager import ConfigurationManager
from . import URLBase
from .AppriseAsset import AppriseAsset
from . import common
@ -37,6 +38,9 @@ from .utils import parse_list
from .utils import is_exclusive_match
from .logger import logger
# Grant access to our Configuration Manager Singleton
C_MGR = ConfigurationManager()
class AppriseConfig:
"""
@ -251,7 +255,7 @@ class AppriseConfig:
logger.debug("Loading raw configuration: {}".format(content))
# Create ourselves a ConfigMemory Object to store our configuration
instance = config.ConfigMemory(
instance = config.ConfigMemory.ConfigMemory(
content=content, format=format, asset=asset, tag=tag,
recursion=recursion, insecure_includes=insecure_includes)
@ -326,7 +330,7 @@ class AppriseConfig:
schema = GET_SCHEMA_RE.match(url)
if schema is None:
# Plan B is to assume we're dealing with a file
schema = config.ConfigFile.protocol
schema = config.ConfigFile.ConfigFile.protocol
url = '{}://{}'.format(schema, URLBase.quote(url))
else:
@ -334,13 +338,13 @@ class AppriseConfig:
schema = schema.group('schema').lower()
# Some basic validation
if schema not in common.CONFIG_SCHEMA_MAP:
if schema not in C_MGR:
logger.warning('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 = common.CONFIG_SCHEMA_MAP[schema].parse_url(url)
results = C_MGR[schema].parse_url(url)
if not results:
# Failed to parse the server URL
@ -368,8 +372,7 @@ class AppriseConfig:
try:
# Attempt to create an instance of our plugin using the parsed
# URL information
cfg_plugin = \
common.CONFIG_SCHEMA_MAP[results['schema']](**results)
cfg_plugin = C_MGR[results['schema']](**results)
except Exception:
# the arguments are invalid or can not be used.
@ -379,7 +382,7 @@ class AppriseConfig:
else:
# Attempt to create an instance of our plugin using the parsed
# URL information but don't wrap it in a try catch
cfg_plugin = common.CONFIG_SCHEMA_MAP[results['schema']](**results)
cfg_plugin = C_MGR[results['schema']](**results)
return cfg_plugin

54
apprise/AttachmentManager.py Normal file
View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from os.path import dirname
from os.path import abspath
from os.path import join
from .manager import PluginManager
class AttachmentManager(PluginManager):
"""
Designed to be a singleton object to maintain all initialized
attachment plugins/modules in memory.
"""
# Description (used for logging)
name = 'Attachment Plugin'
# Filename Prefix to filter on
fname_prefix = 'Attach'
# Memory Space
_id = 'attachment'
# Our Module Python path name
module_name_prefix = f'apprise.{_id}'
# The module path to scan
module_path = join(abspath(dirname(__file__)), _id)

54
apprise/ConfigurationManager.py Normal file
View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from os.path import dirname
from os.path import abspath
from os.path import join
from .manager import PluginManager
class ConfigurationManager(PluginManager):
"""
Designed to be a singleton object to maintain all initialized
configuration plugins/modules in memory.
"""
# Description (used for logging)
name = 'Configuration Plugin'
# Filename Prefix to filter on
fname_prefix = 'Config'
# Memory Space
_id = 'config'
# Our Module Python path name
module_name_prefix = f'apprise.{_id}'
# The module path to scan
module_path = join(abspath(dirname(__file__)), _id)

54
apprise/NotificationManager.py Normal file
View File

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from os.path import dirname
from os.path import abspath
from os.path import join
from .manager import PluginManager
class NotificationManager(PluginManager):
"""
Designed to be a singleton object to maintain all initialized notifications
in memory.
"""
# Description (used for logging)
name = 'Notification Plugin'
# Filename Prefix to filter on
fname_prefix = 'Notify'
# Memory Space
_id = 'plugins'
# Our Module Python path name
module_name_prefix = f'apprise.{_id}'
# The module path to scan
module_path = join(abspath(dirname(__file__)), _id)

97
apprise/attachment/__init__.py
View File

@ -26,93 +26,14 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import re
from os import listdir
from os.path import dirname
from os.path import abspath
from ..common import ATTACHMENT_SCHEMA_MAP
# Used for testing
from .AttachBase import AttachBase
from ..AttachmentManager import AttachmentManager
__all__ = []
# Initalize our Attachment Manager Singleton
A_MGR = AttachmentManager()
# Load our Lookup Matrix
def __load_matrix(path=abspath(dirname(__file__)), name='apprise.attachment'):
"""
Dynamically load our schema map; this allows us to gracefully
skip over modules we simply don't have the dependencies for.
"""
# Used for the detection of additional Attachment Services objects
# The .py extension is optional as we support loading directories too
module_re = re.compile(r'^(?P<name>Attach[a-z0-9]+)(\.py)?$', re.I)
for f in listdir(path):
match = module_re.match(f)
if not match:
# keep going
continue
# Store our notification/plugin name:
plugin_name = match.group('name')
try:
module = __import__(
'{}.{}'.format(name, plugin_name),
globals(), locals(),
fromlist=[plugin_name])
except ImportError:
# No problem, we can't use this object
continue
if not hasattr(module, plugin_name):
# Not a library we can load as it doesn't follow the simple rule
# that the class must bear the same name as the notification
# file itself.
continue
# Get our plugin
plugin = getattr(module, plugin_name)
if not hasattr(plugin, 'app_id'):
# Filter out non-notification modules
continue
elif plugin_name in __all__:
# we're already handling this object
continue
# 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:
globals()[plugin_name] = plugin
# Load protocol(s) if defined
proto = getattr(plugin, 'protocol', None)
if isinstance(proto, str):
if proto not in ATTACHMENT_SCHEMA_MAP:
ATTACHMENT_SCHEMA_MAP[proto] = plugin
elif isinstance(proto, (set, list, tuple)):
# Support iterables list types
for p in proto:
if p not in ATTACHMENT_SCHEMA_MAP:
ATTACHMENT_SCHEMA_MAP[p] = plugin
# Load secure protocol(s) if defined
protos = getattr(plugin, 'secure_protocol', None)
if isinstance(protos, str):
if protos not in ATTACHMENT_SCHEMA_MAP:
ATTACHMENT_SCHEMA_MAP[protos] = plugin
if isinstance(protos, (set, list, tuple)):
# Support iterables list types
for p in protos:
if p not in ATTACHMENT_SCHEMA_MAP:
ATTACHMENT_SCHEMA_MAP[p] = plugin
return ATTACHMENT_SCHEMA_MAP
# Dynamically build our schema base
__load_matrix()
__all__ = [
# Reference
'AttachBase',
]

44
apprise/common.py
View File

@ -26,50 +26,6 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
# 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
# It is also used as a means of preventing a module from being reloaded
# in the event it already exists
NOTIFY_MODULE_MAP = {}
# Maintains a mapping of all of the Notification services
NOTIFY_SCHEMA_MAP = {}
# This contains a mapping of all plugins dynamicaly loaded at runtime from
# external modules such as the @notify decorator
#
# The elements here will be additionally added to the NOTIFY_SCHEMA_MAP if
# there is no conflict otherwise.
# The structure looks like the following:
# Module path, e.g. /usr/share/apprise/plugins/my_notify_hook.py
# {
# 'path': path,
#
# 'notify': {
# 'schema': {
# 'name': 'Custom schema name',
# 'fn_name': 'name_of_function_decorator_was_found_on',
# 'url': 'schema://any/additional/info/found/on/url'
# 'plugin': <CustomNotifyWrapperPlugin>
# },
# 'schema2': {
# 'name': 'Custom schema name',
# 'fn_name': 'name_of_function_decorator_was_found_on',
# 'url': 'schema://any/additional/info/found/on/url'
# 'plugin': <CustomNotifyWrapperPlugin>
# }
# }
#
# Note: that the <CustomNotifyWrapperPlugin> inherits from
# NotifyBase
NOTIFY_CUSTOM_MODULE_MAP = {}
# Maintains a mapping of all configuration schema's supported
CONFIG_SCHEMA_MAP = {}
# Maintains a mapping of all attachment schema's supported
ATTACHMENT_SCHEMA_MAP = {}
class NotifyType:
"""

38
apprise/config/ConfigBase.py
View File

@ -35,16 +35,24 @@ from .. import plugins
from .. import common
from ..AppriseAsset import AppriseAsset
from ..URLBase import URLBase
from ..ConfigurationManager import ConfigurationManager
from ..utils import GET_SCHEMA_RE
from ..utils import parse_list
from ..utils import parse_bool
from ..utils import parse_urls
from ..utils import cwe312_url
from ..NotificationManager import NotificationManager
# Test whether token is valid or not
VALID_TOKEN = re.compile(
r'(?P<token>[a-z0-9][a-z0-9_]+)', re.I)
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
# Grant access to our Configuration Manager Singleton
C_MGR = ConfigurationManager()
class ConfigBase(URLBase):
"""
@ -229,7 +237,7 @@ class ConfigBase(URLBase):
schema = schema.group('schema').lower()
# Some basic validation
if schema not in common.CONFIG_SCHEMA_MAP:
if schema not in C_MGR:
ConfigBase.logger.warning(
'Unsupported include schema {}.'.format(schema))
continue
@ -240,7 +248,7 @@ class ConfigBase(URLBase):
# Parse our url details of the server object as dictionary
# containing all of the information parsed from our URL
results = common.CONFIG_SCHEMA_MAP[schema].parse_url(url)
results = C_MGR[schema].parse_url(url)
if not results:
# Failed to parse the server URL
self.logger.warning(
@ -248,11 +256,10 @@ class ConfigBase(URLBase):
continue
# Handle cross inclusion based on allow_cross_includes rules
if (common.CONFIG_SCHEMA_MAP[schema].allow_cross_includes ==
if (C_MGR[schema].allow_cross_includes ==
common.ContentIncludeMode.STRICT
and schema not in self.schemas()
and not self.insecure_includes) or \
common.CONFIG_SCHEMA_MAP[schema] \
and not self.insecure_includes) or C_MGR[schema] \
.allow_cross_includes == \
common.ContentIncludeMode.NEVER:
@ -280,8 +287,7 @@ class ConfigBase(URLBase):
try:
# Attempt to create an instance of our plugin using the
# parsed URL information
cfg_plugin = \
common.CONFIG_SCHEMA_MAP[results['schema']](**results)
cfg_plugin = C_MGR[results['schema']](**results)
except Exception as e:
# the arguments are invalid or can not be used.
@ -765,8 +771,7 @@ class ConfigBase(URLBase):
try:
# Attempt to create an instance of our plugin using the
# parsed URL information
plugin = common.NOTIFY_SCHEMA_MAP[
results['schema']](**results)
plugin = N_MGR[results['schema']](**results)
# Create log entry of loaded URL
ConfigBase.logger.debug(
@ -1094,7 +1099,7 @@ class ConfigBase(URLBase):
del entries['schema']
# support our special tokens (if they're present)
if schema in common.NOTIFY_SCHEMA_MAP:
if schema in N_MGR:
entries = ConfigBase._special_token_handler(
schema, entries)
@ -1106,7 +1111,7 @@ class ConfigBase(URLBase):
elif isinstance(tokens, dict):
# support our special tokens (if they're present)
if schema in common.NOTIFY_SCHEMA_MAP:
if schema in N_MGR:
tokens = ConfigBase._special_token_handler(
schema, tokens)
@ -1140,7 +1145,7 @@ class ConfigBase(URLBase):
# Grab our first item
_results = results.pop(0)
if _results['schema'] not in common.NOTIFY_SCHEMA_MAP:
if _results['schema'] not in N_MGR:
# the arguments are invalid or can not be used.
ConfigBase.logger.warning(
'An invalid Apprise schema ({}) in YAML configuration '
@ -1213,8 +1218,7 @@ class ConfigBase(URLBase):
try:
# Attempt to create an instance of our plugin using the
# parsed URL information
plugin = common.\
NOTIFY_SCHEMA_MAP[results['schema']](**results)
plugin = N_MGR[results['schema']](**results)
# Create log entry of loaded URL
ConfigBase.logger.debug(
@ -1267,8 +1271,7 @@ class ConfigBase(URLBase):
# Create a copy of our dictionary
tokens = tokens.copy()
for kw, meta in common.NOTIFY_SCHEMA_MAP[schema]\
.template_kwargs.items():
for kw, meta in N_MGR[schema].template_kwargs.items():
# Determine our prefix:
prefix = meta.get('prefix', '+')
@ -1311,8 +1314,7 @@ class ConfigBase(URLBase):
#
# This function here allows these mappings to take place within the
# YAML file as independant arguments.
class_templates = \
plugins.details(common.NOTIFY_SCHEMA_MAP[schema])
class_templates = plugins.details(N_MGR[schema])
for key in list(tokens.keys()):

88
apprise/config/__init__.py
View File

@ -26,84 +26,14 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import re
from os import listdir
from os.path import dirname
from os.path import abspath
from ..logger import logger
from ..common import CONFIG_SCHEMA_MAP
# Used for testing
from .ConfigBase import ConfigBase
from ..ConfigurationManager import ConfigurationManager
__all__ = []
# Initalize our Config Manager Singleton
C_MGR = ConfigurationManager()
# Load our Lookup Matrix
def __load_matrix(path=abspath(dirname(__file__)), name='apprise.config'):
"""
Dynamically load our schema map; this allows us to gracefully
skip over modules we simply don't have the dependencies for.
"""
# Used for the detection of additional Configuration Services objects
# The .py extension is optional as we support loading directories too
module_re = re.compile(r'^(?P<name>Config[a-z0-9]+)(\.py)?$', re.I)
for f in listdir(path):
match = module_re.match(f)
if not match:
# keep going
continue
# Store our notification/plugin name:
plugin_name = match.group('name')
try:
module = __import__(
'{}.{}'.format(name, plugin_name),
globals(), locals(),
fromlist=[plugin_name])
except ImportError:
# No problem, we can't use this object
continue
if not hasattr(module, plugin_name):
# Not a library we can load as it doesn't follow the simple rule
# that the class must bear the same name as the notification
# file itself.
continue
# Get our plugin
plugin = getattr(module, plugin_name)
if not hasattr(plugin, 'app_id'):
# Filter out non-notification modules
continue
elif plugin_name in __all__:
# we're already handling this object
continue
# 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:
globals()[plugin_name] = plugin
fn = getattr(plugin, 'schemas', None)
schemas = set([]) if not callable(fn) else fn(plugin)
# map our schema to our plugin
for schema in schemas:
if schema in CONFIG_SCHEMA_MAP:
logger.error(
"Config schema ({}) mismatch detected - {} to {}"
.format(schema, CONFIG_SCHEMA_MAP[schema], plugin))
continue
# Assign plugin
CONFIG_SCHEMA_MAP[schema] = plugin
return CONFIG_SCHEMA_MAP
# Dynamically build our schema base
__load_matrix()
__all__ = [
# Reference
'ConfigBase',
]

63
apprise/decorators/CustomNotifyPlugin.py
View File

@ -28,6 +28,7 @@
# POSSIBILITY OF SUCH DAMAGE.
from ..plugins.NotifyBase import NotifyBase
from ..NotificationManager import NotificationManager
from ..utils import URL_DETAILS_RE
from ..utils import parse_url
from ..utils import url_assembly
@ -36,6 +37,9 @@ from .. import common
from ..logger import logger
import inspect
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
class CustomNotifyPlugin(NotifyBase):
"""
@ -91,17 +95,17 @@ class CustomNotifyPlugin(NotifyBase):
logger.warning(msg)
return None
# Acquire our plugin name
plugin_name = re_match.group('schema').lower()
# Acquire our schema
schema = re_match.group('schema').lower()
if not re_match.group('base'):
url = '{}://'.format(plugin_name)
url = '{}://'.format(schema)
# Keep a default set of arguments to apply to all called references
base_args = parse_url(
url, default_schema=plugin_name, verify_host=False, simple=True)
url, default_schema=schema, verify_host=False, simple=True)
if plugin_name in common.NOTIFY_SCHEMA_MAP:
if schema in N_MGR:
# we're already handling this object
msg = 'The schema ({}) is already defined and could not be ' \
'loaded from custom notify function {}.' \
@ -117,10 +121,10 @@ class CustomNotifyPlugin(NotifyBase):
# Our Service Name
service_name = name if isinstance(name, str) \
and name else 'Custom - {}'.format(plugin_name)
and name else 'Custom - {}'.format(schema)
# Store our matched schema
secure_protocol = plugin_name
secure_protocol = schema
requirements = {
# Define our required packaging in order to work
@ -181,51 +185,26 @@ class CustomNotifyPlugin(NotifyBase):
# Unhandled Exception
self.logger.warning(
'An exception occured sending a %s notification.',
common.
NOTIFY_SCHEMA_MAP[self.secure_protocol].service_name)
N_MGR[self.secure_protocol].service_name)
self.logger.debug(
'%s Exception: %s',
common.NOTIFY_SCHEMA_MAP[self.secure_protocol], str(e))
N_MGR[self.secure_protocol], str(e))
return False
if response:
self.logger.info(
'Sent %s notification.',
common.
NOTIFY_SCHEMA_MAP[self.secure_protocol].service_name)
N_MGR[self.secure_protocol].service_name)
else:
self.logger.warning(
'Failed to send %s notification.',
common.
NOTIFY_SCHEMA_MAP[self.secure_protocol].service_name)
N_MGR[self.secure_protocol].service_name)
return response
# Store our plugin into our core map file
common.NOTIFY_SCHEMA_MAP[plugin_name] = CustomNotifyPluginWrapper
# Update our custom plugin map
module_pyname = str(send_func.__module__)
if module_pyname not in common.NOTIFY_CUSTOM_MODULE_MAP:
# Support non-dynamic includes as well...
common.NOTIFY_CUSTOM_MODULE_MAP[module_pyname] = {
'path': inspect.getfile(send_func),
# Initialize our template
'notify': {},
}
common.\
NOTIFY_CUSTOM_MODULE_MAP[module_pyname]['notify'][plugin_name] = {
# Our Serivice Description (for API and CLI --details view)
'name': CustomNotifyPluginWrapper.service_name,
# The name of the send function the @notify decorator wrapped
'fn_name': send_func.__name__,
# The URL that was provided in the @notify decorator call
# associated with the 'on='
'url': url,
# The Initialized Plugin that was generated based on the above
# parameters
'plugin': CustomNotifyPluginWrapper}
# return our plugin
return common.NOTIFY_SCHEMA_MAP[plugin_name]
return N_MGR.add(
plugin=CustomNotifyPluginWrapper,
schemas=schema,
send_func=send_func,
url=url,
)

19
apprise/emojis.py
View File

@ -27,6 +27,8 @@
# POSSIBILITY OF SUCH DAMAGE.
import re
import time
from .logger import logger
# All Emoji's are wrapped in this character
DELIM = ':'
@ -2242,10 +2244,8 @@ EMOJI_MAP = {
DELIM + r'wales' + DELIM: '🏴󠁧󠁢󠁷󠁬󠁳󠁿',
}
# Our compiled mapping
EMOJI_COMPILED_MAP = re.compile(
r'(' + '|'.join(EMOJI_MAP.keys()) + r')',
re.IGNORECASE)
# Define our singlton
EMOJI_COMPILED_MAP = None
def apply_emojis(content):
@ -2254,6 +2254,17 @@ def apply_emojis(content):
utf-8 encoded mapping
"""
global EMOJI_COMPILED_MAP
if EMOJI_COMPILED_MAP is None:
t_start = time.time()
# Perform our compilation
EMOJI_COMPILED_MAP = re.compile(
r'(' + '|'.join(EMOJI_MAP.keys()) + r')',
re.IGNORECASE)
logger.trace(
'Emoji engine loaded in {:.4f}s'.format((time.time() - t_start)))
try:
return EMOJI_COMPILED_MAP.sub(lambda x: EMOJI_MAP[x.group()], content)

718
apprise/manager.py Normal file
View File

@ -0,0 +1,718 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import os
import re
import sys
import time
import hashlib
import inspect
from .utils import import_module
from .utils import Singleton
from .utils import parse_list
from os.path import dirname
from os.path import abspath
from os.path import join
from .logger import logger
class PluginManager(metaclass=Singleton):
"""
Designed to be a singleton object to maintain all initialized loading
of modules in memory.
"""
# Description (used for logging)
name = 'Singleton Plugin'
# Memory Space
_id = 'undefined'
# Our Module Python path name
module_name_prefix = f'apprise.{_id}'
# The module path to scan
module_path = join(abspath(dirname(__file__)), _id)
def __init__(self, *args, **kwargs):
"""
Over-ride our class instantiation to provide a singleton
"""
self._module_map = None
self._schema_map = None
# This contains a mapping of all plugins dynamicaly loaded at runtime
# from external modules such as the @notify decorator
#
# The elements here will be additionally added to the _schema_map if
# there is no conflict otherwise.
# The structure looks like the following:
# Module path, e.g. /usr/share/apprise/plugins/my_notify_hook.py
# {
# 'path': path,
#
# 'notify': {
# 'schema': {
# 'name': 'Custom schema name',
# 'fn_name': 'name_of_function_decorator_was_found_on',
# 'url': 'schema://any/additional/info/found/on/url'
# 'plugin': <CustomNotifyWrapperPlugin>
# },
# 'schema2': {
# 'name': 'Custom schema name',
# 'fn_name': 'name_of_function_decorator_was_found_on',
# 'url': 'schema://any/additional/info/found/on/url'
# 'plugin': <CustomNotifyWrapperPlugin>
# }
# }
# Note: that the <CustomNotifyWrapperPlugin> inherits from
# NotifyBase
self._custom_module_map = {}
# Track manually disabled modules (by their schema)
self._disabled = set()
# Hash of all paths previously scanned so we don't waste
# effort/overhead doing it again
self._paths_previously_scanned = set()
def unload_modules(self, disable_native=False):
"""
Reset our object and unload all modules
"""
if self._custom_module_map:
# Handle Custom Module Assignments
for meta in self._custom_module_map.values():
if meta['name'] not in self._module_map:
# Nothing to remove
continue
# For the purpose of tidying up un-used modules in memory
loaded = [m for m in sys.modules.keys()
if m.startswith(
self._module_map[meta['name']]['path'])]
for module_path in loaded:
del sys.modules[module_path]
# Reset disabled plugins (if any)
for schema in self._disabled:
self._schema_map[schema].enabled = True
self._disabled.clear()
# Reset our variables
self._module_map = None if not disable_native else {}
self._schema_map = {}
self._custom_module_map = {}
# Reset our path cache
self._paths_previously_scanned = set()
def load_modules(self, path=None, name=None):
"""
Load our modules into memory
"""
# Default value
module_name_prefix = self.module_name_prefix if name is None else name
module_path = self.module_path if path is None else path
if not self:
# Initialize our maps
self._module_map = {}
self._schema_map = {}
self._custom_module_map = {}
# Used for the detection of additional Notify Services objects
# The .py extension is optional as we support loading directories too
module_re = re.compile(
r'^(?P<name>' + self.fname_prefix + r'[a-z0-9]+)(\.py)?$', re.I)
t_start = time.time()
for f in os.listdir(module_path):
tl_start = time.time()
match = module_re.match(f)
if not match:
# keep going
continue
elif match.group('name') == f'{self.fname_prefix}Base':
# keep going
continue
# Store our notification/plugin name:
module_name = match.group('name')
module_pyname = '{}.{}'.format(module_name_prefix, module_name)
if module_name in self._module_map:
logger.warning(
"%s(s) (%s) already loaded; ignoring %s",
self.name, module_name, os.path.join(module_path, f))
continue
try:
module = __import__(
module_pyname,
globals(), locals(),
fromlist=[module_name])
except ImportError:
# No problem, we can try again another way...
module = import_module(
os.path.join(module_path, f), module_pyname)
if not module:
# logging found in import_module and not needed here
continue
if not hasattr(module, module_name):
# Not a library we can load as it doesn't follow the simple
# rule that the class must bear the same name as the
# notification file itself.
logger.trace(
"%s (%s) import failed; no filename/Class "
"match found in %s",
self.name, module_name, os.path.join(module_path, f))
continue
# Get our plugin
plugin = getattr(module, module_name)
if not hasattr(plugin, 'app_id'):
# Filter out non-notification modules
logger.trace(
"(%s) import failed; no app_id defined in %s",
self.name, module_name, os.path.join(module_path, f))
continue
# Add our plugin name to our module map
self._module_map[module_name] = {
'plugin': set([plugin]),
'module': module,
'path': '{}.{}'.format(module_name_prefix, module_name),
'native': True,
}
fn = getattr(plugin, 'schemas', None)
schemas = set([]) if not callable(fn) else fn(plugin)
# map our schema to our plugin
for schema in schemas:
if schema in self._schema_map:
logger.error(
"{} schema ({}) mismatch detected - {} to {}"
.format(self.name, schema, self._schema_map, plugin))
continue
# Assign plugin
self._schema_map[schema] = plugin
logger.trace(
'{} {} loaded in {:.6f}s'.format(
self.name, module_name, (time.time() - tl_start)))
logger.debug(
'{} {}(s) and {} Schema(s) loaded in {:.4f}s'
.format(
self.name, len(self._module_map), len(self._schema_map),
(time.time() - t_start)))
def module_detection(self, paths, cache=True):
"""
Leverage the @notify decorator and load all objects found matching
this.
"""
# A simple restriction that we don't allow periods in the filename at
# all so it can't be hidden (Linux OS's) and it won't conflict with
# Python path naming. This also prevents us from loading any python
# file that starts with an underscore or dash
# We allow for __init__.py as well
module_re = re.compile(
r'^(?P<name>[_a-z0-9][a-z0-9._-]+)?(\.py)?$', re.I)
# Validate if we're a loadable Python file or not
valid_python_file_re = re.compile(r'.+\.py(o|c)?$', re.IGNORECASE)
if isinstance(paths, str):
paths = [paths, ]
if not paths or not isinstance(paths, (tuple, list)):
# We're done
return
def _import_module(path):
# Since our plugin name can conflict (as a module) with another
# we want to generate random strings to avoid steping on
# another's namespace
if not (path and valid_python_file_re.match(path)):
# Ignore file/module type
logger.trace('Plugin Scan: Skipping %s', path)
return
t_start = time.time()
module_name = hashlib.sha1(path.encode('utf-8')).hexdigest()
module_pyname = "{prefix}.{name}".format(
prefix='apprise.custom.module', name=module_name)
if module_pyname in self._custom_module_map:
# First clear out existing entries
for schema in \
self._custom_module_map[module_pyname]['notify']\
.keys():
# Remove any mapped modules to this file
del self._schema_map[schema]
# Reset
del self._custom_module_map[module_pyname]
# Load our module
module = import_module(path, module_pyname)
if not module:
# No problem, we can't use this object
logger.warning('Failed to load custom module: %s', _path)
return
# Print our loaded modules if any
if module_pyname in self._custom_module_map:
logger.debug(
'Custom module %s - %d schema(s) (name=%s) '
'loaded in {:.6f}s',
_path, module_name,
len(self._custom_module_map[module_pyname]['notify']),
(time.time() - t_start))
# Add our plugin name to our module map
self._module_map[module_name] = {
'plugin': set(),
'module': module,
'path': module_pyname,
'native': False,
}
for schema, meta in\
self._custom_module_map[module_pyname]['notify']\
.items():
# For mapping purposes; map our element in our main list
self._module_map[module_name]['plugin'].add(
self._schema_map[schema])
# Log our success
logger.info('Loaded custom notification: %s://', schema)
else:
# The code reaches here if we successfully loaded the Python
# module but no hooks/triggers were found. So we can safely
# just remove/ignore this entry
del sys.modules[module_pyname]
return
# end of _import_module()
return
for _path in paths:
path = os.path.abspath(os.path.expanduser(_path))
if (cache and path in self._paths_previously_scanned) \
or not os.path.exists(path):
# We're done as we've already scanned this
continue
# Store our path as a way of hashing it has been handled
self._paths_previously_scanned.add(path)
if os.path.isdir(path) and not \
os.path.isfile(os.path.join(path, '__init__.py')):
logger.debug('Scanning for custom plugins in: %s', path)
for entry in os.listdir(path):
re_match = module_re.match(entry)
if not re_match:
# keep going
logger.trace('Plugin Scan: Ignoring %s', entry)
continue
new_path = os.path.join(path, entry)
if os.path.isdir(new_path):
# Update our path
new_path = os.path.join(path, entry, '__init__.py')
if not os.path.isfile(new_path):
logger.trace(
'Plugin Scan: Ignoring %s',
os.path.join(path, entry))
continue
if not cache or \
(cache and
new_path not in self._paths_previously_scanned):
# Load our module
_import_module(new_path)
# Add our subdir path
self._paths_previously_scanned.add(new_path)
else:
if os.path.isdir(path):
# This logic is safe to apply because we already validated
# the directories state above; update our path
path = os.path.join(path, '__init__.py')
if cache and path in self._paths_previously_scanned:
continue
self._paths_previously_scanned.add(path)
# directly load as is
re_match = module_re.match(os.path.basename(path))
# must be a match and must have a .py extension
if not re_match or not re_match.group(1):
# keep going
logger.trace('Plugin Scan: Ignoring %s', path)
continue
# Load our module
_import_module(path)
return None
def add(self, plugin, schemas=None, url=None, send_func=None):
"""
Ability to manually add Notification services to our stack
"""
if not self:
# Lazy load
self.load_modules()
# Acquire a list of schemas
p_schemas = parse_list(plugin.secure_protocol, plugin.protocol)
if isinstance(schemas, str):
schemas = [schemas, ]
elif schemas is None:
# Default
schemas = p_schemas
if not schemas or not isinstance(schemas, (set, tuple, list)):
# We're done
logger.error(
'The schemas provided (type %s) is unsupported; '
'loaded from %s.',
type(schemas),
send_func.__name__ if send_func else plugin.__class__.__name__)
return False
# Convert our schemas into a set
schemas = set([s.lower() for s in schemas]) | set(p_schemas)
# Valdation
conflict = [s for s in schemas if s in self]
if conflict:
# we're already handling this schema
logger.warning(
'The schema(s) (%s) are already defined and could not be '
'loaded from %s%s.',
', '.join(conflict),
'custom notify function ' if send_func else '',
send_func.__name__ if send_func else plugin.__class__.__name__)
return False
if send_func:
# Acquire the function name
fn_name = send_func.__name__
# Acquire the python filename path
path = inspect.getfile(send_func)
# Acquire our path to our module
module_name = str(send_func.__module__)
if module_name not in self._custom_module_map:
# Support non-dynamic includes as well...
self._custom_module_map[module_name] = {
# Name can be useful for indexing back into the
# _module_map object; this is the key to do it with:
'name': module_name.split('.')[-1],
# The path to the module loaded
'path': path,
# Initialize our template
'notify': {},
}
for schema in schemas:
self._custom_module_map[module_name]['notify'][schema] = {
# The name of the send function the @notify decorator
# wrapped
'fn_name': fn_name,
# The URL that was provided in the @notify decorator call
# associated with the 'on='
'url': url,
}
else:
module_name = hashlib.sha1(
''.join(schemas).encode('utf-8')).hexdigest()
module_pyname = "{prefix}.{name}".format(
prefix='apprise.adhoc.module', name=module_name)
# Add our plugin name to our module map
self._module_map[module_name] = {
'plugin': set([plugin]),
'module': None,
'path': module_pyname,
'native': False,
}
for schema in schemas:
# Assign our mapping
self._schema_map[schema] = plugin
return True
def remove(self, *schemas):
"""
Removes a loaded element (if defined)
"""
if not self:
# Lazy load
self.load_modules()
for schema in schemas:
try:
del self[schema]
except KeyError:
pass
def plugins(self, include_disabled=True):
"""
Return all of our loaded plugins
"""
if not self:
# Lazy load
self.load_modules()
for module in self._module_map.values():
for plugin in module['plugin']:
if not include_disabled and not plugin.enabled:
continue
yield plugin
def schemas(self, include_disabled=True):
"""
Return all of our loaded schemas
if include_disabled == True, then even disabled notifications are
returned
"""
if not self:
# Lazy load
self.load_modules()
# Return our list
return list(self._schema_map.keys()) if include_disabled else \
[s for s in self._schema_map.keys() if self._schema_map[s].enabled]
def disable(self, *schemas):
"""
Disables the modules associated with the specified schemas
"""
if not self:
# Lazy load
self.load_modules()
for schema in schemas:
if schema not in self._schema_map:
continue
if not self._schema_map[schema].enabled:
continue
# Disable
self._schema_map[schema].enabled = False
self._disabled.add(schema)
def enable_only(self, *schemas):
"""
Disables the modules associated with the specified schemas
"""
if not self:
# Lazy load
self.load_modules()
# convert to set for faster indexing
schemas = set(schemas)
for plugin in self.plugins():
# Get our plugin's schema list
p_schemas = set(
parse_list(plugin.secure_protocol, plugin.protocol))
if not schemas & p_schemas:
if plugin.enabled:
# Disable it (only if previously enabled); this prevents us
# from adjusting schemas that were disabled due to missing
# libraries or other environment reasons
plugin.enabled = False
self._disabled |= p_schemas
continue
# If we reach here, our schema was flagged to be enabled
if p_schemas & self._disabled:
# Previously disabled; no worries, let's clear this up
self._disabled -= p_schemas
plugin.enabled = True
def __contains__(self, schema):
"""
Checks if a schema exists
"""
if not self:
# Lazy load
self.load_modules()
return schema in self._schema_map
def __delitem__(self, schema):
if not self:
# Lazy load
self.load_modules()
# Get our plugin (otherwise we throw a KeyError) which is
# intended on del action that doesn't align
plugin = self._schema_map[schema]
# Our list of all schema entries
p_schemas = set([schema])
for key in list(self._module_map.keys()):
if plugin in self._module_map[key]['plugin']:
# Remove our plugin
self._module_map[key]['plugin'].remove(plugin)
# Custom Plugin Entry; Clean up cross reference
module_pyname = self._module_map[key]['path']
if not self._module_map[key]['native'] and \
module_pyname in self._custom_module_map:
del self.\
_custom_module_map[module_pyname]['notify'][schema]
if not self._custom_module_map[module_pyname]['notify']:
#
# Last custom loaded element
#
# Free up custom object entry
del self._custom_module_map[module_pyname]
if not self._module_map[key]['plugin']:
#
# Last element
#
if self._module_map[key]['native']:
# Get our plugin's schema list
p_schemas = \
set([s for s in parse_list(
plugin.secure_protocol, plugin.protocol)
if s in self._schema_map])
# free system memory
if self._module_map[key]['module']:
del sys.modules[self._module_map[key]['path']]
# free last remaining pointer in module map
del self._module_map[key]
for schema in p_schemas:
# Final Tidy
del self._schema_map[schema]
def __setitem__(self, schema, plugin):
"""
Support fast assigning of Plugin/Notification Objects
"""
if not self:
# Lazy load
self.load_modules()
# Set default values if not otherwise set
if not plugin.service_name:
# Assign service name if one doesn't exist
plugin.service_name = f'{schema}://'
p_schemas = set(
parse_list(plugin.secure_protocol, plugin.protocol))
if not p_schemas:
# Assign our protocol
plugin.secure_protocol = schema
p_schemas.add(schema)
elif schema not in p_schemas:
# Add our others (if defined)
plugin.secure_protocol = \
set([schema] + parse_list(plugin.secure_protocol))
p_schemas.add(schema)
if not self.add(plugin, schemas=p_schemas):
raise KeyError('Conflicting Assignment')
def __getitem__(self, schema):
"""
Returns the indexed plugin identified by the schema specified
"""
if not self:
# Lazy load
self.load_modules()
return self._schema_map[schema]
def __iter__(self):
"""
Returns an iterator so we can iterate over our loaded modules
"""
if not self:
# Lazy load
self.load_modules()
return iter(self._module_map.values())
def __len__(self):
"""
Returns the number of modules/plugins loaded
"""
if not self:
# Lazy load
self.load_modules()
return len(self._module_map)
def __bool__(self):
"""
Determines if object has loaded or not
"""
return True if self._module_map is not None else False

1
apprise/plugins/NotifyMacOSX.py
View File

@ -40,7 +40,6 @@ from ..AppriseLocale import gettext_lazy as _
NOTIFY_MACOSX_SUPPORT_ENABLED = False
# TODO: The module will be easier to test without module-level code.
if platform.system() == 'Darwin':
# Check this is Mac OS X 10.8, or higher
major, minor = platform.mac_ver()[0].split('.')[:2]

127
apprise/plugins/__init__.py
View File

@ -27,12 +27,8 @@
# POSSIBILITY OF SUCH DAMAGE.
import os
import re
import copy
from os.path import dirname
from os.path import abspath
# Used for testing
from .NotifyBase import NotifyBase
@ -40,13 +36,17 @@ from ..common import NotifyImageSize
from ..common import NOTIFY_IMAGE_SIZES
from ..common import NotifyType
from ..common import NOTIFY_TYPES
from .. import common
from ..utils import parse_list
from ..utils import cwe312_url
from ..utils import GET_SCHEMA_RE
from ..logger import logger
from ..AppriseLocale import gettext_lazy as _
from ..AppriseLocale import LazyTranslation
from ..NotificationManager import NotificationManager
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
__all__ = [
# Reference
@ -58,101 +58,6 @@ __all__ = [
]
# Load our Lookup Matrix
def __load_matrix(path=abspath(dirname(__file__)), name='apprise.plugins'):
"""
Dynamically load our schema map; this allows us to gracefully
skip over modules we simply don't have the dependencies for.
"""
# Used for the detection of additional Notify Services objects
# The .py extension is optional as we support loading directories too
module_re = re.compile(r'^(?P<name>Notify[a-z0-9]+)(\.py)?$', re.I)
for f in os.listdir(path):
match = module_re.match(f)
if not match:
# keep going
continue
# Store our notification/plugin name:
plugin_name = match.group('name')
try:
module = __import__(
'{}.{}'.format(name, plugin_name),
globals(), locals(),
fromlist=[plugin_name])
except ImportError:
# No problem, we can't use this object
continue
if not hasattr(module, plugin_name):
# Not a library we can load as it doesn't follow the simple rule
# that the class must bear the same name as the notification
# file itself.
continue
# Get our plugin
plugin = getattr(module, plugin_name)
if not hasattr(plugin, 'app_id'):
# Filter out non-notification modules
continue
elif plugin_name in common.NOTIFY_MODULE_MAP:
# we're already handling this object
continue
# Add our plugin name to our module map
common.NOTIFY_MODULE_MAP[plugin_name] = {
'plugin': plugin,
'module': module,
}
# Add our module name to our __all__
__all__.append(plugin_name)
fn = getattr(plugin, 'schemas', None)
schemas = set([]) if not callable(fn) else fn(plugin)
# map our schema to our plugin
for schema in schemas:
if schema in common.NOTIFY_SCHEMA_MAP:
logger.error(
"Notification schema ({}) mismatch detected - {} to {}"
.format(schema, common.NOTIFY_SCHEMA_MAP[schema], plugin))
continue
# Assign plugin
common.NOTIFY_SCHEMA_MAP[schema] = plugin
return common.NOTIFY_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
common.NOTIFY_SCHEMA_MAP.clear()
# Iterate over our module map so we can clear out our __all__ and globals
for plugin_name in common.NOTIFY_MODULE_MAP.keys():
# Remove element from plugins
__all__.remove(plugin_name)
# Clear out our module map
common.NOTIFY_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
@ -176,6 +81,10 @@ def _sanitize_token(tokens, default_delimiter):
# Do not touch this field
continue
elif 'name' not in tokens[key]:
# Default to key
tokens[key]['name'] = key
if 'map_to' not in tokens[key]:
# Default type to key
tokens[key]['map_to'] = key
@ -538,16 +447,16 @@ def url_to_dict(url, secure_logging=True):
# Ensure our schema is always in lower case
schema = schema.group('schema').lower()
if schema not in common.NOTIFY_SCHEMA_MAP:
if schema not in N_MGR:
# Give the user the benefit of the doubt that the user may be using
# one of the URLs provided to them by their notification service.
# Before we fail for good, just scan all the plugins that support the
# native_url() parse function
results = \
next((r['plugin'].parse_native_url(_url)
for r in common.NOTIFY_MODULE_MAP.values()
if r['plugin'].parse_native_url(_url) is not None),
None)
results = None
for plugin in N_MGR.plugins():
results = plugin.parse_native_url(_url)
if results:
break
if not results:
logger.error('Unparseable URL {}'.format(loggable_url))
@ -560,14 +469,14 @@ def url_to_dict(url, secure_logging=True):
else:
# Parse our url details of the server object as dictionary
# containing all of the information parsed from our URL
results = common.NOTIFY_SCHEMA_MAP[schema].parse_url(_url)
results = N_MGR[schema].parse_url(_url)
if not results:
logger.error('Unparseable {} URL {}'.format(
common.NOTIFY_SCHEMA_MAP[schema].service_name, loggable_url))
N_MGR[schema].service_name, loggable_url))
return None
logger.trace('{} URL {} unpacked as:{}{}'.format(
common.NOTIFY_SCHEMA_MAP[schema].service_name, url,
N_MGR[schema].service_name, url,
os.linesep, os.linesep.join(
['{}="{}"'.format(k, v) for k, v in results.items()])))

160
apprise/utils.py
View File

@ -31,14 +31,12 @@ import sys
import json
import contextlib
import os
import hashlib
import locale
from itertools import chain
from os.path import expanduser
from functools import reduce
from . import common
from .logger import logger
from urllib.parse import unquote
from urllib.parse import quote
from urllib.parse import urlparse
@ -70,16 +68,12 @@ def import_module(path, name):
module = None
logger.debug(
'Custom module exception raised from %s (name=%s) %s',
'Module exception raised from %s (name=%s) %s',
path, name, str(e))
return module
# Hash of all paths previously scanned so we don't waste effort/overhead doing
# it again
PATHS_PREVIOUSLY_SCANNED = set()
# URL Indexing Table for returns via parse_url()
# The below accepts and scans for:
# - schema://
@ -214,6 +208,22 @@ VALID_PYTHON_FILE_RE = re.compile(r'.+\.py(o|c)?$', re.IGNORECASE)
REGEX_VALIDATE_LOOKUP = {}
class Singleton(type):
"""
Our Singleton MetaClass
"""
_instances = {}
def __call__(cls, *args, **kwargs):
"""
instantiate our singleton meta entry
"""
if cls not in cls._instances:
# we have not every built an instance before. Build one now.
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class TemplateType:
"""
Defines the different template types we can perform parsing on
@ -1561,142 +1571,6 @@ def remove_suffix(value, suffix):
return value[:-len(suffix)] if value.endswith(suffix) else value
def module_detection(paths, cache=True):
"""
Iterates over a defined path for apprise decorators to load such as
@notify.
"""
# A simple restriction that we don't allow periods in the filename at all
# so it can't be hidden (Linux OS's) and it won't conflict with Python
# path naming. This also prevents us from loading any python file that
# starts with an underscore or dash
# We allow __init__.py as well
module_re = re.compile(
r'^(?P<name>[_a-z0-9][a-z0-9._-]+)?(\.py)?$', re.I)
if isinstance(paths, str):
paths = [paths, ]
if not paths or not isinstance(paths, (tuple, list)):
# We're done
return None
def _import_module(path):
# Since our plugin name can conflict (as a module) with another
# we want to generate random strings to avoid steping on
# another's namespace
if not (path and VALID_PYTHON_FILE_RE.match(path)):
# Ignore file/module type
logger.trace('Plugin Scan: Skipping %s', path)
return None
module_name = hashlib.sha1(path.encode('utf-8')).hexdigest()
module_pyname = "{prefix}.{name}".format(
prefix='apprise.custom.module', name=module_name)
if module_pyname in common.NOTIFY_CUSTOM_MODULE_MAP:
# First clear out existing entries
for schema in common.\
NOTIFY_CUSTOM_MODULE_MAP[module_pyname]['notify'] \
.keys():
# Remove any mapped modules to this file
del common.NOTIFY_SCHEMA_MAP[schema]
# Reset
del common.NOTIFY_CUSTOM_MODULE_MAP[module_pyname]
# Load our module
module = import_module(path, module_pyname)
if not module:
# No problem, we can't use this object
logger.warning('Failed to load custom module: %s', _path)
return None
# Print our loaded modules if any
if module_pyname in common.NOTIFY_CUSTOM_MODULE_MAP:
logger.debug(
'Loaded custom module: %s (name=%s)',
_path, module_name)
for schema, meta in common.\
NOTIFY_CUSTOM_MODULE_MAP[module_pyname]['notify']\
.items():
logger.info('Loaded custom notification: %s://', schema)
else:
# The code reaches here if we successfully loaded the Python
# module but no hooks/triggers were found. So we can safely
# just remove/ignore this entry
del sys.modules[module_pyname]
return None
# end of _import_module()
return None
for _path in paths:
path = os.path.abspath(os.path.expanduser(_path))
if (cache and path in PATHS_PREVIOUSLY_SCANNED) \
or not os.path.exists(path):
# We're done as we've already scanned this
continue
# Store our path as a way of hashing it has been handled
PATHS_PREVIOUSLY_SCANNED.add(path)
if os.path.isdir(path) and not \
os.path.isfile(os.path.join(path, '__init__.py')):
logger.debug('Scanning for custom plugins in: %s', path)
for entry in os.listdir(path):
re_match = module_re.match(entry)
if not re_match:
# keep going
logger.trace('Plugin Scan: Ignoring %s', entry)
continue
new_path = os.path.join(path, entry)
if os.path.isdir(new_path):
# Update our path
new_path = os.path.join(path, entry, '__init__.py')
if not os.path.isfile(new_path):
logger.trace(
'Plugin Scan: Ignoring %s',
os.path.join(path, entry))
continue
if not cache or \
(cache and new_path not in PATHS_PREVIOUSLY_SCANNED):
# Load our module
_import_module(new_path)
# Add our subdir path
PATHS_PREVIOUSLY_SCANNED.add(new_path)
else:
if os.path.isdir(path):
# This logic is safe to apply because we already validated
# the directories state above; update our path
path = os.path.join(path, '__init__.py')
if cache and path in PATHS_PREVIOUSLY_SCANNED:
continue
PATHS_PREVIOUSLY_SCANNED.add(path)
# directly load as is
re_match = module_re.match(os.path.basename(path))
# must be a match and must have a .py extension
if not re_match or not re_match.group(1):
# keep going
logger.trace('Plugin Scan: Ignoring %s', path)
continue
# Load our module
_import_module(path)
return None
def dict_full_update(dict1, dict2):
"""
Takes 2 dictionaries (dict1 and dict2) that contain sub-dictionaries and

17
packaging/redhat/apprise-no-macosx-testing.patch
View File

@ -1,12 +1,15 @@
diff -Naur apprise-1.2.0/test/test_plugin_macosx.py apprise-1.2.0.patched/test/test_plugin_macosx.py
--- apprise-1.2.0/test/test_plugin_macosx.py 2022-11-02 14:59:59.000000000 -0400
+++ apprise-1.2.0.patched/test/test_plugin_macosx.py 2022-11-15 13:25:27.991284405 -0500
@@ -38,7 +38,7 @@
diff -Naur apprise-1.6.0/test/test_plugin_macosx.py apprise-1.6.0.patched/test/test_plugin_macosx.py
--- apprise-1.6.0/test/test_plugin_macosx.py 2023-12-22 16:51:24.000000000 -0500
+++ apprise-1.6.0.patched/test/test_plugin_macosx.py 2023-12-22 17:38:35.720131819 -0500
@@ -42,9 +42,8 @@
logging.disable(logging.CRITICAL)
-if sys.platform not in ["darwin", "linux"]:
+if sys.platform not in ["darwin"]:
pytest.skip("Only makes sense on macOS, but also works on Linux",
allow_module_level=True)
- pytest.skip("Only makes sense on macOS, but testable in Linux",
- allow_module_level=True)
+if sys.platform != "darwin":
+ pytest.skip("MacOS test only", allow_module_level=True)
@pytest.fixture

19
packaging/redhat/apprise-pytest-session_mocker-removal.patch
View File

@ -1,19 +1,20 @@
diff -Naur apprise-1.2.0/test/conftest.py apprise-1.2.0.patched/test/conftest.py
--- apprise-1.2.0/test/conftest.py 2022-11-02 14:00:13.000000000 -0400
+++ apprise-1.2.0.patched/test/conftest.py 2022-11-15 10:28:23.793969418 -0500
@@ -32,12 +32,12 @@
sys.path.append(os.path.join(os.path.dirname(__file__), 'helpers'))
diff -Naur apprise-1.6.0/test/conftest.py apprise-1.6.0-patched/test/conftest.py
--- apprise-1.6.0/test/conftest.py 2023-12-27 11:20:40.000000000 -0500
+++ apprise-1.6.0-patched/test/conftest.py 2023-12-27 13:43:22.583100037 -0500
@@ -45,8 +45,8 @@
A_MGR = AttachmentManager()
-@pytest.fixture(scope="session", autouse=True)
-@pytest.fixture(scope="function", autouse=True)
-def no_throttling_everywhere(session_mocker):
+@pytest.fixture(autouse=True)
+def no_throttling_everywhere(mocker):
"""
A pytest session fixture which disables throttling on all notifiers.
It is automatically enabled.
"""
for notifier in NOTIFY_MODULE_MAP.values():
plugin = notifier["plugin"]
@@ -57,4 +57,4 @@
A_MGR.unload_modules()
for plugin in N_MGR.plugins():
- session_mocker.patch.object(plugin, "request_rate_per_sec", 0)
+ mocker.patch.object(plugin, "request_rate_per_sec", 0)

21
test/conftest.py
View File

@ -31,17 +31,30 @@ import os
import pytest
from apprise.common import NOTIFY_MODULE_MAP
from apprise.NotificationManager import NotificationManager
from apprise.ConfigurationManager import ConfigurationManager
from apprise.AttachmentManager import AttachmentManager
sys.path.append(os.path.join(os.path.dirname(__file__), 'helpers'))
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
# Grant access to our Config Manager Singleton
C_MGR = ConfigurationManager()
# Grant access to our Attachment Manager Singleton
A_MGR = AttachmentManager()
@pytest.fixture(scope="session", autouse=True)
@pytest.fixture(scope="function", autouse=True)
def no_throttling_everywhere(session_mocker):
"""
A pytest session fixture which disables throttling on all notifiers.
It is automatically enabled.
"""
for notifier in NOTIFY_MODULE_MAP.values():
plugin = notifier["plugin"]
# Ensure we're working with a clean slate for each test
N_MGR.unload_modules()
C_MGR.unload_modules()
A_MGR.unload_modules()
for plugin in N_MGR.plugins():
session_mocker.patch.object(plugin, "request_rate_per_sec", 0)

89
test/helpers/module.py
View File

@ -26,11 +26,21 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from itertools import chain
from importlib import import_module, reload
from apprise.NotificationManager import NotificationManager
from apprise.AttachmentManager import AttachmentManager
import sys
import re
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
# Grant access to our Attachment Manager Singleton
A_MGR = AttachmentManager()
def reload_plugin(name, replace_in=None):
def reload_plugin(name):
"""
Reload built-in plugin module, e.g. `NotifyGnome`.
@ -40,21 +50,72 @@ def reload_plugin(name, replace_in=None):
See also https://stackoverflow.com/questions/31363311.
"""
module_name = f"apprise.plugins.{name}"
A_MGR.unload_modules()
reload(sys.modules['apprise.common'])
reload(sys.modules['apprise.attachment'])
reload(sys.modules['apprise.config'])
if module_name in sys.modules:
reload(sys.modules[module_name])
reload(sys.modules['apprise.plugins'])
reload(sys.modules['apprise.attachment.AttachBase'])
reload(sys.modules['apprise.AppriseAttachment'])
new_apprise_attachment_mod = import_module('apprise.AppriseAttachment')
reload(sys.modules['apprise.AttachmentManager'])
module_pyname = '{}.{}'.format(N_MGR.module_name_prefix, name)
if module_pyname in sys.modules:
reload(sys.modules[module_pyname])
new_notify_mod = import_module(module_pyname)
reload(sys.modules['apprise.NotificationManager'])
reload(sys.modules['apprise.Apprise'])
reload(sys.modules['apprise.utils'])
reload(sys.modules['apprise'])
# Fix reference to new plugin class in given module.
# Needed for updating the module-level import reference like
# `from apprise.plugins.NotifyMacOSX import NotifyMacOSX`.
if replace_in is not None:
mod = import_module(module_name)
setattr(replace_in, name, getattr(mod, name))
# Filter our keys
tests = [k for k in sys.modules.keys() if re.match(r'^test_.+$', k)]
for module_name in tests:
possible_matches = \
[m for m in dir(sys.modules[module_name])
if re.match(r'^(?P<name>Notify[a-z0-9]+)$', m, re.I)]
if not possible_matches:
continue
# Fix reference to new plugin class in given module.
# Needed for updating the module-level import reference like
# `from apprise.plugins.NotifyABCDE import NotifyABCDE`.
#
# We reload NotifyABCDE and place it back in its spot
test_mod = import_module(module_name)
setattr(test_mod, name, getattr(new_notify_mod, name))
# Detect our Apprise Modules (include helpers)
apprise_modules = \
[k for k in sys.modules.keys()
if re.match(r'^(apprise|helpers)(\.|.+)$', k)]
for entry in A_MGR:
reload(sys.modules[entry['path']])
for module_pyname in chain(apprise_modules, tests):
detect = re.compile(
r'^(?P<name>(AppriseAttachment|AttachBase|' +
entry['path'].split('.')[-1] + r'))$')
possible_matches = \
[m for m in dir(sys.modules[module_pyname]) if detect.match(m)]
if not possible_matches:
continue
apprise_mod = import_module(module_pyname)
# Fix reference to new plugin class in given module.
# Needed for updating the module-level import reference
# like `from apprise.<etc> import NotifyABCDE`.
#
# We reload NotifyABCDE and place it back in its spot
# new_attach = import_module(entry['path'])
for name in possible_matches:
if name == 'AppriseAttachment':
setattr(
apprise_mod, name,
getattr(new_apprise_attachment_mod, name))
else:
module_pyname = '{}.{}'.format(
A_MGR.module_name_prefix, name)
new_attach_mod = import_module(module_pyname)
setattr(apprise_mod, name, getattr(new_attach_mod, name))

196
test/test_api.py
View File

@ -32,6 +32,7 @@ import re
import sys
import pytest
import requests
from inspect import cleandoc
from unittest import mock
from os.path import dirname
@ -50,9 +51,7 @@ from apprise import PrivacyMode
from apprise.AppriseLocale import LazyTranslation
from apprise.AppriseLocale import gettext_lazy as _
from apprise import common
from apprise.plugins import __load_matrix
from apprise.plugins import __reset_matrix
from apprise.NotificationManager import NotificationManager
from apprise.utils import parse_list
from helpers import OuterEventLoop
import inspect
@ -64,8 +63,11 @@ logging.disable(logging.CRITICAL)
# Attachment Directory
TEST_VAR_DIR = join(dirname(__file__), 'var')
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
def test_apprise():
def test_apprise_object():
"""
API: Apprise() object
@ -90,11 +92,6 @@ def test_apprise_async():
def apprise_test(do_notify):
# 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
__load_matrix()
a = Apprise()
# no items
@ -217,10 +214,10 @@ def apprise_test(do_notify):
return NotifyBase.parse_url(url, verify_host=False)
# Store our bad notification in our schema map
common.NOTIFY_SCHEMA_MAP['bad'] = BadNotification
N_MGR['bad'] = BadNotification
# Store our good notification in our schema map
common.NOTIFY_SCHEMA_MAP['good'] = GoodNotification
N_MGR['good'] = GoodNotification
# Just to explain what is happening here, we would have parsed the
# url properly but failed when we went to go and create an instance
@ -334,13 +331,13 @@ def apprise_test(do_notify):
return ''
# Store our bad notification in our schema map
common.NOTIFY_SCHEMA_MAP['throw'] = ThrowNotification
N_MGR['throw'] = ThrowNotification
# Store our good notification in our schema map
common.NOTIFY_SCHEMA_MAP['fail'] = FailNotification
N_MGR['fail'] = FailNotification
# Store our good notification in our schema map
common.NOTIFY_SCHEMA_MAP['runtime'] = RuntimeNotification
N_MGR['runtime'] = RuntimeNotification
for async_mode in (True, False):
# Create an Asset object
@ -368,7 +365,11 @@ def apprise_test(do_notify):
# Support URL
return ''
common.NOTIFY_SCHEMA_MAP['throw'] = ThrowInstantiateNotification
N_MGR.unload_modules()
N_MGR['throw'] = ThrowInstantiateNotification
# Store our good notification in our schema map
N_MGR['good'] = GoodNotification
# Reset our object
a.clear()
@ -680,11 +681,7 @@ def test_apprise_schemas(tmpdir):
API: Apprise().schema() tests
"""
# 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
__load_matrix()
# Clear loaded modules
a = Apprise()
# no items
@ -711,16 +708,23 @@ def test_apprise_schemas(tmpdir):
secure_protocol = 'markdowns'
# Store our notifications into our schema map
common.NOTIFY_SCHEMA_MAP['text'] = TextNotification
common.NOTIFY_SCHEMA_MAP['html'] = HtmlNotification
common.NOTIFY_SCHEMA_MAP['markdown'] = MarkDownNotification
schemas = URLBase.schemas(TextNotification)
assert isinstance(schemas, set) is True
# We didn't define a protocol or secure protocol
assert len(schemas) == 0
# Store our notifications into our schema map
N_MGR['text'] = TextNotification
N_MGR['html'] = HtmlNotification
N_MGR['markdown'] = MarkDownNotification
schemas = URLBase.schemas(TextNotification)
assert isinstance(schemas, set) is True
# We didn't define a protocol or secure protocol one
# but one got assigned in he above N_MGR call
assert len(schemas) == 1
assert 'text' in schemas
schemas = URLBase.schemas(HtmlNotification)
assert isinstance(schemas, set) is True
assert len(schemas) == 4
@ -784,11 +788,6 @@ def test_apprise_notify_formats(tmpdir):
API: Apprise() Input Formats tests
"""
# 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
__load_matrix()
# Need to set async_mode=False to call notify() instead of async_notify().
asset = AppriseAsset(async_mode=False)
@ -845,9 +844,9 @@ def test_apprise_notify_formats(tmpdir):
return ''
# Store our notifications into our schema map
common.NOTIFY_SCHEMA_MAP['text'] = TextNotification
common.NOTIFY_SCHEMA_MAP['html'] = HtmlNotification
common.NOTIFY_SCHEMA_MAP['markdown'] = MarkDownNotification
N_MGR['text'] = TextNotification
N_MGR['html'] = HtmlNotification
N_MGR['markdown'] = MarkDownNotification
# Test Markdown; the above calls the markdown because our good://
# defined plugin above was defined to default to HTML which triggers
@ -1021,8 +1020,9 @@ def test_apprise_disabled_plugins():
API: Apprise() Disabled Plugin States
"""
# Reset our matrix
__reset_matrix()
# Ensure there are no other drives loaded
N_MGR.unload_modules(disable_native=True)
assert len(N_MGR) == 0
class TestDisabled01Notification(NotifyBase):
"""
@ -1044,7 +1044,7 @@ def test_apprise_disabled_plugins():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['na01'] = TestDisabled01Notification
N_MGR['na01'] = TestDisabled01Notification
class TestDisabled02Notification(NotifyBase):
"""
@ -1069,7 +1069,7 @@ def test_apprise_disabled_plugins():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['na02'] = TestDisabled02Notification
N_MGR['na02'] = TestDisabled02Notification
# Create our Apprise instance
a = Apprise()
@ -1127,7 +1127,7 @@ def test_apprise_disabled_plugins():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['good'] = TesEnabled01Notification
N_MGR['good'] = TesEnabled01Notification
# The last thing we'll simulate is a case where the plugin is just
# disabled at a later time long into it's life. this is just to allow
@ -1148,9 +1148,8 @@ def test_apprise_disabled_plugins():
# our notifications will go okay now
assert plugin.notify("My Message") is True
# Reset our matrix
__reset_matrix()
__load_matrix()
# Restore our modules
N_MGR.unload_modules()
def test_apprise_details():
@ -1158,9 +1157,6 @@ def test_apprise_details():
API: Apprise() Details
"""
# Reset our matrix
__reset_matrix()
# This is a made up class that is just used to verify
class TestDetailNotification(NotifyBase):
"""
@ -1190,22 +1186,23 @@ def test_apprise_details():
# previously defined in the base package and builds onto them
template_tokens = dict(NotifyBase.template_tokens, **{
'notype': {
# Nothing defined is still valid
# name is a minimum requirement
'name': _('no type'),
},
'regex_test01': {
'name': _('RegexTest'),
'type': 'string',
'regex': r'[A-Z0-9]',
'regex': r'^[A-Z0-9]$',
},
'regex_test02': {
'name': _('RegexTest'),
# Support regex options too
'regex': (r'[A-Z0-9]', 'i'),
'regex': (r'^[A-Z0-9]$', 'i'),
},
'regex_test03': {
'name': _('RegexTest'),
# Support regex option without a second option
'regex': (r'[A-Z0-9]'),
'regex': (r'^[A-Z0-9]$'),
},
'regex_test04': {
# this entry would just end up getting removed
@ -1267,7 +1264,7 @@ def test_apprise_details():
return True
# Store our good detail notification in our schema map
common.NOTIFY_SCHEMA_MAP['details'] = TestDetailNotification
N_MGR['details'] = TestDetailNotification
# This is a made up class that is just used to verify
class TestReq01Notification(NotifyBase):
@ -1292,7 +1289,7 @@ def test_apprise_details():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['req01'] = TestReq01Notification
N_MGR['req01'] = TestReq01Notification
# This is a made up class that is just used to verify
class TestReq02Notification(NotifyBase):
@ -1322,7 +1319,7 @@ def test_apprise_details():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['req02'] = TestReq02Notification
N_MGR['req02'] = TestReq02Notification
# This is a made up class that is just used to verify
class TestReq03Notification(NotifyBase):
@ -1348,7 +1345,7 @@ def test_apprise_details():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['req03'] = TestReq03Notification
N_MGR['req03'] = TestReq03Notification
# This is a made up class that is just used to verify
class TestReq04Notification(NotifyBase):
@ -1368,7 +1365,7 @@ def test_apprise_details():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['req04'] = TestReq04Notification
N_MGR['req04'] = TestReq04Notification
# This is a made up class that is just used to verify
class TestReq05Notification(NotifyBase):
@ -1390,7 +1387,7 @@ def test_apprise_details():
# Pretend everything is okay (so we don't break other tests)
return True
common.NOTIFY_SCHEMA_MAP['req05'] = TestReq05Notification
N_MGR['req05'] = TestReq05Notification
# Create our Apprise instance
a = Apprise()
@ -1451,21 +1448,13 @@ def test_apprise_details():
assert isinstance(entry['requirements']['packages_required'], list)
assert isinstance(entry['requirements']['packages_recommended'], list)
# 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()
# Prepare our object
a = Apprise()
# Details object
@ -1757,7 +1746,7 @@ def test_apprise_details_plugin_verification():
)
spec = inspect.getfullargspec(
common.NOTIFY_SCHEMA_MAP[protocols[0]].__init__)
N_MGR._schema_map[protocols[0]].__init__)
function_args = \
(set(parse_list(spec.varkw)) - set(['kwargs'])) \
@ -1773,7 +1762,8 @@ def test_apprise_details_plugin_verification():
'{}.__init__() expects a {}=None entry according to '
'template configuration'
.format(
common.NOTIFY_SCHEMA_MAP[protocols[0]].__name__, arg))
N_MGR._schema_map
[protocols[0]].__name__, arg))
# Iterate over all of the function arguments and make sure that
# it maps back to a key
@ -1783,8 +1773,7 @@ def test_apprise_details_plugin_verification():
raise AssertionError(
'{}.__init__({}) found but not defined in the '
'template configuration'
.format(
common.NOTIFY_SCHEMA_MAP[protocols[0]].__name__, arg))
.format(N_MGR._schema_map[protocols[0]].__name__, arg))
# Iterate over our map_to_aliases and make sure they were defined in
# either the as a token or arg
@ -1935,56 +1924,61 @@ def test_notify_matrix_dynamic_importing(tmpdir):
base.join("__init__.py").write('')
# Test no app_id
base.join('NotifyBadFile1.py').write(
base.join('NotifyBadFile1.py').write(cleandoc(
"""
class NotifyBadFile1:
pass""")
class NotifyBadFile1:
pass
"""))
# No class of the same name
base.join('NotifyBadFile2.py').write(
base.join('NotifyBadFile2.py').write(cleandoc(
"""
class BadClassName:
pass""")
class BadClassName:
pass
"""))
# Exception thrown
base.join('NotifyBadFile3.py').write("""raise ImportError()""")
# Utilizes a schema:// already occupied (as string)
base.join('NotifyGoober.py').write(
base.join('NotifyGoober.py').write(cleandoc(
"""
from apprise import NotifyBase
class NotifyGoober(NotifyBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
from apprise import NotifyBase
class NotifyGoober(NotifyBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
# The default simple (insecure) protocol (used by NotifyMail)
protocol = ('mailto', 'goober')
# The default simple (insecure) protocol (used by NotifyMail)
protocol = ('mailto', 'goober')
# The default secure protocol (used by NotifyMail)
secure_protocol = 'mailtos'
# The default secure protocol (used by NotifyMail)
secure_protocol = 'mailtos'
@staticmethod
def parse_url(url, *args, **kwargs):
# always parseable
return ConfigBase.parse_url(url, verify_host=False)""")
@staticmethod
def parse_url(url, *args, **kwargs):
# always parseable
return ConfigBase.parse_url(url, verify_host=False)
"""))
# Utilizes a schema:// already occupied (as tuple)
base.join('NotifyBugger.py').write("""
from apprise import NotifyBase
class NotifyBugger(NotifyBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
base.join('NotifyBugger.py').write(cleandoc(
"""
from apprise import NotifyBase
class NotifyBugger(NotifyBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
# The default simple (insecure) protocol (used by NotifyMail), the other
# isn't
protocol = ('mailto', 'bugger-test' )
# The default simple (insecure) protocol (used by NotifyMail), the
# other isn't
protocol = ('mailto', 'bugger-test' )
# The default secure protocol (used by NotifyMail), the other isn't
secure_protocol = ('mailtos', ['garbage'])
# The default secure protocol (used by NotifyMail), the other isn't
secure_protocol = ('mailtos', ['garbage'])
@staticmethod
def parse_url(url, *args, **kwargs):
# always parseable
return ConfigBase.parse_url(url, verify_host=False)""")
@staticmethod
def parse_url(url, *args, **kwargs):
# always parseable
return ConfigBase.parse_url(url, verify_host=False)
"""))
__load_matrix(path=str(base), name=module_name)
N_MGR.load_modules(path=str(base), name=module_name)

139
test/test_apprise_attachments.py
View File

@ -26,16 +26,15 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import sys
import pytest
from os.path import getsize
from os.path import join
from os.path import dirname
from inspect import cleandoc
from apprise.AttachmentManager import AttachmentManager
from apprise.AppriseAttachment import AppriseAttachment
from apprise.AppriseAsset import AppriseAsset
from apprise.attachment.AttachBase import AttachBase
from apprise.common import ATTACHMENT_SCHEMA_MAP
from apprise.attachment import __load_matrix
from apprise.common import ContentLocation
# Disable logging for a cleaner testing output
@ -44,6 +43,9 @@ logging.disable(logging.CRITICAL)
TEST_VAR_DIR = join(dirname(__file__), 'var')
# Grant access to our Attachment Manager Singleton
A_MGR = AttachmentManager()
def test_apprise_attachment():
"""
@ -278,7 +280,7 @@ def test_apprise_attachment_instantiate():
raise TypeError()
# Store our bad attachment type in our schema map
ATTACHMENT_SCHEMA_MAP['bad'] = BadAttachType
A_MGR['bad'] = BadAttachType
with pytest.raises(TypeError):
AppriseAttachment.instantiate(
@ -289,77 +291,6 @@ def test_apprise_attachment_instantiate():
'bad://path', suppress_exceptions=True) is None
def test_apprise_attachment_matrix_load():
"""
API: AppriseAttachment() matrix initialization
"""
import apprise
class AttachmentDummy(AttachBase):
"""
A dummy wrapper for testing the different options in the load_matrix
function
"""
# The default descriptive name associated with the Notification
service_name = 'dummy'
# protocol as tuple
protocol = ('uh', 'oh')
# secure protocol as tuple
secure_protocol = ('no', 'yes')
class AttachmentDummy2(AttachBase):
"""
A dummy wrapper for testing the different options in the load_matrix
function
"""
# The default descriptive name associated with the Notification
service_name = 'dummy2'
# secure protocol as tuple
secure_protocol = ('true', 'false')
class AttachmentDummy3(AttachBase):
"""
A dummy wrapper for testing the different options in the load_matrix
function
"""
# The default descriptive name associated with the Notification
service_name = 'dummy3'
# secure protocol as string
secure_protocol = 'true'
class AttachmentDummy4(AttachBase):
"""
A dummy wrapper for testing the different options in the load_matrix
function
"""
# The default descriptive name associated with the Notification
service_name = 'dummy4'
# protocol as string
protocol = 'true'
# Generate ourselves a fake entry
apprise.attachment.AttachmentDummy = AttachmentDummy
apprise.attachment.AttachmentDummy2 = AttachmentDummy2
apprise.attachment.AttachmentDummy3 = AttachmentDummy3
apprise.attachment.AttachmentDummy4 = AttachmentDummy4
__load_matrix()
# Call it again so we detect our entries already loaded
__load_matrix()
def test_attachment_matrix_dynamic_importing(tmpdir):
"""
API: Apprise() Attachment Matrix Importing
@ -372,53 +303,55 @@ def test_attachment_matrix_dynamic_importing(tmpdir):
module_name = 'badattach'
# Update our path to point to our new test suite
sys.path.insert(0, str(suite))
# Create a base area to work within
base = suite.mkdir(module_name)
base.join("__init__.py").write('')
# Test no app_id
base.join('AttachBadFile1.py').write(
base.join('AttachBadFile1.py').write(cleandoc(
"""
class AttachBadFile1:
pass""")
class AttachBadFile1:
pass
"""))
# No class of the same name
base.join('AttachBadFile2.py').write(
base.join('AttachBadFile2.py').write(cleandoc(
"""
class BadClassName:
pass""")
class BadClassName:
pass
"""))
# Exception thrown
base.join('AttachBadFile3.py').write("""raise ImportError()""")
# Utilizes a schema:// already occupied (as string)
base.join('AttachGoober.py').write(
base.join('AttachGoober.py').write(cleandoc(
"""
from apprise import AttachBase
class AttachGoober(AttachBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
from apprise import AttachBase
class AttachGoober(AttachBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
# The default simple (insecure) protocol
protocol = 'http'
# The default simple (insecure) protocol
protocol = 'http'
# The default secure protocol
secure_protocol = 'https'""")
# The default secure protocol
secure_protocol = 'https'
"""))
# Utilizes a schema:// already occupied (as tuple)
base.join('AttachBugger.py').write("""
from apprise import AttachBase
class AttachBugger(AttachBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
base.join('AttachBugger.py').write(cleandoc(
"""
from apprise import AttachBase
class AttachBugger(AttachBase):
# This class tests the fact we have a new class name, but we're
# trying to over-ride items previously used
# The default simple (insecure) protocol
protocol = ('http', 'bugger-test' )
# The default simple (insecure) protocol
protocol = ('http', 'bugger-test' )
# The default secure protocol
secure_protocol = ('https', 'bugger-tests')""")
# The default secure protocol
secure_protocol = ('https', 'bugger-tests')
"""))
__load_matrix(path=str(base), name=module_name)
A_MGR.load_modules(path=str(base), name=module_name)

176
test/test_apprise_cli.py
View File

@ -36,23 +36,21 @@ from os.path import dirname
from os.path import join
from apprise import cli
from apprise import NotifyBase
from apprise.common import NOTIFY_CUSTOM_MODULE_MAP
from apprise.utils import PATHS_PREVIOUSLY_SCANNED
from apprise.NotificationManager import NotificationManager
from click.testing import CliRunner
from apprise.common import NOTIFY_SCHEMA_MAP
from apprise.utils import environ
from apprise.plugins import __load_matrix
from apprise.plugins import __reset_matrix
from apprise.AppriseLocale import gettext_lazy as _
from importlib import reload
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
def test_apprise_cli_nux_env(tmpdir):
"""
@ -89,8 +87,8 @@ def test_apprise_cli_nux_env(tmpdir):
return 'bad://'
# Set up our notification types
NOTIFY_SCHEMA_MAP['good'] = GoodNotification
NOTIFY_SCHEMA_MAP['bad'] = BadNotification
N_MGR['good'] = GoodNotification
N_MGR['bad'] = BadNotification
runner = CliRunner()
result = runner.invoke(cli.main)
@ -639,8 +637,8 @@ def test_apprise_cli_details(tmpdir):
])
assert result.exit_code == 0
# Reset our matrix
__reset_matrix()
# Clear loaded modules
N_MGR.unload_modules()
# This is a made up class that is just used to verify
class TestReq01Notification(NotifyBase):
@ -665,7 +663,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['req01'] = TestReq01Notification
N_MGR['req01'] = TestReq01Notification
# This is a made up class that is just used to verify
class TestReq02Notification(NotifyBase):
@ -695,7 +693,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['req02'] = TestReq02Notification
N_MGR['req02'] = TestReq02Notification
# This is a made up class that is just used to verify
class TestReq03Notification(NotifyBase):
@ -721,7 +719,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['req03'] = TestReq03Notification
N_MGR['req03'] = TestReq03Notification
# This is a made up class that is just used to verify
class TestReq04Notification(NotifyBase):
@ -741,7 +739,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['req04'] = TestReq04Notification
N_MGR['req04'] = TestReq04Notification
# This is a made up class that is just used to verify
class TestReq05Notification(NotifyBase):
@ -762,7 +760,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['req05'] = TestReq05Notification
N_MGR['req05'] = TestReq05Notification
class TestDisabled01Notification(NotifyBase):
"""
@ -784,7 +782,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['na01'] = TestDisabled01Notification
N_MGR['na01'] = TestDisabled01Notification
class TestDisabled02Notification(NotifyBase):
"""
@ -809,7 +807,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['na02'] = TestDisabled02Notification
N_MGR['na02'] = TestDisabled02Notification
# We'll add a good notification to our list
class TesEnabled01Notification(NotifyBase):
@ -829,7 +827,7 @@ def test_apprise_cli_details(tmpdir):
# Pretend everything is okay (so we don't break other tests)
return True
NOTIFY_SCHEMA_MAP['good'] = TesEnabled01Notification
N_MGR['good'] = TesEnabled01Notification
# Verify that we can pass through all of our different details
result = runner.invoke(cli.main, [
@ -842,9 +840,8 @@ def test_apprise_cli_details(tmpdir):
])
assert result.exit_code == 0
# Reset our matrix
__reset_matrix()
__load_matrix()
# Clear loaded modules
N_MGR.unload_modules()
@mock.patch('requests.post')
@ -863,8 +860,8 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
# This simulates an actual call from the CLI. Unfortunately through
# testing were occupying the same memory space so our singleton's
# have already been populated
PATHS_PREVIOUSLY_SCANNED.clear()
NOTIFY_CUSTOM_MODULE_MAP.clear()
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Test a path that has no files to load in it
result = runner.invoke(cli.main, [
@ -877,8 +874,8 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
assert result.exit_code == 0
# Directories that don't exist passed in by the CLI aren't even scanned
assert len(PATHS_PREVIOUSLY_SCANNED) == 0
assert len(NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._paths_previously_scanned) == 0
assert len(N_MGR._custom_module_map) == 0
# Test our current existing path that has no entries in it
result = runner.invoke(cli.main, [
@ -889,18 +886,19 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
# The path is silently loaded but fails... it's okay because the
# notification we're choosing to notify does exist
assert result.exit_code == 0
assert len(PATHS_PREVIOUSLY_SCANNED) == 1
assert join(str(tmpdir), 'empty') in PATHS_PREVIOUSLY_SCANNED
assert len(N_MGR._paths_previously_scanned) == 1
assert join(str(tmpdir), 'empty') in \
N_MGR._paths_previously_scanned
# However there was nothing to load
assert len(NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._custom_module_map) == 0
# Clear our working variables so they don't obstruct the next test
# This simulates an actual call from the CLI. Unfortunately through
# testing were occupying the same memory space so our singleton's
# have already been populated
PATHS_PREVIOUSLY_SCANNED.clear()
NOTIFY_CUSTOM_MODULE_MAP.clear()
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Prepare ourselves a file to work with
notify_hook_a_base = tmpdir.mkdir('random')
@ -921,10 +919,10 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
# The path is silently loaded but fails... it's okay because the
# notification we're choosing to notify does exist
assert len(PATHS_PREVIOUSLY_SCANNED) == 1
assert str(notify_hook_a) in PATHS_PREVIOUSLY_SCANNED
assert len(N_MGR._paths_previously_scanned) == 1
assert str(notify_hook_a) in N_MGR._paths_previously_scanned
# However there was nothing to load
assert len(NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._custom_module_map) == 0
# Prepare ourselves a file to work with
notify_hook_aa = notify_hook_a_base.join('myhook02.py')
@ -932,11 +930,12 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
garbage entry
"""))
N_MGR.plugins()
result = runner.invoke(cli.main, [
'--plugin-path', str(notify_hook_aa),
'-b', 'test\nbody',
# A custom hook:
'clihook://',
'clihook://custom',
])
# It doesn't exist so it will fail
# meanwhile we would have failed to load the myhook path
@ -944,18 +943,20 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
# The path is silently loaded but fails...
# as a result the path stacks with the last
assert len(PATHS_PREVIOUSLY_SCANNED) == 2
assert str(notify_hook_a) in PATHS_PREVIOUSLY_SCANNED
assert str(notify_hook_aa) in PATHS_PREVIOUSLY_SCANNED
assert len(N_MGR._paths_previously_scanned) == 2
assert str(notify_hook_a) in \
N_MGR._paths_previously_scanned
assert str(notify_hook_aa) in \
N_MGR._paths_previously_scanned
# However there was nothing to load
assert len(NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._custom_module_map) == 0
# Clear our working variables so they don't obstruct the next test
# This simulates an actual call from the CLI. Unfortunately through
# testing were occupying the same memory space so our singleton's
# have already been populated
PATHS_PREVIOUSLY_SCANNED.clear()
NOTIFY_CUSTOM_MODULE_MAP.clear()
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Prepare ourselves a file to work with
notify_hook_b = tmpdir.mkdir('goodmodule').join('__init__.py')
@ -968,6 +969,14 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
# A simple test - print to screen
print("{}: {} - {}".format(notify_type, title, body))
# No return (so a return of None) get's translated to True
# Define another in the same file
@notify(on="clihookA")
def mywrapper(body, title, notify_type, *args, **kwargs):
# A simple test - print to screen
print("!! {}: {} - {}".format(notify_type, title, body))
# No return (so a return of None) get's translated to True
"""))
@ -975,7 +984,7 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
'--plugin-path', str(tmpdir),
'-b', 'test body',
# A custom hook:
'clihook://',
'clihook://still/valid',
])
# We can detect the goodmodule (which has an __init__.py in it)
@ -983,51 +992,52 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
assert result.exit_code == 0
# Let's see how things got loaded:
assert len(PATHS_PREVIOUSLY_SCANNED) == 2
assert str(tmpdir) in PATHS_PREVIOUSLY_SCANNED
assert len(N_MGR._paths_previously_scanned) == 2
assert str(tmpdir) in N_MGR._paths_previously_scanned
# absolute path to detected module is also added
assert join(str(tmpdir), 'goodmodule', '__init__.py') \
in PATHS_PREVIOUSLY_SCANNED
in N_MGR._paths_previously_scanned
# We also loaded our clihook properly
assert len(NOTIFY_CUSTOM_MODULE_MAP) == 1
assert len(N_MGR._custom_module_map) == 1
# We can find our new hook loaded in our NOTIFY_SCHEMA_MAP now...
assert 'clihook' in NOTIFY_SCHEMA_MAP
# We can find our new hook loaded in our schema map now...
assert 'clihook' in N_MGR
# Capture our key for reference
key = [k for k in NOTIFY_CUSTOM_MODULE_MAP.keys()][0]
key = [k for k in N_MGR._custom_module_map.keys()][0]
assert len(NOTIFY_CUSTOM_MODULE_MAP[key]['notify']) == 1
assert 'clihook' in NOTIFY_CUSTOM_MODULE_MAP[key]['notify']
# We loaded 2 entries from the same file
assert len(N_MGR._custom_module_map[key]['notify']) == 2
assert 'clihook' in N_MGR._custom_module_map[key]['notify']
# Converted to lower case
assert 'clihooka' in N_MGR._custom_module_map[key]['notify']
# Our function name
assert NOTIFY_CUSTOM_MODULE_MAP[key]['notify']['clihook']['fn_name'] \
assert N_MGR._custom_module_map[key]['notify']['clihook']['fn_name'] \
== 'mywrapper'
# What we parsed from the `on` keyword in the @notify decorator
assert NOTIFY_CUSTOM_MODULE_MAP[key]['notify']['clihook']['url'] \
assert N_MGR._custom_module_map[key]['notify']['clihook']['url'] \
== 'clihook://'
# our default name Assignment. This can be-overridden on the @notify
# decorator by just adding a name= to the parameter list
assert NOTIFY_CUSTOM_MODULE_MAP[key]['notify']['clihook']['name'] \
== 'Custom - clihook'
assert N_MGR['clihook'].service_name == 'Custom - clihook'
# Our Base Notification object when initialized:
assert isinstance(
NOTIFY_CUSTOM_MODULE_MAP[key]['notify']['clihook']['plugin'](),
NotifyBase)
# This is how it ties together in the backend
assert NOTIFY_CUSTOM_MODULE_MAP[key]['notify']['clihook']['plugin'] == \
NOTIFY_SCHEMA_MAP['clihook']
assert len(
N_MGR._module_map[N_MGR._custom_module_map[key]['name']]['plugin']) \
== 2
for plugin in \
N_MGR._module_map[N_MGR._custom_module_map[key]['name']]['plugin']:
assert isinstance(plugin(), NotifyBase)
# Clear our working variables so they don't obstruct the next test
# This simulates an actual call from the CLI. Unfortunately through
# testing were occupying the same memory space so our singleton's
# have already been populated
PATHS_PREVIOUSLY_SCANNED.clear()
NOTIFY_CUSTOM_MODULE_MAP.clear()
del NOTIFY_SCHEMA_MAP['clihook']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
del N_MGR['clihook']
result = runner.invoke(cli.main, [
'--plugin-path', str(notify_hook_b),
@ -1046,9 +1056,9 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
# This simulates an actual call from the CLI. Unfortunately through
# testing were occupying the same memory space so our singleton's
# have already been populated
PATHS_PREVIOUSLY_SCANNED.clear()
NOTIFY_CUSTOM_MODULE_MAP.clear()
del NOTIFY_SCHEMA_MAP['clihook']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
del N_MGR['clihook']
result = runner.invoke(cli.main, [
'--plugin-path', dirname(str(notify_hook_b)),
@ -1080,9 +1090,9 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
# This simulates an actual call from the CLI. Unfortunately through
# testing were occupying the same memory space so our singleton's
# have already been populated
PATHS_PREVIOUSLY_SCANNED.clear()
NOTIFY_CUSTOM_MODULE_MAP.clear()
del NOTIFY_SCHEMA_MAP['clihook']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
del N_MGR['clihook']
# Prepare ourselves a file to work with
notify_hook_b = tmpdir.mkdir('complex').join('complex.py')
@ -1141,32 +1151,32 @@ def test_apprise_cli_plugin_loading(mock_post, tmpdir):
assert result.exit_code == 1
# Let's see how things got loaded
assert len(PATHS_PREVIOUSLY_SCANNED) == 2
assert len(N_MGR._paths_previously_scanned) == 2
# Our path we specified on the CLI...
assert join(str(tmpdir), 'complex') in PATHS_PREVIOUSLY_SCANNED
assert join(str(tmpdir), 'complex') in N_MGR._paths_previously_scanned
# absolute path to detected module is also added
assert join(str(tmpdir), 'complex', 'complex.py') \
in PATHS_PREVIOUSLY_SCANNED
in N_MGR._paths_previously_scanned
# We loaded our one module successfuly
assert len(NOTIFY_CUSTOM_MODULE_MAP) == 1
assert len(N_MGR._custom_module_map) == 1
# We can find our new hook loaded in our SCHEMA_MAP now...
assert 'willfail' in NOTIFY_SCHEMA_MAP
assert 'clihook1' in NOTIFY_SCHEMA_MAP
assert 'clihook2' in NOTIFY_SCHEMA_MAP
assert 'willfail' in N_MGR
assert 'clihook1' in N_MGR
assert 'clihook2' in N_MGR
# Capture our key for reference
key = [k for k in NOTIFY_CUSTOM_MODULE_MAP.keys()][0]
key = [k for k in N_MGR._custom_module_map.keys()][0]
assert len(NOTIFY_CUSTOM_MODULE_MAP[key]['notify']) == 3
assert 'willfail' in NOTIFY_CUSTOM_MODULE_MAP[key]['notify']
assert 'clihook1' in NOTIFY_CUSTOM_MODULE_MAP[key]['notify']
assert len(N_MGR._custom_module_map[key]['notify']) == 3
assert 'willfail' in N_MGR._custom_module_map[key]['notify']
assert 'clihook1' in N_MGR._custom_module_map[key]['notify']
# We only load 1 instance of the clihook2, the second will fail
assert 'clihook2' in NOTIFY_CUSTOM_MODULE_MAP[key]['notify']
assert 'clihook2' in N_MGR._custom_module_map[key]['notify']
# We can never load previously created notifications
assert 'json' not in NOTIFY_CUSTOM_MODULE_MAP[key]['notify']
assert 'json' not in N_MGR._custom_module_map[key]['notify']
result = runner.invoke(cli.main, [
'--plugin-path', join(str(tmpdir), 'complex'),

32
test/test_apprise_config.py
View File

@ -37,16 +37,21 @@ from apprise import AppriseConfig
from apprise import AppriseAsset
from apprise.config.ConfigBase import ConfigBase
from apprise.plugins.NotifyBase import NotifyBase
from apprise.NotificationManager import NotificationManager
from apprise.ConfigurationManager import ConfigurationManager
from apprise.common import CONFIG_SCHEMA_MAP
from apprise.common import NOTIFY_SCHEMA_MAP
from apprise.config import __load_matrix
from apprise.config.ConfigFile import ConfigFile
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
# Grant access to our Configuration Manager Singleton
C_MGR = ConfigurationManager()
def test_apprise_config(tmpdir):
"""
@ -248,7 +253,7 @@ def test_apprise_multi_config_entries(tmpdir):
return ''
# Store our good notification in our schema map
NOTIFY_SCHEMA_MAP['good'] = GoodNotification
N_MGR._schema_map['good'] = GoodNotification
# Create ourselves a config object
ac = AppriseConfig()
@ -468,7 +473,7 @@ def test_apprise_config_instantiate():
return ConfigBase.parse_url(url, verify_host=False)
# Store our bad configuration in our schema map
CONFIG_SCHEMA_MAP['bad'] = BadConfig
C_MGR['bad'] = BadConfig
with pytest.raises(TypeError):
AppriseConfig.instantiate(
@ -501,7 +506,7 @@ def test_invalid_apprise_config(tmpdir):
return ConfigBase.parse_url(url, verify_host=False)
# Store our bad configuration in our schema map
CONFIG_SCHEMA_MAP['bad'] = BadConfig
C_MGR['bad'] = BadConfig
# temporary file to work with
t = tmpdir.mkdir("apprise-bad-obj").join("invalid")
@ -566,7 +571,7 @@ def test_apprise_config_with_apprise_obj(tmpdir):
return ''
# Store our good notification in our schema map
NOTIFY_SCHEMA_MAP['good'] = GoodNotification
N_MGR._schema_map['good'] = GoodNotification
# Create ourselves a config object
ac = AppriseConfig(cache=False)
@ -765,9 +770,9 @@ def test_recursive_config_inclusion(tmpdir):
allow_cross_includes = ContentIncludeMode.NEVER
# store our entries
CONFIG_SCHEMA_MAP['never'] = ConfigCrossPostNever
CONFIG_SCHEMA_MAP['strict'] = ConfigCrossPostStrict
CONFIG_SCHEMA_MAP['always'] = ConfigCrossPostAlways
C_MGR['never'] = ConfigCrossPostNever
C_MGR['strict'] = ConfigCrossPostStrict
C_MGR['always'] = ConfigCrossPostAlways
# Make our new path valid
suite = tmpdir.mkdir("apprise_config_recursion")
@ -974,11 +979,6 @@ def test_apprise_config_matrix_load():
apprise.config.ConfigDummy3 = ConfigDummy3
apprise.config.ConfigDummy4 = ConfigDummy4
__load_matrix()
# Call it again so we detect our entries already loaded
__load_matrix()
def test_configmatrix_dynamic_importing(tmpdir):
"""
@ -1052,8 +1052,6 @@ class ConfigBugger(ConfigBase):
# always parseable
return ConfigBase.parse_url(url, verify_host=False)""")
__load_matrix(path=str(base), name=module_name)
@mock.patch('os.path.getsize')
def test_config_base_parse_inaccessible_text_file(mock_getsize, tmpdir):

262
test/test_apprise_utils.py
View File

@ -33,7 +33,7 @@ from inspect import cleandoc
from urllib.parse import unquote
from apprise import utils
from apprise import common
from apprise.NotificationManager import NotificationManager
# Disable logging for a cleaner testing output
import logging
@ -42,6 +42,9 @@ logging.disable(logging.CRITICAL)
# Ensure we don't create .pyc files for these tests
sys.dont_write_bytecode = True
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
def test_parse_qsd():
"utils: parse_qsd() testing """
@ -2030,15 +2033,15 @@ def test_module_detection(tmpdir):
"""
# Clear our working variables so they don't obstruct with the tests here
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Test case where we load invalid data
utils.module_detection(None)
N_MGR.module_detection(None)
# Invalid data does not load anything
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 0
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._paths_previously_scanned) == 0
assert len(N_MGR._custom_module_map) == 0
# Prepare ourselves a file to work with
notify_hook_a_base = tmpdir.mkdir('a')
@ -2057,30 +2060,29 @@ def test_module_detection(tmpdir):
"""))
# Not previously loaded
assert 'clihook' not in common.NOTIFY_SCHEMA_MAP
assert 'clihook' not in N_MGR
# load entry by string
utils.module_detection(str(notify_hook_a))
utils.module_detection(str(notify_ignore))
utils.module_detection(str(notify_hook_a_base))
N_MGR.module_detection(str(notify_hook_a))
N_MGR.module_detection(str(notify_ignore))
N_MGR.module_detection(str(notify_hook_a_base))
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 3
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert len(N_MGR._paths_previously_scanned) == 3
assert len(N_MGR._custom_module_map) == 1
# Now loaded
assert 'clihook' in common.NOTIFY_SCHEMA_MAP
assert 'clihook' in N_MGR
# load entry by array
utils.module_detection([str(notify_hook_a)])
N_MGR.module_detection([str(notify_hook_a)])
# No changes to our path
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 3
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert len(N_MGR._paths_previously_scanned) == 3
assert len(N_MGR._custom_module_map) == 1
# Reset our variables for the next test
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
del common.NOTIFY_SCHEMA_MAP['clihook']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Hidden files are ignored
notify_hook_b_base = tmpdir.mkdir('b')
@ -2094,36 +2096,36 @@ def test_module_detection(tmpdir):
pass
"""))
assert 'hidden' not in common.NOTIFY_SCHEMA_MAP
assert 'hidden' not in N_MGR
utils.module_detection([str(notify_hook_b)])
N_MGR.module_detection([str(notify_hook_b)])
# Verify that it did not load
assert 'hidden' not in common.NOTIFY_SCHEMA_MAP
assert 'hidden' not in N_MGR
# Path was scanned; nothing loaded
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 1
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._paths_previously_scanned) == 1
assert len(N_MGR._custom_module_map) == 0
# Reset our variables for the next test
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# modules with no hooks found are ignored
notify_hook_c_base = tmpdir.mkdir('c')
notify_hook_c = notify_hook_c_base.join('empty.py')
notify_hook_c.write("")
utils.module_detection([str(notify_hook_c)])
N_MGR.module_detection([str(notify_hook_c)])
# File was found, no custom modules
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 1
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._paths_previously_scanned) == 1
assert len(N_MGR._custom_module_map) == 0
# A new path scanned
utils.module_detection([str(notify_hook_c_base)])
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
N_MGR.module_detection([str(notify_hook_c_base)])
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 0
def create_hook(tdir, cache=True, on="valid1"):
"""
@ -2139,39 +2141,39 @@ def test_module_detection(tmpdir):
pass
""".format(on)))
utils.module_detection([str(tdir)], cache=cache)
N_MGR.module_detection([str(tdir)], cache=cache)
create_hook(notify_hook_c, on='valid1')
assert 'valid1' not in common.NOTIFY_SCHEMA_MAP
assert 'valid1' not in N_MGR
# Even if we correct our empty file; the fact the directory has been
# scanned and failed to load (same with file), it won't be loaded
# a second time. This is intentional since module_detection() get's
# called for every AppriseAsset() object creation. This prevents us
# from reloading conent over and over again wasting resources
assert 'valid1' not in common.NOTIFY_SCHEMA_MAP
utils.module_detection([str(notify_hook_c)])
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
assert 'valid1' not in N_MGR
N_MGR.module_detection([str(notify_hook_c)])
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 0
# Even by absolute path...
utils.module_detection([str(notify_hook_c)])
assert 'valid1' not in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
N_MGR.module_detection([str(notify_hook_c)])
assert 'valid1' not in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 0
# However we can bypass the cache if we really want to
utils.module_detection([str(notify_hook_c_base)], cache=False)
assert 'valid1' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
N_MGR.module_detection([str(notify_hook_c_base)], cache=False)
assert 'valid1' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 1
# Bypassing it twice causes the module to load twice (not very efficient)
# However we can bypass the cache if we really want to
utils.module_detection([str(notify_hook_c_base)], cache=False)
assert 'valid1' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
N_MGR.module_detection([str(notify_hook_c_base)], cache=False)
assert 'valid1' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 1
# If you update the module (corrupting it in the process and reload)
notify_hook_c.write(cleandoc("""
@ -2179,50 +2181,50 @@ def test_module_detection(tmpdir):
"""))
# Force no cache to cause the file to be replaced
utils.module_detection([str(notify_hook_c_base)], cache=False)
N_MGR.module_detection([str(notify_hook_c_base)], cache=False)
# Our valid entry is no longer loaded
assert 'valid1' not in common.NOTIFY_SCHEMA_MAP
assert 'valid1' not in N_MGR
# No change to scanned paths
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(N_MGR._paths_previously_scanned) == 2
# The previously loaded module is now gone
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
assert len(N_MGR._custom_module_map) == 0
# Reload our valid1 entry
create_hook(notify_hook_c, on='valid1', cache=False)
assert 'valid1' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert 'valid1' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 1
# Prepare an empty file
notify_hook_c.write("")
utils.module_detection([str(notify_hook_c_base)], cache=False)
N_MGR.module_detection([str(notify_hook_c_base)], cache=False)
# Our valid entry is no longer loaded
assert 'valid1' not in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
assert 'valid1' not in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 0
# Now reload our module again (this time rather then an exception, the
# module is read back and swaps `valid1` for `valid2`
create_hook(notify_hook_c, on='valid1', cache=False)
assert 'valid1' in common.NOTIFY_SCHEMA_MAP
assert 'valid2' not in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert 'valid1' in N_MGR
assert 'valid2' not in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 1
create_hook(notify_hook_c, on='valid2', cache=False)
assert 'valid1' not in common.NOTIFY_SCHEMA_MAP
assert 'valid2' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert 'valid1' not in N_MGR
assert 'valid2' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 1
# Reset our variables for the next test
create_hook(notify_hook_c, on='valid1', cache=False)
del common.NOTIFY_SCHEMA_MAP['valid1']
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
del N_MGR['valid1']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
notify_hook_d = notify_hook_c_base.join('.ignore.py')
notify_hook_d.write("")
@ -2241,71 +2243,71 @@ def test_module_detection(tmpdir):
# Try to load our base directory again; this time we search by the
# directory; the only edge case we're testing here is it will not
# even look at the .ignore.py file found since it is invalid
utils.module_detection([str(notify_hook_c_base)])
assert 'valid1' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
N_MGR.module_detection([str(notify_hook_c_base)])
assert 'valid1' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert len(N_MGR._custom_module_map) == 1
# Reset our variables for the next test
del common.NOTIFY_SCHEMA_MAP['valid1']
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
del N_MGR._schema_map['valid1']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Try to load our base directory again
utils.module_detection([str(notify_hook_c)])
assert 'valid1' in common.NOTIFY_SCHEMA_MAP
N_MGR.module_detection([str(notify_hook_c)])
assert 'valid1' in N_MGR
# Hidden directories are not scanned
assert 'valid2' not in common.NOTIFY_SCHEMA_MAP
assert 'valid2' not in N_MGR
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 1
assert str(notify_hook_c) in utils.PATHS_PREVIOUSLY_SCANNED
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert len(N_MGR._paths_previously_scanned) == 1
assert str(notify_hook_c) in N_MGR._paths_previously_scanned
assert len(N_MGR._custom_module_map) == 1
# However a direct reference to the hidden directory is okay
utils.module_detection([str(notify_hook_e_base)])
N_MGR.module_detection([str(notify_hook_e_base)])
# We loaded our module
assert 'valid2' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 3
assert str(notify_hook_c) in utils.PATHS_PREVIOUSLY_SCANNED
assert str(notify_hook_e) in utils.PATHS_PREVIOUSLY_SCANNED
assert str(notify_hook_e_base) in utils.PATHS_PREVIOUSLY_SCANNED
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 2
assert 'valid2' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 3
assert str(notify_hook_c) in N_MGR._paths_previously_scanned
assert str(notify_hook_e) in N_MGR._paths_previously_scanned
assert str(notify_hook_e_base) in N_MGR._paths_previously_scanned
assert len(N_MGR._custom_module_map) == 2
# Reset our variables for the next test
del common.NOTIFY_SCHEMA_MAP['valid1']
del common.NOTIFY_SCHEMA_MAP['valid2']
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
del N_MGR._schema_map['valid1']
del N_MGR._schema_map['valid2']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Load our file directly
assert 'valid2' not in common.NOTIFY_SCHEMA_MAP
utils.module_detection([str(notify_hook_e)])
assert 'valid2' not in N_MGR
N_MGR.module_detection([str(notify_hook_e)])
# Now we have it loaded as expected
assert 'valid2' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 1
assert str(notify_hook_e) in utils.PATHS_PREVIOUSLY_SCANNED
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert 'valid2' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 1
assert str(notify_hook_e) in N_MGR._paths_previously_scanned
assert len(N_MGR._custom_module_map) == 1
# however if we try to load the base directory where the __init__.py
# was already loaded from, it will not change anything
utils.module_detection([str(notify_hook_e_base)])
assert 'valid2' in common.NOTIFY_SCHEMA_MAP
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 2
assert str(notify_hook_e) in utils.PATHS_PREVIOUSLY_SCANNED
assert str(notify_hook_e_base) in utils.PATHS_PREVIOUSLY_SCANNED
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
N_MGR.module_detection([str(notify_hook_e_base)])
assert 'valid2' in N_MGR
assert len(N_MGR._paths_previously_scanned) == 2
assert str(notify_hook_e) in N_MGR._paths_previously_scanned
assert str(notify_hook_e_base) in N_MGR._paths_previously_scanned
assert len(N_MGR._custom_module_map) == 1
# Tidy up for the next test
del common.NOTIFY_SCHEMA_MAP['valid2']
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
del N_MGR._schema_map['valid2']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
assert 'valid1' not in common.NOTIFY_SCHEMA_MAP
assert 'valid2' not in common.NOTIFY_SCHEMA_MAP
assert 'valid3' not in common.NOTIFY_SCHEMA_MAP
assert 'valid1' not in N_MGR
assert 'valid2' not in N_MGR
assert 'valid3' not in N_MGR
notify_hook_f_base = tmpdir.mkdir('f')
notify_hook_f = notify_hook_f_base.join('invalid.py')
notify_hook_f.write(cleandoc("""
@ -2334,20 +2336,20 @@ def test_module_detection(tmpdir):
"""))
utils.module_detection([str(notify_hook_f)])
N_MGR.module_detection([str(notify_hook_f)])
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 1
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 1
assert 'valid1' in common.NOTIFY_SCHEMA_MAP
assert 'valid2' in common.NOTIFY_SCHEMA_MAP
assert 'valid3' in common.NOTIFY_SCHEMA_MAP
assert len(N_MGR._paths_previously_scanned) == 1
assert len(N_MGR._custom_module_map) == 1
assert 'valid1' in N_MGR
assert 'valid2' in N_MGR
assert 'valid3' in N_MGR
# Reset our variables for the next test
del common.NOTIFY_SCHEMA_MAP['valid1']
del common.NOTIFY_SCHEMA_MAP['valid2']
del common.NOTIFY_SCHEMA_MAP['valid3']
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
del N_MGR._schema_map['valid1']
del N_MGR._schema_map['valid2']
del N_MGR._schema_map['valid3']
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
# Now test the handling of just bad data entirely
notify_hook_g_base = tmpdir.mkdir('g')
@ -2355,13 +2357,13 @@ def test_module_detection(tmpdir):
with open(str(notify_hook_g), 'wb') as fout:
fout.write(os.urandom(512))
utils.module_detection([str(notify_hook_g)])
assert len(utils.PATHS_PREVIOUSLY_SCANNED) == 1
assert len(common.NOTIFY_CUSTOM_MODULE_MAP) == 0
N_MGR.module_detection([str(notify_hook_g)])
assert len(N_MGR._paths_previously_scanned) == 1
assert len(N_MGR._custom_module_map) == 0
# Reset our variables before we leave
utils.PATHS_PREVIOUSLY_SCANNED.clear()
common.NOTIFY_CUSTOM_MODULE_MAP.clear()
N_MGR._paths_previously_scanned.clear()
N_MGR._custom_module_map.clear()
def test_exclusive_match():

8
test/test_asyncio.py
View File

@ -31,13 +31,15 @@ import pytest
from apprise import Apprise
from apprise import NotifyBase
from apprise import NotifyFormat
from apprise.common import NOTIFY_SCHEMA_MAP
from apprise.NotificationManager import NotificationManager
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
@pytest.mark.skipif(sys.version_info >= (3, 7),
reason="Requires Python 3.0 to 3.6")
@ -65,7 +67,7 @@ def test_apprise_asyncio_runtime_error():
return NotifyBase.parse_url(url, verify_host=False)
# Store our good notification in our schema map
NOTIFY_SCHEMA_MAP['good'] = GoodNotification
N_MGR['good'] = GoodNotification
# Create ourselves an Apprise object
a = Apprise()

7
test/test_attach_http.py
View File

@ -36,8 +36,8 @@ from os.path import dirname
from os.path import getsize
from apprise.attachment.AttachHTTP import AttachHTTP
from apprise import AppriseAttachment
from apprise.NotificationManager import NotificationManager
from apprise.plugins.NotifyBase import NotifyBase
from apprise.common import NOTIFY_SCHEMA_MAP
from apprise.common import ContentLocation
# Disable logging for a cleaner testing output
@ -46,6 +46,9 @@ logging.disable(logging.CRITICAL)
TEST_VAR_DIR = join(dirname(__file__), 'var')
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
# Some exception handling we'll use
REQUEST_EXCEPTIONS = (
requests.ConnectionError(
@ -131,7 +134,7 @@ def test_attach_http(mock_get):
return ''
# Store our good notification in our schema map
NOTIFY_SCHEMA_MAP['good'] = GoodNotification
N_MGR['good'] = GoodNotification
# Temporary path
path = join(TEST_VAR_DIR, 'apprise-test.gif')

6
test/test_config_http.py
View File

@ -34,12 +34,14 @@ import requests
from apprise.common import ConfigFormat
from apprise.config.ConfigHTTP import ConfigHTTP
from apprise.plugins.NotifyBase import NotifyBase
from apprise.common import NOTIFY_SCHEMA_MAP
from apprise.NotificationManager import NotificationManager
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
# Some exception handling we'll use
REQUEST_EXCEPTIONS = (
@ -77,7 +79,7 @@ def test_config_http(mock_post):
return ''
# Store our good notification in our schema map
NOTIFY_SCHEMA_MAP['good'] = GoodNotification
N_MGR['good'] = GoodNotification
# Our default content
default_content = """taga,tagb=good://server01"""

27
test/test_decorator_notify.py
View File

@ -34,11 +34,15 @@ from apprise import AppriseConfig
from apprise import AppriseAsset
from apprise import AppriseAttachment
from apprise import common
from apprise.NotificationManager import NotificationManager
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
TEST_VAR_DIR = join(dirname(__file__), 'var')
@ -48,7 +52,7 @@ def test_notify_simple_decoration():
# Verify our schema we're about to declare doesn't already exist
# in our schema map:
assert 'utiltest' not in common.NOTIFY_SCHEMA_MAP
assert 'utiltest' not in N_MGR
verify_obj = {}
@ -68,7 +72,7 @@ def test_notify_simple_decoration():
})
# Now after our hook being inline... it's been loaded
assert 'utiltest' in common.NOTIFY_SCHEMA_MAP
assert 'utiltest' in N_MGR
# Create ourselves an apprise object
aobj = Apprise()
@ -146,7 +150,7 @@ def test_notify_simple_decoration():
assert verify_obj['kwargs']['meta']['schema'] == 'utiltest'
assert verify_obj['kwargs']['meta']['url'] == 'utiltest://'
assert 'notexc' not in common.NOTIFY_SCHEMA_MAP
assert 'notexc' not in N_MGR
# Define a function here on the spot
@notify(on="notexc", name="Apprise @notify Exception Handling")
@ -154,7 +158,7 @@ def test_notify_simple_decoration():
body, title, notify_type, attach, *args, **kwargs):
raise ValueError("An exception was thrown!")
assert 'notexc' in common.NOTIFY_SCHEMA_MAP
assert 'notexc' in N_MGR
# Create ourselves an apprise object
aobj = Apprise()
@ -165,8 +169,7 @@ def test_notify_simple_decoration():
assert aobj.notify("Exceptions will be thrown!") is False
# Tidy
del common.NOTIFY_SCHEMA_MAP['utiltest']
del common.NOTIFY_SCHEMA_MAP['notexc']
N_MGR.remove('utiltest', 'notexc')
def test_notify_complex_decoration():
@ -175,7 +178,7 @@ def test_notify_complex_decoration():
# Verify our schema we're about to declare doesn't already exist
# in our schema map:
assert 'utiltest' not in common.NOTIFY_SCHEMA_MAP
assert 'utiltest' not in N_MGR
verify_obj = {}
@ -196,7 +199,7 @@ def test_notify_complex_decoration():
})
# Now after our hook being inline... it's been loaded
assert 'utiltest' in common.NOTIFY_SCHEMA_MAP
assert 'utiltest' in N_MGR
# Create ourselves an apprise object
aobj = Apprise()
@ -312,7 +315,7 @@ def test_notify_complex_decoration():
assert 'key2=another' in verify_obj['kwargs']['meta']['url']
# Tidy
del common.NOTIFY_SCHEMA_MAP['utiltest']
N_MGR.remove('utiltest')
def test_notify_multi_instance_decoration(tmpdir):
@ -321,7 +324,7 @@ def test_notify_multi_instance_decoration(tmpdir):
# Verify our schema we're about to declare doesn't already exist
# in our schema map:
assert 'multi' not in common.NOTIFY_SCHEMA_MAP
assert 'multi' not in N_MGR
verify_obj = []
@ -342,7 +345,7 @@ def test_notify_multi_instance_decoration(tmpdir):
})
# Now after our hook being inline... it's been loaded
assert 'multi' in common.NOTIFY_SCHEMA_MAP
assert 'multi' in N_MGR
# Prepare our config
t = tmpdir.mkdir("multi-test").join("apprise.yml")
@ -449,4 +452,4 @@ def test_notify_multi_instance_decoration(tmpdir):
assert meta['url'] == 'multi://user2:pass2@hostname'
# Tidy
del common.NOTIFY_SCHEMA_MAP['multi']
N_MGR.remove('multi')

389
test/test_notification_manager.py Normal file
View File

@ -0,0 +1,389 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import re
import pytest
import types
from inspect import cleandoc
from apprise.NotificationManager import NotificationManager
from apprise.plugins.NotifyBase import NotifyBase
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Grant access to our Notification Manager Singleton
N_MGR = NotificationManager()
def test_notification_manager_general():
"""
N_MGR: Notification Manager General testing
"""
# Clear our set so we can test init calls
N_MGR.unload_modules()
assert isinstance(N_MGR.schemas(), list)
assert len(N_MGR.schemas()) > 0
N_MGR.unload_modules(disable_native=True)
assert isinstance(N_MGR.schemas(), list)
assert len(N_MGR.schemas()) == 0
N_MGR.unload_modules()
assert len(N_MGR) > 0
N_MGR.unload_modules()
iter(N_MGR)
iter(N_MGR)
N_MGR.unload_modules()
assert bool(N_MGR) is False
assert len([x for x in iter(N_MGR)]) > 0
assert bool(N_MGR) is True
N_MGR.unload_modules()
assert isinstance(N_MGR.plugins(), types.GeneratorType)
assert len([x for x in N_MGR.plugins()]) > 0
N_MGR.unload_modules(disable_native=True)
assert isinstance(N_MGR.plugins(), types.GeneratorType)
assert len([x for x in N_MGR.plugins()]) == 0
N_MGR.unload_modules()
assert isinstance(N_MGR['json'](host='localhost'), NotifyBase)
N_MGR.unload_modules()
assert 'json' in N_MGR
# Define our good:// url
class DisabledNotification(NotifyBase):
# Always disabled
enabled = False
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def notify(self, *args, **kwargs):
# Pretend everything is okay
return True
def url(self, **kwargs):
# Support url() function
return ''
# Define our good:// url
class GoodNotification(NotifyBase):
secure_protocol = ('good', 'goods')
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def notify(self, *args, **kwargs):
# Pretend everything is okay
return True
def url(self, **kwargs):
# Support url() function
return ''
N_MGR.unload_modules()
assert N_MGR.add(GoodNotification)
assert 'good' in N_MGR
assert 'goods' in N_MGR
assert 'abcd' not in N_MGR
assert 'xyz' not in N_MGR
N_MGR.unload_modules()
assert N_MGR.add(GoodNotification, 'abcd')
assert 'good' in N_MGR
assert 'goods' in N_MGR
assert 'abcd' in N_MGR
assert 'xyz' not in N_MGR
N_MGR.unload_modules()
assert N_MGR.add(GoodNotification, ['abcd', 'xYz'])
assert 'good' in N_MGR
assert 'goods' in N_MGR
assert 'abcd' in N_MGR
# Lower case
assert 'xyz' in N_MGR
N_MGR.unload_modules()
# Not going to work; schemas must be a list of string
assert N_MGR.add(GoodNotification, object) is False
N_MGR.unload_modules()
with pytest.raises(KeyError):
del N_MGR['good']
N_MGR['good'] = GoodNotification
del N_MGR['good']
N_MGR.unload_modules()
N_MGR['good'] = GoodNotification
assert N_MGR['good'].enabled is True
N_MGR.enable_only('json', 'xml')
assert N_MGR['good'].enabled is False
assert N_MGR['json'].enabled is True
assert N_MGR['jsons'].enabled is True
assert N_MGR['xml'].enabled is True
assert N_MGR['xmls'].enabled is True
# Only two plugins are enabled
assert len([p for p in N_MGR.plugins(include_disabled=False)]) == 2
N_MGR.enable_only('good')
assert N_MGR['good'].enabled is True
assert N_MGR['json'].enabled is False
assert N_MGR['jsons'].enabled is False
assert N_MGR['xml'].enabled is False
assert N_MGR['xmls'].enabled is False
assert len([p for p in N_MGR.plugins(include_disabled=False)]) == 1
N_MGR.unload_modules()
N_MGR['disabled'] = DisabledNotification
assert N_MGR['disabled'].enabled is False
N_MGR.enable_only('disabled')
# Can't enable items that aren't supposed to be:
assert N_MGR['disabled'].enabled is False
N_MGR['good'] = GoodNotification
assert N_MGR['good'].enabled is True
# You can't disable someething already disabled
N_MGR.disable('disabled')
assert N_MGR['disabled'].enabled is False
N_MGR.unload_modules()
N_MGR.enable_only('form', 'xml')
for schema in N_MGR.schemas(include_disabled=False):
assert re.match(r'^(form|xml)s?$', schema, re.IGNORECASE) is not None
N_MGR.unload_modules()
assert N_MGR['form'].enabled is True
assert N_MGR['xml'].enabled is True
assert N_MGR['json'].enabled is True
N_MGR.enable_only('form', 'xml')
assert N_MGR['form'].enabled is True
assert N_MGR['xml'].enabled is True
assert N_MGR['json'].enabled is False
N_MGR.disable('invalid', 'xml')
assert N_MGR['form'].enabled is True
assert N_MGR['xml'].enabled is False
assert N_MGR['json'].enabled is False
# Detect that our json object is enabled
with pytest.raises(KeyError):
# The below can not be indexed
N_MGR['invalid']
N_MGR.unload_modules()
N_MGR.disable('invalid', 'xml')
N_MGR.unload_modules()
assert N_MGR['json'].enabled is True
# Work with an empty module tree
N_MGR.unload_modules(disable_native=True)
with pytest.raises(KeyError):
# The below can not be indexed
N_MGR['good']
N_MGR.unload_modules()
assert 'hello' not in N_MGR
assert 'good' not in N_MGR
assert 'goods' not in N_MGR
N_MGR['hello'] = GoodNotification
assert 'hello' in N_MGR
assert 'good' in N_MGR
assert 'goods' in N_MGR
N_MGR.unload_modules()
N_MGR['good'] = GoodNotification
with pytest.raises(KeyError):
# Can not assign the value again without getting a Conflict
N_MGR['good'] = GoodNotification
N_MGR.unload_modules()
N_MGR.remove('good', 'invalid')
assert 'good' not in N_MGR
assert 'goods' not in N_MGR
def test_notification_manager_module_loading(tmpdir):
"""
N_MGR: Notification Manager Module Loading
"""
# Handle loading modules twice (they gracefully handle not loading more in
# memory then needed)
N_MGR.load_modules()
N_MGR.load_modules()
def test_notification_manager_decorators(tmpdir):
"""
N_MGR: Notification Manager Decorator testing
"""
# Prepare ourselves a file to work with
notify_hook = tmpdir.mkdir('goodmodule').join('__init__.py')
notify_hook.write(cleandoc("""
from apprise.decorators import notify
# We want to trigger on anyone who configures a call to clihook://
@notify(on="clihooka")
def mywrapper(body, title, notify_type, *args, **kwargs):
# A simple test - print to screen
print("A {}: {} - {}".format(notify_type, title, body))
# No return (so a return of None) get's translated to True
# Define another in the same file; uppercase goes to lower
@notify(on="CLIhookb")
def mywrapper(body, title, notify_type, *args, **kwargs):
# A simple test - print to screen
print("B {}: {} - {}".format(notify_type, title, body))
# No return (so a return of None) get's translated to True
"""))
N_MGR.module_detection(str(notify_hook))
assert 'clihooka' in N_MGR
assert 'clihookb' in N_MGR
N_MGR.unload_modules()
assert 'clihooka' not in N_MGR
assert 'clihookb' not in N_MGR
N_MGR.module_detection(str(notify_hook))
assert 'clihooka' in N_MGR
assert 'clihookb' in N_MGR
del N_MGR['clihookb']
assert 'clihooka' in N_MGR
assert 'clihookb' not in N_MGR
del N_MGR['clihooka']
assert 'clihooka' not in N_MGR
assert 'clihookb' not in N_MGR
# Prepare ourselves a file to work with
notify_base = tmpdir.mkdir('plugins')
notify_test = notify_base.join('NotifyTest.py')
notify_test.write(cleandoc("""
#
# Bare Minimum Valid Object
#
from apprise.plugins.NotifyBase import NotifyBase
from apprise.common import NotifyType
class NotifyTest(NotifyBase):
service_name = 'Test'
# The services URL
service_url = 'https://github.com/caronc/apprise/'
# All boxcar notifications are secure
secure_protocol = 'mytest'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_mytest'
# Define object templates
templates = (
'{schema}://',
)
def __init__(self, **kwargs):
super().__init__(**kwargs)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
return True
def url(self):
return 'mytest://'
"""))
assert 'mytest' not in N_MGR
N_MGR.load_modules(path=str(notify_base))
assert 'mytest' in N_MGR
del N_MGR['mytest']
assert 'mytest' not in N_MGR
# Prepare ourselves a file to work with
notify_test = notify_base.join('NotifyNoAlign.py')
notify_test.write(cleandoc("""
#
# Bare Minimum Valid Object
#
from apprise.plugins.NotifyBase import NotifyBase
from apprise.common import NotifyType
class NotifyDifferentName(NotifyBase):
service_name = 'Unloadable'
# The services URL
service_url = 'https://github.com/caronc/apprise/'
# All boxcar notifications are secure
secure_protocol = 'noload'
# A URL that takes you to the setup/help of the specific protocol
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_mytest'
# Define object templates
templates = (
'{schema}://',
)
def __init__(self, **kwargs):
super().__init__(**kwargs)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
return True
def url(self):
return 'mytest://'
"""))
assert 'mytest' not in N_MGR
N_MGR.load_modules(path=str(notify_base))
assert 'mytest' in N_MGR
# Could not be loaded because the filename did not align with the class
# name.
assert 'noload' not in N_MGR
# Double load will test section of code that prevents a notification
# From reloading if previously already loaded
N_MGR.load_modules(path=str(notify_base))
# Our item is still loaded as expected
assert 'mytest' in N_MGR

12
test/test_plugin_glib.py
View File

@ -114,8 +114,7 @@ def setup_glib_environment():
# When patching something which has a side effect on the module-level code
# of a plugin, make sure to reload it.
current_module = sys.modules[__name__]
reload_plugin('NotifyDBus', replace_in=current_module)
reload_plugin('NotifyDBus')
@pytest.fixture
@ -416,8 +415,7 @@ def test_plugin_dbus_gi_missing(dbus_glib_environment):
# When patching something which has a side effect on the module-level code
# of a plugin, make sure to reload it.
current_module = sys.modules[__name__]
reload_plugin('NotifyDBus', replace_in=current_module)
reload_plugin('NotifyDBus')
# Create the instance.
obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False)
@ -444,8 +442,7 @@ def test_plugin_dbus_gi_require_version_error(dbus_glib_environment):
# When patching something which has a side effect on the module-level code
# of a plugin, make sure to reload it.
current_module = sys.modules[__name__]
reload_plugin('NotifyDBus', replace_in=current_module)
reload_plugin('NotifyDBus')
# Create instance.
obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False)
@ -472,8 +469,7 @@ def test_plugin_dbus_module_croaks(mocker, dbus_glib_environment):
# When patching something which has a side effect on the module-level code
# of a plugin, make sure to reload it.
current_module = sys.modules[__name__]
reload_plugin('NotifyDBus', replace_in=current_module)
reload_plugin('NotifyDBus')
# Verify plugin is not available.
obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False)

6
test/test_plugin_gnome.py
View File

@ -97,8 +97,7 @@ def setup_glib_environment():
# When patching something which has a side effect on the module-level code
# of a plugin, make sure to reload it.
current_module = sys.modules[__name__]
reload_plugin('NotifyGnome', replace_in=current_module)
reload_plugin('NotifyGnome')
@pytest.fixture
@ -345,8 +344,7 @@ def test_plugin_gnome_gi_croaks():
# When patching something which has a side effect on the module-level code
# of a plugin, make sure to reload it.
current_module = sys.modules[__name__]
reload_plugin('NotifyGnome', replace_in=current_module)
reload_plugin('NotifyGnome')
# Create instance.
obj = apprise.Apprise.instantiate('gnome://', suppress_exceptions=False)

6
test/test_plugin_macosx.py
View File

@ -43,7 +43,7 @@ logging.disable(logging.CRITICAL)
if sys.platform not in ["darwin", "linux"]:
pytest.skip("Only makes sense on macOS, but also works on Linux",
pytest.skip("Only makes sense on macOS, but testable in Linux",
allow_module_level=True)
@ -56,8 +56,7 @@ def pretend_macos(mocker):
mocker.patch("platform.mac_ver", return_value=('10.8', ('', '', ''), ''))
# Reload plugin module, in order to re-run module-level code.
current_module = sys.modules[__name__]
reload_plugin("NotifyMacOSX", replace_in=current_module)
reload_plugin("NotifyMacOSX")
@pytest.fixture
@ -93,6 +92,7 @@ def test_plugin_macosx_general_success(macos_notify_environment):
NotifyMacOSX() general checks
"""
# Toggle Enable Flag
obj = apprise.Apprise.instantiate(
'macosx://_/?image=True', suppress_exceptions=False)
assert isinstance(obj, NotifyMacOSX) is True