From 1dfde961e8f1b7e078a6e60377c16fd5381585cf Mon Sep 17 00:00:00 2001 From: linkmauve Date: Wed, 24 Nov 2021 01:15:12 +0100 Subject: [PATCH] Fixed XMPP Support & SleekXMPP dropped (#485) --- README.md | 2 +- .../plugins/NotifyXMPP/SleekXmppAdapter.py | 208 ---------- apprise/plugins/NotifyXMPP/SliXmppAdapter.py | 52 +-- apprise/plugins/NotifyXMPP/__init__.py | 63 +-- apprise/plugins/__init__.py | 4 - dev-requirements.txt | 1 - test/test_xmpp_plugin.py | 372 +----------------- 7 files changed, 40 insertions(+), 662 deletions(-) delete mode 100644 apprise/plugins/NotifyXMPP/SleekXmppAdapter.py 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'])