diff --git a/README.md b/README.md index 91101d97..3a4cd9c6 100644 --- a/README.md +++ b/README.md @@ -96,7 +96,6 @@ 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://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/all-plugin-requirements.txt b/all-plugin-requirements.txt index b6edc6d2..cb17ed16 100644 --- a/all-plugin-requirements.txt +++ b/all-plugin-requirements.txt @@ -5,9 +5,6 @@ # Provides fcm:// and spush:// cryptography -# Provides xmpp:// support -slixmpp; python_version >= '3.7' - # Provides growl:// support gntp diff --git a/apprise/plugins/NotifyXMPP/SliXmppAdapter.py b/apprise/plugins/NotifyXMPP/SliXmppAdapter.py deleted file mode 100644 index 6c2fc1c7..00000000 --- a/apprise/plugins/NotifyXMPP/SliXmppAdapter.py +++ /dev/null @@ -1,226 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Copyright (C) 2022 Chris Caron -# All rights reserved. -# -# This code is licensed under the MIT License. -# -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files(the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions : -# -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. -# -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -# THE SOFTWARE. - -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' - - # 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, subject=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.subject = subject - 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 - - # If the user specified a port, skip SRV resolving, otherwise it is a - # lot easier to let slixmpp handle DNS instead of the user. - self.override_connection = \ - None if not self.port else (self.host, self.port) - - # We're good - return True - - def process(self): - """ - Thread that handles the server/client i/o - - """ - - # Instruct slixmpp to connect to the XMPP service. - if not self.xmpp.connect( - self.override_connection, use_ssl=self.secure): - return False - - # Run the asyncio event loop, and return once disconnected, - # for any reason. - self.xmpp.process(forever=False) - - 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, msubject=self.subject, - 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/__init__.py b/apprise/plugins/NotifyXMPP/__init__.py deleted file mode 100644 index d5fb9a2c..00000000 --- a/apprise/plugins/NotifyXMPP/__init__.py +++ /dev/null @@ -1,311 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Copyright (C) 2019 Chris Caron -# All rights reserved. -# -# This code is licensed under the MIT License. -# -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files(the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions : -# -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. -# -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -# THE SOFTWARE. - -import re - -from ..NotifyBase import NotifyBase -from ...URLBase import PrivacyMode -from ...common import NotifyType -from ...utils import parse_list -from ...AppriseLocale import gettext_lazy as _ -from .SliXmppAdapter import SliXmppAdapter - -# xep string parser -XEP_PARSE_RE = re.compile('^[^1-9]*(?P[1-9][0-9]{0,3})$') - - -class NotifyXMPP(NotifyBase): - """ - A wrapper for XMPP Notifications - """ - # Set our global enabled flag - enabled = SliXmppAdapter._enabled - - requirements = { - # Define our required packaging in order to work - 'packages_required': [ - "slixmpp; python_version >= '3.7'", - ] - } - - # The default descriptive name associated with the Notification - service_name = 'XMPP' - - # The services URL - service_url = 'https://xmpp.org/' - - # The default protocol - protocol = 'xmpp' - - # The default secure protocol - secure_protocol = 'xmpps' - - # A URL that takes you to the setup/help of the specific protocol - setup_url = 'https://github.com/caronc/apprise/wiki/Notify_xmpp' - - # Lower throttle rate for XMPP - request_rate_per_sec = 0.5 - - # Our XMPP Adapter we use to communicate through - _adapter = SliXmppAdapter if SliXmppAdapter._enabled else None - - # Define object templates - templates = ( - '{schema}://{user}:{password}@{host}', - '{schema}://{user}:{password}@{host}:{port}', - '{schema}://{user}:{password}@{host}/{targets}', - '{schema}://{user}:{password}@{host}:{port}/{targets}', - ) - - # Define our tokens - template_tokens = dict(NotifyBase.template_tokens, **{ - 'host': { - 'name': _('Hostname'), - 'type': 'string', - 'required': True, - }, - 'port': { - 'name': _('Port'), - 'type': 'int', - 'min': 1, - 'max': 65535, - }, - 'user': { - 'name': _('Username'), - 'type': 'string', - 'required': True, - }, - 'password': { - 'name': _('Password'), - 'type': 'string', - 'private': True, - 'required': True, - }, - 'target_jid': { - 'name': _('Target JID'), - 'type': 'string', - 'map_to': 'targets', - }, - 'targets': { - 'name': _('Targets'), - 'type': 'list:string', - }, - }) - - # Define our template arguments - template_args = dict(NotifyBase.template_args, **{ - 'to': { - 'alias_of': 'targets', - }, - 'xep': { - 'name': _('XEP'), - 'type': 'list:string', - 'prefix': 'xep-', - 'regex': (r'^[1-9][0-9]{0,3}$', 'i'), - }, - 'jid': { - 'name': _('Source JID'), - 'type': 'string', - }, - }) - - def __init__(self, targets=None, jid=None, xep=None, **kwargs): - """ - Initialize XMPP Object - """ - super(NotifyXMPP, self).__init__(**kwargs) - - # JID Details: - # - JID's normally have an @ symbol in them, but it is not required - # - Each allowable portion of a JID MUST NOT be more than 1023 bytes - # in length. - # - JID's can identify resource paths at the end separated by slashes - # hence the following is valid: user@example.com/resource/path - - # Since JID's can clash with URLs offered by aprise (specifically the - # resource paths we need to allow users an alternative character to - # represent the slashes. The grammer is defined here: - # https://xmpp.org/extensions/xep-0029.html as follows: - # - # ::= ["@"]["/"] - # ::= []* - # ::= ["."]* - # ::= []* - # ::= |[[||"-"]*|] - # ::= [a-z] | [A-Z] - # ::= [0-9] - # ::= #x21 | [#x23-#x25] | [#x28-#x2E] | - # [#x30-#x39] | #x3B | #x3D | #x3F | - # [#x41-#x7E] | [#x80-#xD7FF] | - # [#xE000-#xFFFD] | [#x10000-#x10FFFF] - # ::= [#x20-#xD7FF] | [#xE000-#xFFFD] | - # [#x10000-#x10FFFF] - - # The best way to do this is to choose characters that aren't allowed - # in this case we will use comma and/or space. - - # Assemble our jid using the information available to us: - self.jid = jid - - if not (self.user or self.password): - # you must provide a jid/pass for this to work; if no password - # is specified then the user field acts as the password instead - # so we know that if there is no user specified, our url was - # really busted up. - msg = 'You must specify a XMPP password' - self.logger.warning(msg) - raise TypeError(msg) - - # See https://xmpp.org/extensions/ for details on xep values - if xep is None: - # Default xep setting - self.xep = [ - # xep_0030: Service Discovery - 30, - # xep_0199: XMPP Ping - 199, - ] - - else: - # Prepare the list - _xep = parse_list(xep) - self.xep = [] - - for xep in _xep: - 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 {}".format(xep)) - - # 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() - - def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): - """ - Perform XMPP Notification - """ - - # Detect our JID if it isn't otherwise specified - jid = self.jid - password = self.password - if not jid: - jid = '{}@{}'.format(self.user, self.host) - - try: - # Communicate with XMPP. - xmpp_adapter = self._adapter( - host=self.host, port=self.port, secure=self.secure, - verify_certificate=self.verify_certificate, xep=self.xep, - jid=jid, password=password, body=body, subject=title, - targets=self.targets, before_message=self.throttle, - logger=self.logger) - - except ValueError: - # We failed - return False - - # Initialize XMPP machinery and begin processing the XML stream. - outcome = xmpp_adapter.process() - - return outcome - - def url(self, privacy=False, *args, **kwargs): - """ - Returns the URL built dynamically based on specified arguments. - """ - - # Our URL parameters - params = self.url_parameters(privacy=privacy, *args, **kwargs) - - if self.jid: - params['jid'] = self.jid - - if self.xep: - # xep are integers, so we need to just iterate over a list and - # switch them to a string - params['xep'] = ','.join([str(xep) for xep in self.xep]) - - # Target JID(s) can clash with our existing paths, so we just use comma - # and/or space as a delimiters - %20 = space - jids = '%20'.join([NotifyXMPP.quote(x, safe='') for x in self.targets]) - - default_schema = self.secure_protocol if self.secure else self.protocol - - auth = '{user}:{password}'.format( - user=NotifyXMPP.quote(self.user, safe=''), - password=self.pprint( - self.password, privacy, mode=PrivacyMode.Secret, safe='')) - - return '{schema}://{auth}@{hostname}{port}/{jids}?{params}'.format( - auth=auth, - schema=default_schema, - # never encode hostname since we're expecting it to be a valid one - hostname=self.host, - port='' if not self.port - else ':{}'.format(self.port), - jids=jids, - params=NotifyXMPP.urlencode(params), - ) - - @staticmethod - def parse_url(url): - """ - Parses the URL and returns enough arguments that can allow - us to re-instantiate this object. - - """ - results = NotifyBase.parse_url(url) - if not results: - # We're done early as we couldn't load the results - return results - - # Get our targets; we ignore path slashes since they identify - # our resources - results['targets'] = NotifyXMPP.parse_list(results['fullpath']) - - # Over-ride the xep plugins - if 'xep' in results['qsd'] and len(results['qsd']['xep']): - results['xep'] = \ - NotifyXMPP.parse_list(results['qsd']['xep']) - - # Over-ride the default (and detected) jid - if 'jid' in results['qsd'] and len(results['qsd']['jid']): - results['jid'] = NotifyXMPP.unquote(results['qsd']['jid']) - - # Over-ride the default (and detected) jid - if 'to' in results['qsd'] and len(results['qsd']['to']): - results['targets'] += \ - NotifyXMPP.parse_list(results['qsd']['to']) - - return results diff --git a/apprise/plugins/__init__.py b/apprise/plugins/__init__.py index 08a27d94..177c0e24 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 SliXmppAdapter # NotifyBase object is passed in as a module not class from . import NotifyBase @@ -62,9 +61,6 @@ __all__ = [ # Tokenizer 'url_to_dict', - - # slixmpp access points (used for NotifyXMPP Testing) - 'SliXmppAdapter', ] # we mirror our base purely for the ability to reset everything; this diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 48841a89..763c7119 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -56,8 +56,7 @@ Notifico, ntfy, Office365, OneSignal, Opsgenie, PagerDuty, ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, Pushover, PushSafer, Reddit, Rocket.Chat, SendGrid, ServerChan, Signal, SimplePush, Sinch, Slack, SMTP2Go, Spontit, SparkPost, Super Toasty, Streamlabs, Stride, Syslog, -Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP, Vonage, Webex -Teams} +Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, Vonage, Webex Teams} Name: python-%{pypi_name} Version: 0.9.9 @@ -173,9 +172,6 @@ BuildRequires: python%{python3_pkgversion}-pytest-runner %patch0 -p1 # rhel7 doesn't like the new asyncio syntax rm -f apprise/py3compat/asyncio.py -# rhel7 doesn't support slixmpp and formatting in place -# within the dev-requirements.txt file to reference it -sed -i -e '/^slixmpp.*/d' dev-requirements.txt %endif %build diff --git a/test/test_plugin_xmpp.py b/test/test_plugin_xmpp.py deleted file mode 100644 index 22439892..00000000 --- a/test/test_plugin_xmpp.py +++ /dev/null @@ -1,334 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Copyright (C) 2020 Chris Caron -# All rights reserved. -# -# This code is licensed under the MIT License. -# -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files(the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions : -# -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. -# -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -# THE SOFTWARE. - -import six -import sys -import ssl -import mock -import pytest -import apprise - -# Disable logging for a cleaner testing output -import logging -logging.disable(logging.CRITICAL) - - -@pytest.mark.skipif( - 'slixmpp' in sys.modules, reason="Requires that slixmpp NOT be installed") -def test_plugin_xmpp_slixmpp_import_error(): - """ - NotifyXMPP() 'slixmpp' Import Error - - """ - # Without the slixmpp library, we can not load our object - obj = apprise.Apprise.instantiate('xmpp://user:pass@localhost') - assert obj is None - - -@pytest.mark.skipif( - 'slixmpp' not in sys.modules, reason="Requires slixmpp") -def test_plugin_xmpp_general(tmpdir): - """ - NotifyXMPP() General Checks - """ - - # 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://user:pass@localhost/?verify=false&to=' - 'user@test.com, user2@test.com/resource', - 'p': 'xmpps://user:****@localhost', - }, { - 'u': 'xmpps://user:pass@localhost:5226?' - 'jid=user@test.com&verify=no', - 'p': 'xmpps://user:****@localhost:5226', - }, { - 'u': 'xmpps://user:pass@localhost?jid=user@test.com&verify=False', - 'p': 'xmpps://user:****@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 - - print(obj.url()) - # 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://user: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_plugin_xmpp_slixmpp_callbacks(): - """ - NotifyXMPP() 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)