mirror of https://github.com/caronc/apprise
402 lines
12 KiB
Python
402 lines
12 KiB
Python
# BSD 2-Clause License
|
|
#
|
|
# Apprise - Push Notification Library.
|
|
# Copyright (c) 2025, 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 inspect import cleandoc
|
|
|
|
# Disable logging for a cleaner testing output
|
|
import logging
|
|
import re
|
|
import threading
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from apprise import Apprise, NotificationManager
|
|
from apprise.plugins import NotifyBase
|
|
|
|
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(list(iter(N_MGR))) > 0
|
|
assert bool(N_MGR)
|
|
|
|
N_MGR.unload_modules()
|
|
assert isinstance(N_MGR.plugins(), types.GeneratorType)
|
|
assert len(list(N_MGR.plugins())) > 0
|
|
N_MGR.unload_modules(disable_native=True)
|
|
assert isinstance(N_MGR.plugins(), types.GeneratorType)
|
|
assert len(list(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(list(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(list(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)
|