diff --git a/README.md b/README.md
index ffceb291..b29b8a71 100644
--- a/README.md
+++ b/README.md
@@ -87,7 +87,7 @@ The table below identifies the services this tool supports and some example serv
| [Twitter](https://github.com/caronc/apprise/wiki/Notify_twitter) | twitter:// | (TCP) 443 | twitter://CKey/CSecret/AKey/ASecret
twitter://user@CKey/CSecret/AKey/ASecret
twitter://CKey/CSecret/AKey/ASecret/User1/User2/User2
twitter://CKey/CSecret/AKey/ASecret?mode=tweet
| [Twist](https://github.com/caronc/apprise/wiki/Notify_twist) | twist:// | (TCP) 443 | twist://pasword:login
twist://password:login/#channel
twist://password:login/#team:channel
twist://password:login/#team:channel1/channel2/#team3:channel
| [XBMC](https://github.com/caronc/apprise/wiki/Notify_xbmc) | xbmc:// or xbmcs:// | (TCP) 8080 or 443 | xbmc://hostname
xbmc://user@hostname
xbmc://user:password@hostname:port
-| [XMPP](https://github.com/caronc/apprise/wiki/Notify_xmpp) | xmpp:// or xmpps:// | (TCP) 5222 or 5223 | xmpp://password@hostname
xmpp://user:password@hostname
xmpps://user:password@hostname:port?jid=user@hostname/resource
xmpps://password@hostname/target@myhost, target2@myhost/resource
+| [XMPP](https://github.com/caronc/apprise/wiki/Notify_xmpp) | xmpp:// or xmpps:// | (TCP) 5222 or 5223 | xmpp://user:password@hostname
xmpps://user:password@hostname:port?jid=user@hostname/resource
xmpps://user:password@hostname/target@myhost, target2@myhost/resource
| [Webex Teams (Cisco)](https://github.com/caronc/apprise/wiki/Notify_wxteams) | wxteams:// | (TCP) 443 | wxteams://Token
| [Zulip Chat](https://github.com/caronc/apprise/wiki/Notify_zulip) | zulip:// | (TCP) 443 | zulip://botname@Organization/Token
zulip://botname@Organization/Token/Stream
zulip://botname@Organization/Token/Email
diff --git a/apprise/plugins/NotifyXMPP/SleekXmppAdapter.py b/apprise/plugins/NotifyXMPP/SleekXmppAdapter.py
deleted file mode 100644
index a28e9ce5..00000000
--- a/apprise/plugins/NotifyXMPP/SleekXmppAdapter.py
+++ /dev/null
@@ -1,208 +0,0 @@
-# -*- coding: utf-8 -*-
-
-import ssl
-from os.path import isfile
-import logging
-
-
-# Default our global support flag
-SLEEKXMPP_SUPPORT_AVAILABLE = False
-
-try:
- # Import sleekxmpp if available
- import sleekxmpp
-
- SLEEKXMPP_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 sleekxmpp installed.
- pass
-
-
-class SleekXmppAdapter(object):
- """
- Wrapper to sleekxmpp
-
- """
-
- # 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 sleekxmpp 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 = SLEEKXMPP_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 SleekXmppAdapter 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 sleekxmpp logger.
- apprise_logger = logging.getLogger('apprise')
- sleek_logger = logging.getLogger('sleekxmpp')
- for handler in apprise_logger.handlers:
- sleek_logger.addHandler(handler)
- sleek_logger.setLevel(apprise_logger.level)
-
- if not self.load():
- raise ValueError("Invalid XMPP Configuration")
-
- def load(self):
-
- # Prepare our object
- self.xmpp = sleekxmpp.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 sleekxmpp.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
diff --git a/apprise/plugins/NotifyXMPP/SliXmppAdapter.py b/apprise/plugins/NotifyXMPP/SliXmppAdapter.py
index aa4bed60..99c23b46 100644
--- a/apprise/plugins/NotifyXMPP/SliXmppAdapter.py
+++ b/apprise/plugins/NotifyXMPP/SliXmppAdapter.py
@@ -39,12 +39,6 @@ class SliXmppAdapter(object):
# 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.
@@ -70,7 +64,8 @@ class SliXmppAdapter(object):
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):
+ body=None, subject=None, targets=None, before_message=None,
+ logger=None):
"""
Initialize our SliXmppAdapter object
"""
@@ -85,6 +80,7 @@ class SliXmppAdapter(object):
self.password = password
self.body = body
+ self.subject = subject
self.targets = targets
self.before_message = before_message
@@ -151,6 +147,16 @@ class SliXmppAdapter(object):
'no local CA certificate file')
return False
+ # If the user specified a port, skip SRV resolving, otherwise it is a
+ # lot easier to let slixmpp handle DNS instead of the user.
+ if self.port is None:
+ override_connection = None
+ else:
+ override_connection = (self.host, self.port)
+
+ # Instruct slixmpp to connect to the XMPP service.
+ self.xmpp.connect(override_connection, use_ssl=self.secure)
+
# We're good
return True
@@ -160,32 +166,9 @@ class SliXmppAdapter(object):
"""
- # 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)
+ # Run the asyncio event loop, and return once disconnected,
+ # for any reason.
+ self.xmpp.process(forever=False)
return self.success
@@ -208,7 +191,8 @@ class SliXmppAdapter(object):
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')
+ self.xmpp.send_message(mto=target, msubject=self.subject,
+ mbody=self.body, mtype='chat')
# Using wait=True ensures that the send queue will be
# emptied before ending the session.
diff --git a/apprise/plugins/NotifyXMPP/__init__.py b/apprise/plugins/NotifyXMPP/__init__.py
index f8346cfb..369708d0 100644
--- a/apprise/plugins/NotifyXMPP/__init__.py
+++ b/apprise/plugins/NotifyXMPP/__init__.py
@@ -30,7 +30,6 @@ from ...URLBase import PrivacyMode
from ...common import NotifyType
from ...utils import parse_list
from ...AppriseLocale import gettext_lazy as _
-from .SleekXmppAdapter import SleekXmppAdapter
from .SliXmppAdapter import SliXmppAdapter
# xep string parser
@@ -57,35 +56,20 @@ class NotifyXMPP(NotifyBase):
# Lower throttle rate for XMPP
request_rate_per_sec = 0.5
- # The default XMPP port
- default_unsecure_port = 5222
-
- # The default XMPP secure port
- default_secure_port = 5223
-
- # XMPP does not support a title
- title_maxlen = 0
-
# This entry is a bit hacky, but it allows us to unit-test this library
- # in an environment that simply doesn't have the sleekxmpp package
+ # 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 = SleekXmppAdapter._enabled or SliXmppAdapter._enabled
- _adapter = SliXmppAdapter if SliXmppAdapter._enabled else SleekXmppAdapter
+ _enabled = SliXmppAdapter._enabled
+ _adapter = SliXmppAdapter if SliXmppAdapter._enabled else None
# Define object templates
templates = (
- '{schema}://{host}',
- '{schema}://{password}@{host}',
- '{schema}://{password}@{host}:{port}',
'{schema}://{user}:{password}@{host}',
'{schema}://{user}:{password}@{host}:{port}',
- '{schema}://{host}/{targets}',
- '{schema}://{password}@{host}/{targets}',
- '{schema}://{password}@{host}:{port}/{targets}',
'{schema}://{user}:{password}@{host}/{targets}',
'{schema}://{user}:{password}@{host}:{port}/{targets}',
)
@@ -106,6 +90,7 @@ class NotifyXMPP(NotifyBase):
'user': {
'name': _('Username'),
'type': 'string',
+ 'required': True,
},
'password': {
'name': _('Password'),
@@ -216,6 +201,7 @@ class NotifyXMPP(NotifyBase):
# By default we send ourselves a message
if targets:
self.targets = parse_list(targets)
+ self.targets[0] = self.targets[0][1:]
else:
self.targets = list()
@@ -228,37 +214,31 @@ class NotifyXMPP(NotifyBase):
if not self._enabled:
self.logger.warning(
'XMPP Notifications are not supported by this system '
- '- install sleekxmpp or slixmpp.')
+ '- install slixmpp.')
return False
# Detect our JID if it isn't otherwise specified
jid = self.jid
password = self.password
if not jid:
- if self.user and self.password:
- # xmpp://user:password@hostname
- jid = '{}@{}'.format(self.user, self.host)
-
- else:
- # xmpp://password@hostname
- jid = self.host
- password = self.password if self.password else self.user
-
- # Compute port number
- if not self.port:
- port = self.default_secure_port \
- if self.secure else self.default_unsecure_port
-
- else:
- port = self.port
+ if not self.user:
+ self.logger.error(
+ 'Invalid JID, must be of the form username@server.')
+ return False
+ jid = '{}@{}'.format(self.user, self.host)
+
+ if not self.password:
+ self.logger.error('You must specify a XMPP password')
+ return False
try:
# Communicate with XMPP.
xmpp_adapter = self._adapter(
- host=self.host, port=port, secure=self.secure,
+ host=self.host, port=self.port, secure=self.secure,
verify_certificate=self.verify_certificate, xep=self.xep,
- jid=jid, password=password, body=body, targets=self.targets,
- before_message=self.throttle, logger=self.logger)
+ jid=jid, password=password, body=body, subject=title,
+ targets=self.targets, before_message=self.throttle,
+ logger=self.logger)
except ValueError:
# We failed
@@ -289,9 +269,6 @@ class NotifyXMPP(NotifyBase):
# and/or space as a delimiters - %20 = space
jids = '%20'.join([NotifyXMPP.quote(x, safe='') for x in self.targets])
- 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
if self.user and self.password:
@@ -310,7 +287,7 @@ class NotifyXMPP(NotifyBase):
schema=default_schema,
# never encode hostname since we're expecting it to be a valid one
hostname=self.host,
- port='' if not self.port or self.port == default_port
+ port='' if not self.port
else ':{}'.format(self.port),
jids=jids,
params=NotifyXMPP.urlencode(params),
diff --git a/apprise/plugins/__init__.py b/apprise/plugins/__init__.py
index 1d35f0b1..8145c2d7 100644
--- a/apprise/plugins/__init__.py
+++ b/apprise/plugins/__init__.py
@@ -33,7 +33,6 @@ from os.path import abspath
# Used for testing
from . import NotifyEmail as NotifyEmailBase
-from .NotifyXMPP import SleekXmppAdapter
from .NotifyXMPP import SliXmppAdapter
# NotifyBase object is passed in as a module not class
@@ -64,9 +63,6 @@ __all__ = [
# Tokenizer
'url_to_dict',
- # sleekxmpp access points (used for NotifyXMPP Testing)
- 'SleekXmppAdapter',
-
# slixmpp access points (used for NotifyXMPP Testing)
'SliXmppAdapter',
]
diff --git a/dev-requirements.txt b/dev-requirements.txt
index ca8df0a2..e725e94d 100644
--- a/dev-requirements.txt
+++ b/dev-requirements.txt
@@ -11,7 +11,6 @@ babel
#
# Provides xmpp:// support
-sleekxmpp
slixmpp; python_version >= '3.7'
# Provides growl:// support
diff --git a/test/test_xmpp_plugin.py b/test/test_xmpp_plugin.py
index 2e687feb..e921e0f6 100644
--- a/test/test_xmpp_plugin.py
+++ b/test/test_xmpp_plugin.py
@@ -30,382 +30,13 @@ import ssl
import mock
import pytest
import apprise
-
-try:
- # Python v3.4+
- from importlib import reload
-except ImportError:
- try:
- # Python v3.0-v3.3
- from imp import reload
- except ImportError:
- # Python v2.7
- pass
-
+from importlib import reload
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
-@pytest.mark.skipif(
- 'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
-def test_sleekxmpp_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 sleekxmpp dependancy. Since
- # sleekxmpp 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 sleekxmpp (simulating the
- # library itself) and writing an __init__.py in it that does nothing
- # but throw an ImportError exception (simulating that the library
- # isn't found).
- suite = tmpdir.mkdir("sleekxmpp")
- suite.join("__init__.py").write('')
- module_name = 'sleekxmpp'
- 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 sleekxmpp
- # path.
-
- # Update our path to point to our new test suite
- sys.path.insert(0, str(suite))
-
- # We need to remove the sleekxmpp 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.SleekXmppAdapter']
-
- # 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 sleekxmpp.
- # XMPP objects can still be instantiated in these cases.
- 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(
- 'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
-def test_sleekxmpp_plugin(tmpdir):
- """
- API: NotifyXMPP Plugin()
- """
-
- # Set success flag
- apprise.plugins.SleekXmppAdapter.success = True
-
- # Enforce Adapter
- apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SleekXmppAdapter
-
- # Create a restore point
- ca_backup = apprise.plugins.SleekXmppAdapter\
- .CA_CERTIFICATE_FILE_LOCATIONS
-
- # Clear CA Certificates
- apprise.plugins.SleekXmppAdapter.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('sleekxmpp.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('sleekxmpp.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('sleekxmpp.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('sleekxmpp.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('sleekxmpp.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_sleekxmpp_test").join('ca_cert')
- ca_cert.write('')
-
- # Update our path
- apprise.plugins.SleekXmppAdapter.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('sleekxmpp.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.SleekXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = \
- ca_backup
-
-
-@pytest.mark.skipif(
- 'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
-def test_sleekxmpp_callbacks():
- """
- API: NotifyXMPP Plugin() Sleekxmpp callback tests
-
- The tests identified here just test the basic callbacks defined for
- sleekxmpp. 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.SleekXmppAdapter.success = False
-
- # Enforce Adapter
- apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SleekXmppAdapter
-
- with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
- client_stream = mock.Mock()
- client_stream.send_message.return_value = True
- mock_stream.return_value = client_stream
-
- adapter = apprise.plugins.SleekXmppAdapter(**kwargs)
- assert isinstance(adapter, apprise.plugins.SleekXmppAdapter)
-
- # 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.SleekXmppAdapter(**kwargs)
- assert isinstance(adapter, apprise.plugins.SleekXmppAdapter)
-
- # success flag should be back to a False state
- assert adapter.success is False
-
- with mock.patch('sleekxmpp.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.SleekXmppAdapter(**kwargs)
-
-
@pytest.mark.skipif(
'slixmpp' not in sys.modules, reason="requires slixmpp")
def test_slixmpp_plugin_import_error(tmpdir):
@@ -472,7 +103,6 @@ def test_slixmpp_plugin_import_error(tmpdir):
# 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'])