mirror of
https://github.com/caronc/apprise.git
synced 2024-10-04 17:22:18 +02:00
402 lines
12 KiB
Python
402 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
# BSD 2-Clause License
|
|
#
|
|
# Apprise - Push Notification Library.
|
|
# Copyright (c) 2024, 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
|
|
import threading
|
|
from inspect import cleandoc
|
|
|
|
from apprise import Apprise
|
|
from apprise import NotificationManager
|
|
from apprise.plugins 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)
|
|
|
|
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()
|
|
|
|
#
|
|
# Thread Testing
|
|
#
|
|
|
|
# This tests against a racing condition when the modules have not been
|
|
# loaded. When multiple instances of Apprise are all instantiated,
|
|
# the loading of the modules will occur for each instance if detected
|
|
# having not been previously done, this tests that we can dynamically
|
|
# support the loading of modules once whe multiple instances to apprise
|
|
# are instantiated.
|
|
thread_count = 10
|
|
|
|
def thread_test(result, no):
|
|
"""
|
|
Load our apprise object with valid URLs and store our result
|
|
"""
|
|
apobj = Apprise()
|
|
result[no] = apobj.add('json://localhost') and \
|
|
apobj.add('form://localhost') and \
|
|
apobj.add('xml://localhost')
|
|
|
|
# Unload our modules
|
|
N_MGR.unload_modules()
|
|
|
|
# Prepare threads to load
|
|
results = [None] * thread_count
|
|
threads = [
|
|
threading.Thread(target=thread_test, args=(results, no))
|
|
for no in range(thread_count)
|
|
]
|
|
|
|
# Verify we can safely load our modules in a thread safe environment
|
|
for t in threads:
|
|
t.start()
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# Verify we loaded our urls in all threads successfully
|
|
for result in results:
|
|
assert result is True
|
|
|
|
|
|
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 import NotifyBase
|
|
from apprise.common import NotifyType
|
|
|
|
class NotifyTest(NotifyBase):
|
|
|
|
service_name = 'Test'
|
|
|
|
# The services URL
|
|
service_url = 'https://github.com/caronc/apprise/'
|
|
|
|
# Define our protocol
|
|
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
|
|
|
|
assert 'mytest' not in N_MGR
|
|
N_MGR.load_modules(path=str(notify_base))
|
|
|
|
# It's still not loaded because the path has already been scanned
|
|
assert 'mytest' not in N_MGR
|
|
N_MGR.load_modules(path=str(notify_base), force=True)
|
|
assert 'mytest' 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
|
|
|
|
# Simple test to make sure we can handle duplicate entries loaded
|
|
N_MGR.load_modules(path=str(notify_base), force=True)
|
|
N_MGR.load_modules(path=str(notify_base), force=True)
|