diff --git a/apprise/plugins/NotifyXMPP/SleekXmppAdapter.py b/apprise/plugins/NotifyXMPP/SleekXmppAdapter.py new file mode 100644 index 00000000..86650995 --- /dev/null +++ b/apprise/plugins/NotifyXMPP/SleekXmppAdapter.py @@ -0,0 +1,185 @@ +# -*- coding: utf-8 -*- + +import ssl +from os.path import isfile +import sleekxmpp +import logging + + +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", + ] + + 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.py b/apprise/plugins/NotifyXMPP/__init__.py similarity index 68% rename from apprise/plugins/NotifyXMPP.py rename to apprise/plugins/NotifyXMPP/__init__.py index 7788c7c7..41539796 100644 --- a/apprise/plugins/NotifyXMPP.py +++ b/apprise/plugins/NotifyXMPP/__init__.py @@ -24,15 +24,12 @@ # THE SOFTWARE. import re -import ssl -import logging -from os.path import isfile -from .NotifyBase import NotifyBase -from ..URLBase import PrivacyMode -from ..common import NotifyType -from ..utils import parse_list -from ..AppriseLocale import gettext_lazy as _ +from ..NotifyBase import NotifyBase +from ...URLBase import PrivacyMode +from ...common import NotifyType +from ...utils import parse_list +from ...AppriseLocale import gettext_lazy as _ # xep string parser XEP_PARSE_RE = re.compile('^[^1-9]*(?P[1-9][0-9]{0,3})$') @@ -40,23 +37,10 @@ XEP_PARSE_RE = re.compile('^[^1-9]*(?P[1-9][0-9]{0,3})$') # Default our global support flag NOTIFY_XMPP_SUPPORT_ENABLED = False -# 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", -] - try: # Import sleekxmpp if available - import sleekxmpp + + from .SleekXmppAdapter import SleekXmppAdapter NOTIFY_XMPP_SUPPORT_ENABLED = True @@ -235,10 +219,11 @@ class NotifyXMPP(NotifyBase): result = XEP_PARSE_RE.match(xep) if result is not None: self.xep.append(int(result.group('xep'))) + self.logger.debug('Loaded XMPP {}'.format(xep)) else: self.logger.warning( - "Could not load XMPP xep {}".format(xep)) + "Could not load XMPP {}".format(xep)) # By default we send ourselves a message if targets: @@ -279,18 +264,17 @@ class NotifyXMPP(NotifyBase): else: port = self.port - # Handler function to be called before each message. - # Always call throttle before any remote server i/o is made. - def on_before_message(): - self.throttle() + try: + # Communicate with XMPP. + xmpp_adapter = SleekXmppAdapter( + host=self.host, port=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) - # Communicate with XMPP. - xmpp_adapter = SleekXmppAdapter( - host=self.host, port=port, secure=self.secure, - verify_certificate=self.verify_certificate, - xep=self.xep, jid=jid, password=password, - body=body, targets=self.targets, before_message=on_before_message, - logger=self.logger) + except ValueError: + # We failed + return False # Initialize XMPP machinery and begin processing the XML stream. outcome = xmpp_adapter.process() @@ -379,132 +363,3 @@ class NotifyXMPP(NotifyBase): NotifyXMPP.parse_list(results['qsd']['to']) return results - - -class SleekXmppAdapter(object): - """ - Wrapper to SleekXmpp - """ - - def __init__(self, - host=None, port=None, secure=None, verify_certificate=None, - xep=None, jid=None, password=None, body=None, targets=None, - before_message=None, logger=None): - - 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__) - - # Reference to XMPP client. - self.xmpp = None - - # Whether everything succeeded. - self.success = False - - self.configure_logging() - self.setup() - - def configure_logging(self): - - # 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) - - def setup(self): - - # Prepare our object - self.xmpp = sleekxmpp.ClientXMPP(self.jid, self.password) - - self.xmpp.add_event_handler("session_start", self.session_start) - self.xmpp.add_event_handler("failed_auth", self.failed_auth) - - for xep in self.xep: - # Load xep entries - self.xmpp.register_plugin('xep_{0:04d}'.format(xep)) - - 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 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 CA certificate found') - - def process(self): - - # 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): - return False - - # Process XMPP communication. - self.xmpp.process(block=True) - - return self.success - - def session_start(self, event): - """ - 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. - # Here, it will indirectly invoke the throttling feature, - # which adds a delay before any remote server i/o is made. - if callable(self.before_message): - 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) - - self.success = True - - def failed_auth(self, event): - self.logger.error('Authentication with XMPP server failed') diff --git a/apprise/plugins/__init__.py b/apprise/plugins/__init__.py index 21ff47fc..fd41cb7f 100644 --- a/apprise/plugins/__init__.py +++ b/apprise/plugins/__init__.py @@ -34,6 +34,7 @@ from os.path import abspath # Used for testing from . import NotifyEmail as NotifyEmailBase from .NotifyGrowl import gntp +from .NotifyXMPP import SleekXmppAdapter # NotifyBase object is passed in as a module not class from . import NotifyBase @@ -63,6 +64,9 @@ __all__ = [ # gntp (used for NotifyGrowl Testing) 'gntp', + + # sleekxmpp access points (used for NotifyXMPP Testing) + 'SleekXmppAdapter', ] # we mirror our base purely for the ability to reset everything; this diff --git a/test/test_xmpp_plugin.py b/test/test_xmpp_plugin.py index 56ec154d..90b181c0 100644 --- a/test/test_xmpp_plugin.py +++ b/test/test_xmpp_plugin.py @@ -1,6 +1,6 @@ # -*- coding: utf-8 -*- # -# Copyright (C) 2019 Chris Caron +# Copyright (C) 2020 Chris Caron # All rights reserved. # # This code is licensed under the MIT License. @@ -24,9 +24,11 @@ # THE SOFTWARE. import six +import os import sys import ssl import mock +import pytest import apprise @@ -46,13 +48,43 @@ import logging logging.disable(logging.CRITICAL) -def test_xmpp_plugin(tmpdir): - """ - API: NotifyXMPP Plugin() +def test_xmpp_plugin_import_error(tmpdir): """ + API: NotifyXMPP Plugin() Import Error - # Mock the sleekxmpp module completely. - sys.modules['sleekxmpp'] = mock.MagicMock() + """ + # 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 hte 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 @@ -66,17 +98,40 @@ def test_xmpp_plugin(tmpdir): reload(sys.modules['apprise.Apprise']) reload(sys.modules['apprise']) - # Mock the XMPP adapter to override "self.success". - # This will signal a successful message delivery. - from apprise.plugins.NotifyXMPP import SleekXmppAdapter - class MockedSleekXmppAdapter(SleekXmppAdapter): + # This tests that Apprise still works without sleekxmpp. + # XMPP objects can't be istantiated though. + obj = apprise.Apprise.instantiate('xmpp://user:pass@localhost') + assert obj is not None - def __init__(self, *args, **kwargs): - super(MockedSleekXmppAdapter, self).__init__(*args, **kwargs) - self.success = True + # Tidy-up / restore things to how they were + # Remove our garbage library + os.unlink(str(suite.join("{}.py".format(module_name)))) - NotifyXMPP = sys.modules['apprise.plugins.NotifyXMPP'] - NotifyXMPP.SleekXmppAdapter = MockedSleekXmppAdapter + # Remove our custom entry into the path + sys.path.remove(str(suite)) + + # Reload the libraries we care about + reload(sys.modules['apprise.plugins.NotifyXMPP']) + reload(sys.modules['apprise.plugins.NotifyXMPP.SleekXmppAdapter']) + reload(sys.modules['apprise.plugins']) + reload(sys.modules['apprise.Apprise']) + reload(sys.modules['apprise']) + + +def test_xmpp_plugin(tmpdir): + """ + API: NotifyXMPP Plugin() + """ + + # Set success flag + apprise.plugins.SleekXmppAdapter.success = True + + # 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 @@ -87,18 +142,9 @@ def test_xmpp_plugin(tmpdir): # Not possible because no password or host was specified assert obj is None - try: - obj = apprise.Apprise.instantiate( + with pytest.raises(TypeError): + apprise.Apprise.instantiate( 'xmpp://hostname', suppress_exceptions=False) - # We should not reach here; we should have thrown an exception - assert False - - except TypeError: - # we're good - assert True - - # Not possible because no password was specified - assert obj is None # SSL Flags if hasattr(ssl, "PROTOCOL_TLS"): @@ -107,13 +153,24 @@ def test_xmpp_plugin(tmpdir): del ssl.PROTOCOL_TLS # Test our URL - url = 'xmpps://user:pass@localhost' + 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 - assert obj.notify( - title='title', body='body', - notify_type=apprise.NotifyType.INFO) 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) @@ -124,32 +181,42 @@ def test_xmpp_plugin(tmpdir): # 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 - assert obj.notify( - title='title', body='body', - notify_type=apprise.NotifyType.INFO) 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': 'xmpps://user:pass@localhost', - 'p': 'xmpps://user:****@localhost', + 'u': 'xmpp://user:pass@localhost', + 'p': 'xmpp://user:****@localhost', }, { - 'u': 'xmpps://user:pass@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://user:pass@localhost?xep=ignored', - 'p': 'xmpps://user:****@localhost', - }, { - 'u': 'xmpps://pass@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', + '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', @@ -158,7 +225,7 @@ def test_xmpp_plugin(tmpdir): 'u': 'xmpps://user:pass@localhost?verify=False', 'p': 'xmpps://user:****@localhost', }, { - 'u': 'xmpp://user:pass@localhost?to=user@test.com', + 'u': 'xmpp://user:pass@localhost?to=user@test.com&verify=no', 'p': 'xmpp://user:****@localhost', } ) @@ -185,21 +252,43 @@ def test_xmpp_plugin(tmpdir): assert obj.url(privacy=True).startswith(privacy_url) - # test notifications - assert obj.notify( - title='title', body='body', - notify_type=apprise.NotifyType.INFO) 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 - # test notification without a title - assert obj.notify( - title='', body='body', notify_type=apprise.NotifyType.INFO) is True + # 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 - # Verify that we can't send content now - assert obj.notify( - title='', body='body', notify_type=apprise.NotifyType.INFO) is 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 @@ -207,14 +296,98 @@ def test_xmpp_plugin(tmpdir): # create an empty file for now ca_cert = tmpdir.mkdir("apprise_xmpp_test").join('ca_cert') ca_cert.write('') + # Update our path - sys.modules['apprise.plugins.NotifyXMPP']\ - .CA_CERTIFICATE_FILE_LOCATIONS = [str(ca_cert), ] + apprise.plugins.SleekXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = \ + [str(ca_cert), ] obj = apprise.Apprise.instantiate( - 'xmpps://pass@localhost/user@test.com', + 'xmpps://pass@localhost/user@test.com?verify=yes', suppress_exceptions=False) + assert isinstance(obj, apprise.plugins.NotifyXMPP) is 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 + 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 + + +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 + + 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)