mirror of
https://github.com/caronc/apprise.git
synced 2025-06-21 18:21:29 +02:00
slixmpp library support added to XMPP plugin (#240)
This commit is contained in:
parent
9e9b140642
commit
64a103f14a
218
apprise/plugins/NotifyXMPP/SliXmppAdapter.py
Normal file
218
apprise/plugins/NotifyXMPP/SliXmppAdapter.py
Normal file
@ -0,0 +1,218 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
import ssl
|
||||||
|
from os.path import isfile
|
||||||
|
import logging
|
||||||
|
|
||||||
|
|
||||||
|
# Default our global support flag
|
||||||
|
SLIXMPP_SUPPORT_AVAILABLE = False
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Import slixmpp if available
|
||||||
|
import slixmpp
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
SLIXMPP_SUPPORT_AVAILABLE = True
|
||||||
|
|
||||||
|
except ImportError:
|
||||||
|
# No problem; we just simply can't support this plugin because we're
|
||||||
|
# either using Linux, or simply do not have slixmpp installed.
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class SliXmppAdapter(object):
|
||||||
|
"""
|
||||||
|
Wrapper to slixmpp
|
||||||
|
|
||||||
|
"""
|
||||||
|
|
||||||
|
# Reference to XMPP client.
|
||||||
|
xmpp = None
|
||||||
|
|
||||||
|
# Whether everything succeeded
|
||||||
|
success = False
|
||||||
|
|
||||||
|
# The default protocol
|
||||||
|
protocol = 'xmpp'
|
||||||
|
|
||||||
|
# The default secure protocol
|
||||||
|
secure_protocol = 'xmpps'
|
||||||
|
|
||||||
|
# The default XMPP port
|
||||||
|
default_unsecure_port = 5222
|
||||||
|
|
||||||
|
# The default XMPP secure port
|
||||||
|
default_secure_port = 5223
|
||||||
|
|
||||||
|
# Taken from https://golang.org/src/crypto/x509/root_linux.go
|
||||||
|
CA_CERTIFICATE_FILE_LOCATIONS = [
|
||||||
|
# Debian/Ubuntu/Gentoo etc.
|
||||||
|
"/etc/ssl/certs/ca-certificates.crt",
|
||||||
|
# Fedora/RHEL 6
|
||||||
|
"/etc/pki/tls/certs/ca-bundle.crt",
|
||||||
|
# OpenSUSE
|
||||||
|
"/etc/ssl/ca-bundle.pem",
|
||||||
|
# OpenELEC
|
||||||
|
"/etc/pki/tls/cacert.pem",
|
||||||
|
# CentOS/RHEL 7
|
||||||
|
"/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem",
|
||||||
|
]
|
||||||
|
|
||||||
|
# This entry is a bit hacky, but it allows us to unit-test this library
|
||||||
|
# in an environment that simply doesn't have the slixmpp package
|
||||||
|
# available to us.
|
||||||
|
#
|
||||||
|
# If anyone is seeing this had knows a better way of testing this
|
||||||
|
# outside of what is defined in test/test_xmpp_plugin.py, please
|
||||||
|
# let me know! :)
|
||||||
|
_enabled = SLIXMPP_SUPPORT_AVAILABLE
|
||||||
|
|
||||||
|
def __init__(self, host=None, port=None, secure=False,
|
||||||
|
verify_certificate=True, xep=None, jid=None, password=None,
|
||||||
|
body=None, targets=None, before_message=None, logger=None):
|
||||||
|
"""
|
||||||
|
Initialize our SliXmppAdapter object
|
||||||
|
"""
|
||||||
|
|
||||||
|
self.host = host
|
||||||
|
self.port = port
|
||||||
|
self.secure = secure
|
||||||
|
self.verify_certificate = verify_certificate
|
||||||
|
|
||||||
|
self.xep = xep
|
||||||
|
self.jid = jid
|
||||||
|
self.password = password
|
||||||
|
|
||||||
|
self.body = body
|
||||||
|
self.targets = targets
|
||||||
|
self.before_message = before_message
|
||||||
|
|
||||||
|
self.logger = logger or logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# Use the Apprise log handlers for configuring the slixmpp logger.
|
||||||
|
apprise_logger = logging.getLogger('apprise')
|
||||||
|
sli_logger = logging.getLogger('slixmpp')
|
||||||
|
for handler in apprise_logger.handlers:
|
||||||
|
sli_logger.addHandler(handler)
|
||||||
|
sli_logger.setLevel(apprise_logger.level)
|
||||||
|
|
||||||
|
if not self.load():
|
||||||
|
raise ValueError("Invalid XMPP Configuration")
|
||||||
|
|
||||||
|
def load(self):
|
||||||
|
|
||||||
|
try:
|
||||||
|
asyncio.get_event_loop()
|
||||||
|
|
||||||
|
except RuntimeError:
|
||||||
|
# slixmpp can not handle not having an event_loop
|
||||||
|
# see: https://lab.louiz.org/poezio/slixmpp/-/issues/3456
|
||||||
|
# This is a work-around to this problem
|
||||||
|
asyncio.set_event_loop(asyncio.new_event_loop())
|
||||||
|
|
||||||
|
# Prepare our object
|
||||||
|
self.xmpp = slixmpp.ClientXMPP(self.jid, self.password)
|
||||||
|
|
||||||
|
# Register our session
|
||||||
|
self.xmpp.add_event_handler("session_start", self.session_start)
|
||||||
|
|
||||||
|
for xep in self.xep:
|
||||||
|
# Load xep entries
|
||||||
|
try:
|
||||||
|
self.xmpp.register_plugin('xep_{0:04d}'.format(xep))
|
||||||
|
|
||||||
|
except slixmpp.plugins.base.PluginNotFound:
|
||||||
|
self.logger.warning(
|
||||||
|
'Could not register plugin {}'.format(
|
||||||
|
'xep_{0:04d}'.format(xep)))
|
||||||
|
return False
|
||||||
|
|
||||||
|
if self.secure:
|
||||||
|
# Don't even try to use the outdated ssl.PROTOCOL_SSLx
|
||||||
|
self.xmpp.ssl_version = ssl.PROTOCOL_TLSv1
|
||||||
|
|
||||||
|
# If the python version supports it, use highest TLS version
|
||||||
|
# automatically
|
||||||
|
if hasattr(ssl, "PROTOCOL_TLS"):
|
||||||
|
# Use the best version of TLS available to us
|
||||||
|
self.xmpp.ssl_version = ssl.PROTOCOL_TLS
|
||||||
|
|
||||||
|
self.xmpp.ca_certs = None
|
||||||
|
if self.verify_certificate:
|
||||||
|
# Set the ca_certs variable for certificate verification
|
||||||
|
self.xmpp.ca_certs = next(
|
||||||
|
(cert for cert in self.CA_CERTIFICATE_FILE_LOCATIONS
|
||||||
|
if isfile(cert)), None)
|
||||||
|
|
||||||
|
if self.xmpp.ca_certs is None:
|
||||||
|
self.logger.warning(
|
||||||
|
'XMPP Secure comunication can not be verified; '
|
||||||
|
'no local CA certificate file')
|
||||||
|
return False
|
||||||
|
|
||||||
|
# We're good
|
||||||
|
return True
|
||||||
|
|
||||||
|
def process(self):
|
||||||
|
"""
|
||||||
|
Thread that handles the server/client i/o
|
||||||
|
|
||||||
|
"""
|
||||||
|
|
||||||
|
# Establish connection to XMPP server.
|
||||||
|
# To speed up sending messages, don't use the "reattempt" feature,
|
||||||
|
# it will add a nasty delay even before connecting to XMPP server.
|
||||||
|
if not self.xmpp.connect((self.host, self.port),
|
||||||
|
use_ssl=self.secure, reattempt=False):
|
||||||
|
|
||||||
|
default_port = self.default_secure_port \
|
||||||
|
if self.secure else self.default_unsecure_port
|
||||||
|
|
||||||
|
default_schema = self.secure_protocol \
|
||||||
|
if self.secure else self.protocol
|
||||||
|
|
||||||
|
# Log connection issue
|
||||||
|
self.logger.warning(
|
||||||
|
'Failed to authenticate {jid} with: {schema}://{host}{port}'
|
||||||
|
.format(
|
||||||
|
jid=self.jid,
|
||||||
|
schema=default_schema,
|
||||||
|
host=self.host,
|
||||||
|
port='' if not self.port or self.port == default_port
|
||||||
|
else ':{}'.format(self.port),
|
||||||
|
))
|
||||||
|
return False
|
||||||
|
|
||||||
|
# Process XMPP communication.
|
||||||
|
self.xmpp.process(block=True)
|
||||||
|
|
||||||
|
return self.success
|
||||||
|
|
||||||
|
def session_start(self, *args, **kwargs):
|
||||||
|
"""
|
||||||
|
Session Manager
|
||||||
|
"""
|
||||||
|
|
||||||
|
targets = list(self.targets)
|
||||||
|
if not targets:
|
||||||
|
# We always default to notifying ourselves
|
||||||
|
targets.append(self.jid)
|
||||||
|
|
||||||
|
while len(targets) > 0:
|
||||||
|
|
||||||
|
# Get next target (via JID)
|
||||||
|
target = targets.pop(0)
|
||||||
|
|
||||||
|
# Invoke "before_message" event hook.
|
||||||
|
self.before_message()
|
||||||
|
|
||||||
|
# The message we wish to send, and the JID that will receive it.
|
||||||
|
self.xmpp.send_message(mto=target, mbody=self.body, mtype='chat')
|
||||||
|
|
||||||
|
# Using wait=True ensures that the send queue will be
|
||||||
|
# emptied before ending the session.
|
||||||
|
self.xmpp.disconnect(wait=True)
|
||||||
|
|
||||||
|
# Toggle our success flag
|
||||||
|
self.success = True
|
@ -31,6 +31,7 @@ from ...common import NotifyType
|
|||||||
from ...utils import parse_list
|
from ...utils import parse_list
|
||||||
from ...AppriseLocale import gettext_lazy as _
|
from ...AppriseLocale import gettext_lazy as _
|
||||||
from .SleekXmppAdapter import SleekXmppAdapter
|
from .SleekXmppAdapter import SleekXmppAdapter
|
||||||
|
from .SliXmppAdapter import SliXmppAdapter
|
||||||
|
|
||||||
# xep string parser
|
# xep string parser
|
||||||
XEP_PARSE_RE = re.compile('^[^1-9]*(?P<xep>[1-9][0-9]{0,3})$')
|
XEP_PARSE_RE = re.compile('^[^1-9]*(?P<xep>[1-9][0-9]{0,3})$')
|
||||||
@ -72,7 +73,8 @@ class NotifyXMPP(NotifyBase):
|
|||||||
# If anyone is seeing this had knows a better way of testing this
|
# If anyone is seeing this had knows a better way of testing this
|
||||||
# outside of what is defined in test/test_xmpp_plugin.py, please
|
# outside of what is defined in test/test_xmpp_plugin.py, please
|
||||||
# let me know! :)
|
# let me know! :)
|
||||||
_enabled = SleekXmppAdapter._enabled
|
_enabled = SleekXmppAdapter._enabled or SliXmppAdapter._enabled
|
||||||
|
_adapter = SliXmppAdapter if SliXmppAdapter._enabled else SleekXmppAdapter
|
||||||
|
|
||||||
# Define object templates
|
# Define object templates
|
||||||
templates = (
|
templates = (
|
||||||
@ -226,7 +228,7 @@ class NotifyXMPP(NotifyBase):
|
|||||||
if not self._enabled:
|
if not self._enabled:
|
||||||
self.logger.warning(
|
self.logger.warning(
|
||||||
'XMPP Notifications are not supported by this system '
|
'XMPP Notifications are not supported by this system '
|
||||||
'- install sleekxmpp.')
|
'- install sleekxmpp or slixmpp.')
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Detect our JID if it isn't otherwise specified
|
# Detect our JID if it isn't otherwise specified
|
||||||
@ -252,7 +254,7 @@ class NotifyXMPP(NotifyBase):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
# Communicate with XMPP.
|
# Communicate with XMPP.
|
||||||
xmpp_adapter = SleekXmppAdapter(
|
xmpp_adapter = self._adapter(
|
||||||
host=self.host, port=port, secure=self.secure,
|
host=self.host, port=port, secure=self.secure,
|
||||||
verify_certificate=self.verify_certificate, xep=self.xep,
|
verify_certificate=self.verify_certificate, xep=self.xep,
|
||||||
jid=jid, password=password, body=body, targets=self.targets,
|
jid=jid, password=password, body=body, targets=self.targets,
|
||||||
|
@ -34,6 +34,7 @@ from os.path import abspath
|
|||||||
# Used for testing
|
# Used for testing
|
||||||
from . import NotifyEmail as NotifyEmailBase
|
from . import NotifyEmail as NotifyEmailBase
|
||||||
from .NotifyXMPP import SleekXmppAdapter
|
from .NotifyXMPP import SleekXmppAdapter
|
||||||
|
from .NotifyXMPP import SliXmppAdapter
|
||||||
|
|
||||||
# NotifyBase object is passed in as a module not class
|
# NotifyBase object is passed in as a module not class
|
||||||
from . import NotifyBase
|
from . import NotifyBase
|
||||||
@ -64,6 +65,9 @@ __all__ = [
|
|||||||
|
|
||||||
# sleekxmpp access points (used for NotifyXMPP Testing)
|
# sleekxmpp access points (used for NotifyXMPP Testing)
|
||||||
'SleekXmppAdapter',
|
'SleekXmppAdapter',
|
||||||
|
|
||||||
|
# slixmpp access points (used for NotifyXMPP Testing)
|
||||||
|
'SliXmppAdapter',
|
||||||
]
|
]
|
||||||
|
|
||||||
# we mirror our base purely for the ability to reset everything; this
|
# we mirror our base purely for the ability to reset everything; this
|
||||||
|
@ -12,6 +12,7 @@ babel
|
|||||||
|
|
||||||
# Provides xmpp:// support
|
# Provides xmpp:// support
|
||||||
sleekxmpp
|
sleekxmpp
|
||||||
|
slixmpp; python_version >= '3.7'
|
||||||
|
|
||||||
# Provides growl:// support
|
# Provides growl:// support
|
||||||
gntp
|
gntp
|
||||||
|
@ -50,7 +50,7 @@ logging.disable(logging.CRITICAL)
|
|||||||
|
|
||||||
@pytest.mark.skipif(
|
@pytest.mark.skipif(
|
||||||
'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
|
'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
|
||||||
def test_xmpp_plugin_import_error(tmpdir):
|
def test_sleekxmpp_plugin_import_error(tmpdir):
|
||||||
"""
|
"""
|
||||||
API: NotifyXMPP Plugin() Import Error
|
API: NotifyXMPP Plugin() Import Error
|
||||||
|
|
||||||
@ -113,6 +113,7 @@ def test_xmpp_plugin_import_error(tmpdir):
|
|||||||
sys.path.remove(str(suite))
|
sys.path.remove(str(suite))
|
||||||
|
|
||||||
# Reload the libraries we care about
|
# Reload the libraries we care about
|
||||||
|
reload(sys.modules['apprise.plugins.NotifyXMPP.SliXmppAdapter'])
|
||||||
reload(sys.modules['apprise.plugins.NotifyXMPP.SleekXmppAdapter'])
|
reload(sys.modules['apprise.plugins.NotifyXMPP.SleekXmppAdapter'])
|
||||||
reload(sys.modules['apprise.plugins.NotifyXMPP'])
|
reload(sys.modules['apprise.plugins.NotifyXMPP'])
|
||||||
reload(sys.modules['apprise.plugins'])
|
reload(sys.modules['apprise.plugins'])
|
||||||
@ -122,7 +123,7 @@ def test_xmpp_plugin_import_error(tmpdir):
|
|||||||
|
|
||||||
@pytest.mark.skipif(
|
@pytest.mark.skipif(
|
||||||
'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
|
'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
|
||||||
def test_xmpp_plugin(tmpdir):
|
def test_sleekxmpp_plugin(tmpdir):
|
||||||
"""
|
"""
|
||||||
API: NotifyXMPP Plugin()
|
API: NotifyXMPP Plugin()
|
||||||
"""
|
"""
|
||||||
@ -130,6 +131,9 @@ def test_xmpp_plugin(tmpdir):
|
|||||||
# Set success flag
|
# Set success flag
|
||||||
apprise.plugins.SleekXmppAdapter.success = True
|
apprise.plugins.SleekXmppAdapter.success = True
|
||||||
|
|
||||||
|
# Enforce Adapter
|
||||||
|
apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SleekXmppAdapter
|
||||||
|
|
||||||
# Create a restore point
|
# Create a restore point
|
||||||
ca_backup = apprise.plugins.SleekXmppAdapter\
|
ca_backup = apprise.plugins.SleekXmppAdapter\
|
||||||
.CA_CERTIFICATE_FILE_LOCATIONS
|
.CA_CERTIFICATE_FILE_LOCATIONS
|
||||||
@ -298,7 +302,7 @@ def test_xmpp_plugin(tmpdir):
|
|||||||
obj._enabled = True
|
obj._enabled = True
|
||||||
|
|
||||||
# create an empty file for now
|
# create an empty file for now
|
||||||
ca_cert = tmpdir.mkdir("apprise_xmpp_test").join('ca_cert')
|
ca_cert = tmpdir.mkdir("apprise_sleekxmpp_test").join('ca_cert')
|
||||||
ca_cert.write('')
|
ca_cert.write('')
|
||||||
|
|
||||||
# Update our path
|
# Update our path
|
||||||
@ -361,6 +365,9 @@ def test_sleekxmpp_callbacks():
|
|||||||
# Set success flag
|
# Set success flag
|
||||||
apprise.plugins.SleekXmppAdapter.success = False
|
apprise.plugins.SleekXmppAdapter.success = False
|
||||||
|
|
||||||
|
# Enforce Adapter
|
||||||
|
apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SleekXmppAdapter
|
||||||
|
|
||||||
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
|
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
|
||||||
client_stream = mock.Mock()
|
client_stream = mock.Mock()
|
||||||
client_stream.send_message.return_value = True
|
client_stream.send_message.return_value = True
|
||||||
@ -397,3 +404,361 @@ def test_sleekxmpp_callbacks():
|
|||||||
kwargs['xep'] = [1, 999]
|
kwargs['xep'] = [1, 999]
|
||||||
with pytest.raises(ValueError):
|
with pytest.raises(ValueError):
|
||||||
apprise.plugins.SleekXmppAdapter(**kwargs)
|
apprise.plugins.SleekXmppAdapter(**kwargs)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
'slixmpp' not in sys.modules, reason="requires slixmpp")
|
||||||
|
def test_slixmpp_plugin_import_error(tmpdir):
|
||||||
|
"""
|
||||||
|
API: NotifyXMPP Plugin() Import Error
|
||||||
|
|
||||||
|
"""
|
||||||
|
# This is a really confusing test case; it can probably be done better,
|
||||||
|
# but this was all I could come up with. Effectively Apprise is will
|
||||||
|
# still work flawlessly without the slixmpp dependancy. Since
|
||||||
|
# slixmpp is actually required to be installed to run these unit tests
|
||||||
|
# we need to do some hacky tricks into fooling our test cases that the
|
||||||
|
# package isn't available.
|
||||||
|
|
||||||
|
# So we create a temporary directory called slixmpp (simulating the
|
||||||
|
# library itself) and writing an __init__.py in it that does nothing
|
||||||
|
# but throw an ImportError exception (simulating that hte library
|
||||||
|
# isn't found).
|
||||||
|
suite = tmpdir.mkdir("slixmpp")
|
||||||
|
suite.join("__init__.py").write('')
|
||||||
|
module_name = 'slixmpp'
|
||||||
|
suite.join("{}.py".format(module_name)).write('raise ImportError()')
|
||||||
|
|
||||||
|
# The second part of the test is to update our PYTHON_PATH to look
|
||||||
|
# into this new directory first (before looking where the actual
|
||||||
|
# valid paths are). This will allow us to override 'JUST' the slixmpp
|
||||||
|
# path.
|
||||||
|
|
||||||
|
# Update our path to point to our new test suite
|
||||||
|
sys.path.insert(0, str(suite))
|
||||||
|
|
||||||
|
# We need to remove the slixmpp modules that have already been loaded
|
||||||
|
# in memory otherwise they'll just be used instead. Python is smart and
|
||||||
|
# won't go try and reload everything again if it doesn't have to.
|
||||||
|
for name in list(sys.modules.keys()):
|
||||||
|
if name.startswith('{}.'.format(module_name)):
|
||||||
|
del sys.modules[name]
|
||||||
|
del sys.modules[module_name]
|
||||||
|
del sys.modules['apprise.plugins.NotifyXMPP.SliXmppAdapter']
|
||||||
|
|
||||||
|
# The following libraries need to be reloaded to prevent
|
||||||
|
# TypeError: super(type, obj): obj must be an instance or subtype of type
|
||||||
|
# This is better explained in this StackOverflow post:
|
||||||
|
# https://stackoverflow.com/questions/31363311/\
|
||||||
|
# any-way-to-manually-fix-operation-of-\
|
||||||
|
# super-after-ipython-reload-avoiding-ty
|
||||||
|
#
|
||||||
|
reload(sys.modules['apprise.plugins.NotifyXMPP'])
|
||||||
|
reload(sys.modules['apprise.plugins'])
|
||||||
|
reload(sys.modules['apprise.Apprise'])
|
||||||
|
reload(sys.modules['apprise'])
|
||||||
|
|
||||||
|
# This tests that Apprise still works without slixmpp.
|
||||||
|
# XMPP objects can't be istantiated though.
|
||||||
|
obj = apprise.Apprise.instantiate('xmpp://user:pass@localhost')
|
||||||
|
assert obj is not None
|
||||||
|
|
||||||
|
# Tidy-up / restore things to how they were
|
||||||
|
# Remove our garbage library
|
||||||
|
os.unlink(str(suite.join("{}.py".format(module_name))))
|
||||||
|
|
||||||
|
# Remove our custom entry into the path
|
||||||
|
sys.path.remove(str(suite))
|
||||||
|
|
||||||
|
# Reload the libraries we care about
|
||||||
|
reload(sys.modules['apprise.plugins.NotifyXMPP.SliXmppAdapter'])
|
||||||
|
reload(sys.modules['apprise.plugins.NotifyXMPP.SleekXmppAdapter'])
|
||||||
|
reload(sys.modules['apprise.plugins.NotifyXMPP'])
|
||||||
|
reload(sys.modules['apprise.plugins'])
|
||||||
|
reload(sys.modules['apprise.Apprise'])
|
||||||
|
reload(sys.modules['apprise'])
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
'slixmpp' not in sys.modules, reason="requires slixmpp")
|
||||||
|
def test_slixmpp_plugin(tmpdir):
|
||||||
|
"""
|
||||||
|
API: NotifyXMPP Plugin()
|
||||||
|
"""
|
||||||
|
|
||||||
|
# Set success flag
|
||||||
|
apprise.plugins.SliXmppAdapter.success = True
|
||||||
|
|
||||||
|
# Enforce Adapter
|
||||||
|
apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SliXmppAdapter
|
||||||
|
|
||||||
|
# Create a restore point
|
||||||
|
ca_backup = apprise.plugins.SliXmppAdapter\
|
||||||
|
.CA_CERTIFICATE_FILE_LOCATIONS
|
||||||
|
|
||||||
|
# Clear CA Certificates
|
||||||
|
apprise.plugins.SliXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = []
|
||||||
|
|
||||||
|
# Disable Throttling to speed testing
|
||||||
|
apprise.plugins.NotifyBase.request_rate_per_sec = 0
|
||||||
|
|
||||||
|
# Create our instance
|
||||||
|
obj = apprise.Apprise.instantiate('xmpp://', suppress_exceptions=False)
|
||||||
|
|
||||||
|
# Not possible because no password or host was specified
|
||||||
|
assert obj is None
|
||||||
|
|
||||||
|
with pytest.raises(TypeError):
|
||||||
|
apprise.Apprise.instantiate(
|
||||||
|
'xmpp://hostname', suppress_exceptions=False)
|
||||||
|
|
||||||
|
# SSL Flags
|
||||||
|
if hasattr(ssl, "PROTOCOL_TLS"):
|
||||||
|
# Test cases where PROTOCOL_TLS simply isn't available
|
||||||
|
ssl_temp_swap = ssl.PROTOCOL_TLS
|
||||||
|
del ssl.PROTOCOL_TLS
|
||||||
|
|
||||||
|
# Test our URL
|
||||||
|
url = 'xmpps://user:pass@127.0.0.1'
|
||||||
|
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
|
||||||
|
|
||||||
|
# Test we loaded
|
||||||
|
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
|
||||||
|
|
||||||
|
# Check that it found our mocked environments
|
||||||
|
assert obj._enabled is True
|
||||||
|
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_stream:
|
||||||
|
client_stream = mock.Mock()
|
||||||
|
client_stream.connect.return_value = True
|
||||||
|
mock_stream.return_value = client_stream
|
||||||
|
|
||||||
|
# We fail because we could not verify the host
|
||||||
|
assert obj.notify(
|
||||||
|
title='title', body='body',
|
||||||
|
notify_type=apprise.NotifyType.INFO) is False
|
||||||
|
|
||||||
|
# Restore the variable for remaining tests
|
||||||
|
setattr(ssl, 'PROTOCOL_TLS', ssl_temp_swap)
|
||||||
|
|
||||||
|
else:
|
||||||
|
# Handle case where it is not missing
|
||||||
|
setattr(ssl, 'PROTOCOL_TLS', ssl.PROTOCOL_TLSv1)
|
||||||
|
# Test our URL
|
||||||
|
url = 'xmpps://user:pass@localhost'
|
||||||
|
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
|
||||||
|
|
||||||
|
# Test we loaded
|
||||||
|
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
|
||||||
|
|
||||||
|
# Check that it found our mocked environments
|
||||||
|
assert obj._enabled is True
|
||||||
|
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_stream:
|
||||||
|
client_stream = mock.Mock()
|
||||||
|
client_stream.connect.return_value = True
|
||||||
|
mock_stream.return_value = client_stream
|
||||||
|
|
||||||
|
assert obj.notify(
|
||||||
|
title='title', body='body',
|
||||||
|
notify_type=apprise.NotifyType.INFO) is True
|
||||||
|
|
||||||
|
# Restore settings as they were
|
||||||
|
del ssl.PROTOCOL_TLS
|
||||||
|
|
||||||
|
urls = (
|
||||||
|
{
|
||||||
|
'u': 'xmpp://user:pass@localhost',
|
||||||
|
'p': 'xmpp://user:****@localhost',
|
||||||
|
}, {
|
||||||
|
'u': 'xmpp://user:pass@localhost?'
|
||||||
|
'xep=30,199,garbage,xep_99999999',
|
||||||
|
'p': 'xmpp://user:****@localhost',
|
||||||
|
}, {
|
||||||
|
'u': 'xmpps://user:pass@localhost?xep=ignored&verify=no',
|
||||||
|
'p': 'xmpps://user:****@localhost',
|
||||||
|
}, {
|
||||||
|
'u': 'xmpps://pass@localhost/?verify=false'
|
||||||
|
'user@test.com, user2@test.com/resource',
|
||||||
|
'p': 'xmpps://****@localhost',
|
||||||
|
}, {
|
||||||
|
'u': 'xmpps://pass@localhost:5226?jid=user@test.com&verify=no',
|
||||||
|
'p': 'xmpps://****@localhost:5226',
|
||||||
|
}, {
|
||||||
|
'u': 'xmpps://pass@localhost?jid=user@test.com&verify=False',
|
||||||
|
'p': 'xmpps://****@localhost',
|
||||||
|
}, {
|
||||||
|
'u': 'xmpps://user:pass@localhost?verify=False',
|
||||||
|
'p': 'xmpps://user:****@localhost',
|
||||||
|
}, {
|
||||||
|
'u': 'xmpp://user:pass@localhost?to=user@test.com&verify=no',
|
||||||
|
'p': 'xmpp://user:****@localhost',
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
# Try Different Variations of our URL
|
||||||
|
for entry in urls:
|
||||||
|
|
||||||
|
url = entry['u']
|
||||||
|
privacy_url = entry['p']
|
||||||
|
|
||||||
|
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
|
||||||
|
|
||||||
|
# Test we loaded
|
||||||
|
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
|
||||||
|
|
||||||
|
# Check that it found our mocked environments
|
||||||
|
assert obj._enabled is True
|
||||||
|
|
||||||
|
# Test url() call
|
||||||
|
assert isinstance(obj.url(), six.string_types) is True
|
||||||
|
|
||||||
|
# Test url(privacy=True) call
|
||||||
|
assert isinstance(obj.url(privacy=True), six.string_types) is True
|
||||||
|
|
||||||
|
assert obj.url(privacy=True).startswith(privacy_url)
|
||||||
|
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_stream:
|
||||||
|
client_stream = mock.Mock()
|
||||||
|
client_stream.connect.return_value = True
|
||||||
|
mock_stream.return_value = client_stream
|
||||||
|
|
||||||
|
# test notifications
|
||||||
|
assert obj.notify(
|
||||||
|
title='title', body='body',
|
||||||
|
notify_type=apprise.NotifyType.INFO) is True
|
||||||
|
|
||||||
|
# test notification without a title
|
||||||
|
assert obj.notify(
|
||||||
|
title='', body='body',
|
||||||
|
notify_type=apprise.NotifyType.INFO) is True
|
||||||
|
|
||||||
|
# Test Connection Failure
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_stream:
|
||||||
|
client_stream = mock.Mock()
|
||||||
|
client_stream.connect.return_value = False
|
||||||
|
mock_stream.return_value = client_stream
|
||||||
|
|
||||||
|
# test notifications
|
||||||
|
assert obj.notify(
|
||||||
|
title='title', body='body',
|
||||||
|
notify_type=apprise.NotifyType.INFO) is False
|
||||||
|
|
||||||
|
# Toggle our _enabled flag
|
||||||
|
obj._enabled = False
|
||||||
|
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_client:
|
||||||
|
# Allow a connection to succeed
|
||||||
|
mock_client.connect.return_value = True
|
||||||
|
|
||||||
|
# Verify that we can't send content now
|
||||||
|
assert obj.notify(
|
||||||
|
title='', body='body',
|
||||||
|
notify_type=apprise.NotifyType.INFO) is False
|
||||||
|
|
||||||
|
# Toggle it back so it doesn't disrupt other testing
|
||||||
|
obj._enabled = True
|
||||||
|
|
||||||
|
# create an empty file for now
|
||||||
|
ca_cert = tmpdir.mkdir("apprise_slixmpp_test").join('ca_cert')
|
||||||
|
ca_cert.write('')
|
||||||
|
|
||||||
|
# Update our path
|
||||||
|
apprise.plugins.SliXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = \
|
||||||
|
[str(ca_cert), ]
|
||||||
|
|
||||||
|
obj = apprise.Apprise.instantiate(
|
||||||
|
'xmpps://pass@localhost/user@test.com?verify=yes',
|
||||||
|
suppress_exceptions=False)
|
||||||
|
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
|
||||||
|
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_client:
|
||||||
|
# Allow a connection to succeed
|
||||||
|
mock_client.connect.return_value = True
|
||||||
|
# Our notification now should be able to get a ca_cert to reference
|
||||||
|
assert obj.notify(
|
||||||
|
title='', body='body', notify_type=apprise.NotifyType.INFO) is True
|
||||||
|
|
||||||
|
# Restore our CA Certificates from backup
|
||||||
|
apprise.plugins.SliXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = \
|
||||||
|
ca_backup
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(
|
||||||
|
'slixmpp' not in sys.modules, reason="requires slixmpp")
|
||||||
|
def test_slixmpp_callbacks():
|
||||||
|
"""
|
||||||
|
API: NotifyXMPP Plugin() Slixmpp callback tests
|
||||||
|
|
||||||
|
The tests identified here just test the basic callbacks defined for
|
||||||
|
slixmpp. Emulating a full xmpp server in order to test this plugin
|
||||||
|
proved to be difficult so just here are some basic tests to make sure code
|
||||||
|
doesn't produce any exceptions. This is a perfect solution to get 100%
|
||||||
|
test coverage of the NotifyXMPP plugin, but it's better than nothing at
|
||||||
|
all.
|
||||||
|
"""
|
||||||
|
def dummy_before_message():
|
||||||
|
# Just a dummy function for testing purposes
|
||||||
|
return
|
||||||
|
|
||||||
|
kwargs = {
|
||||||
|
'host': 'localhost',
|
||||||
|
'port': 5555,
|
||||||
|
'secure': False,
|
||||||
|
'verify_certificate': False,
|
||||||
|
'xep': [
|
||||||
|
# xep_0030: Service Discovery
|
||||||
|
30,
|
||||||
|
# xep_0199: XMPP Ping
|
||||||
|
199,
|
||||||
|
],
|
||||||
|
'jid': 'user@localhost',
|
||||||
|
'password': 'secret!',
|
||||||
|
'body': 'my message to delivery!',
|
||||||
|
'targets': ['user2@localhost'],
|
||||||
|
'before_message': dummy_before_message,
|
||||||
|
'logger': None,
|
||||||
|
}
|
||||||
|
|
||||||
|
# Set success flag
|
||||||
|
apprise.plugins.SliXmppAdapter.success = False
|
||||||
|
|
||||||
|
# Enforce Adapter
|
||||||
|
apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SliXmppAdapter
|
||||||
|
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_stream:
|
||||||
|
client_stream = mock.Mock()
|
||||||
|
client_stream.send_message.return_value = True
|
||||||
|
mock_stream.return_value = client_stream
|
||||||
|
|
||||||
|
adapter = apprise.plugins.SliXmppAdapter(**kwargs)
|
||||||
|
assert isinstance(adapter, apprise.plugins.SliXmppAdapter)
|
||||||
|
|
||||||
|
# Ensure we are initialized in a failure state; our return flag after
|
||||||
|
# we actually attempt to send the notification(s). This get's toggled
|
||||||
|
# to true only after a session_start() call is done successfully
|
||||||
|
assert adapter.success is False
|
||||||
|
adapter.session_start()
|
||||||
|
assert adapter.success is True
|
||||||
|
|
||||||
|
# Now we'll do a test with no one to notify
|
||||||
|
kwargs['targets'] = []
|
||||||
|
adapter = apprise.plugins.SliXmppAdapter(**kwargs)
|
||||||
|
assert isinstance(adapter, apprise.plugins.SliXmppAdapter)
|
||||||
|
|
||||||
|
# success flag should be back to a False state
|
||||||
|
assert adapter.success is False
|
||||||
|
|
||||||
|
with mock.patch('slixmpp.ClientXMPP') as mock_stream:
|
||||||
|
client_stream = mock.Mock()
|
||||||
|
client_stream.send_message.return_value = True
|
||||||
|
mock_stream.return_value = client_stream
|
||||||
|
adapter.session_start()
|
||||||
|
# success flag changes to True
|
||||||
|
assert adapter.success is True
|
||||||
|
|
||||||
|
# Restore our target, but set up invalid xep codes
|
||||||
|
kwargs['targets'] = ['user2@localhost']
|
||||||
|
kwargs['xep'] = [1, 999]
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
apprise.plugins.SliXmppAdapter(**kwargs)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user