MQTT Support Added (#443)

README.md

@@ -55,6 +55,7 @@ The table below identifies the services this tool supports and some example serv
 | [Matrix](https://github.com/caronc/apprise/wiki/Notify_matrix) | matrix:// or matrixs://  | (TCP) 80 or 443 | matrix://hostname<br />matrix://user@hostname<br />matrixs://user:pass@hostname:port/#room_alias<br />matrixs://user:pass@hostname:port/!room_id<br />matrixs://user:pass@hostname:port/#room_alias/!room_id/#room2<br />matrixs://token@hostname:port/?webhook=matrix<br />matrix://user:token@hostname/?webhook=slack&format=markdown
 | [Mattermost](https://github.com/caronc/apprise/wiki/Notify_mattermost) | mmost:// or mmosts:// | (TCP) 8065 | mmost://hostname/authkey<br />mmost://hostname:80/authkey<br />mmost://user@hostname:80/authkey<br />mmost://hostname/authkey?channel=channel<br />mmosts://hostname/authkey<br />mmosts://user@hostname/authkey<br />
 | [Microsoft Teams](https://github.com/caronc/apprise/wiki/Notify_msteams) | msteams://  | (TCP) 443   | msteams://TokenA/TokenB/TokenC/
+| [MQTT](https://github.com/caronc/apprise/wiki/Notify_mqtt) | mqtt://  or mqtts:// | (TCP) 1883 or 8883   | mqtt://hostname/topic<br />mqtt://user@hostname/topic<br />mqtts://user:pass@hostname:9883/topic
 | [Nextcloud](https://github.com/caronc/apprise/wiki/Notify_nextcloud) | ncloud:// or nclouds:// | (TCP) 80 or 443 | ncloud://adminuser:pass@host/User<br/>nclouds://adminuser:pass@host/User1/User2/UserN
 | [Notica](https://github.com/caronc/apprise/wiki/Notify_notica) | notica://  | (TCP) 443   | notica://Token/
 | [Notifico](https://github.com/caronc/apprise/wiki/Notify_notifico) | notifico://  | (TCP) 443   | notifico://ProjectID/MessageHook/

apprise/plugins/NotifyMQTT.py

@@ -0,0 +1,547 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2021 Chris Caron <lead2gold@gmail.com>
+# 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.
+
+# PAHO MQTT Documentation:
+#  https://www.eclipse.org/paho/index.php?page=clients/python/docs/index.php
+#
+# Looking at the PAHO MQTT Source can help shed light on what's going on too
+# as their inline documentation is pretty good!
+#   https://github.com/eclipse/paho.mqtt.python\
+#           /blob/master/src/paho/mqtt/client.py
+import ssl
+import re
+import six
+from time import sleep
+from datetime import datetime
+from os.path import isfile
+from .NotifyBase import NotifyBase
+from ..URLBase import PrivacyMode
+from ..common import NotifyType
+from ..utils import parse_list
+from ..utils import parse_bool
+from ..AppriseLocale import gettext_lazy as _
+
+# Default our global support flag
+NOTIFY_MQTT_SUPPORT_ENABLED = False
+
+if six.PY2:
+    # handle Python v2.7 suport
+    class ConnectionError(Exception):
+        pass
+
+try:
+    # 3rd party modules
+    import paho.mqtt.client as mqtt
+
+    # We're good to go!
+    NOTIFY_MQTT_SUPPORT_ENABLED = True
+
+    MQTT_PROTOCOL_MAP = {
+        # v3.1.1
+        "311": mqtt.MQTTv311,
+        # v3.1
+        "31": mqtt.MQTTv31,
+        # v5.0
+        "5": mqtt.MQTTv5,
+        # v5.0 (alias)
+        "50": mqtt.MQTTv5,
+    }
+
+except ImportError:
+    # No problem; we just simply can't support this plugin because we're
+    # either using Linux, or simply do not have pywin32 installed.
+    MQTT_PROTOCOL_MAP = {}
+
+# A lookup map for relaying version to user
+HUMAN_MQTT_PROTOCOL_MAP = {
+    "v3.1.1": "311",
+    "v3.1": "31",
+    "v5.0": "5",
+}
+
+
+class NotifyMQTT(NotifyBase):
+    """
+    A wrapper for MQTT Notifications
+    """
+
+    # The default descriptive name associated with the Notification
+    service_name = 'MQTT Notification'
+
+    # The default protocol
+    protocol = 'mqtt'
+
+    # Secure protocol
+    secure_protocol = 'mqtts'
+
+    # A URL that takes you to the setup/help of the specific protocol
+    setup_url = 'https://github.com/caronc/apprise/wiki/Notify_mqtt'
+
+    # MQTT does not have a title
+    title_maxlen = 0
+
+    # The maximum length a body can be set to
+    body_maxlen = 268435455
+
+    # Use a throttle; but it doesn't need to be so strict since most
+    # MQTT server hostings can handle the small bursts of packets and are
+    # locally hosted anyway
+    request_rate_per_sec = 0.5
+
+    # This entry is a bit hacky, but it allows us to unit-test this library
+    # in an environment that simply doesn't have the mqtt packages
+    # available to us.  It also allows us to handle situations where the
+    # packages actually are present but we need to test that they aren't.
+    # If anyone is seeing this had knows a better way of testing this
+    # outside of what is defined in test/test_mqtt_plugin.py, please
+    # let me know! :)
+    _enabled = NOTIFY_MQTT_SUPPORT_ENABLED
+
+    # Port Defaults (unless otherwise specified)
+    mqtt_insecure_port = 1883
+
+    # The default secure port to use (if mqtts://)
+    mqtt_secure_port = 8883
+
+    # The default mqtt keepalive value
+    mqtt_keepalive = 30
+
+    # The default mqtt transport
+    mqtt_transport = "tcp"
+
+    # The number of seconds to wait for a publish to occur at before
+    # checking to see if it's been sent yet.
+    mqtt_block_time_sec = 0.2
+
+    # Set the maximum number of messages with QoS>0 that can be part way
+    # through their network flow at once.
+    mqtt_inflight_messages = 200
+
+    # 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",
+    ]
+
+    # Define object templates
+    templates = (
+        '{schema}://{user}@{host}/{topic}',
+        '{schema}://{user}@{host}:{port}/{topic}',
+        '{schema}://{user}:{password}@{host}/{topic}',
+        '{schema}://{user}:{password}@{host}:{port}/{topic}',
+    )
+
+    template_tokens = dict(NotifyBase.template_tokens, **{
+        'host': {
+            'name': _('Hostname'),
+            'type': 'string',
+            'required': True,
+        },
+        'port': {
+            'name': _('Port'),
+            'type': 'int',
+            'min': 1,
+            'max': 65535,
+        },
+        'user': {
+            'name': _('User Name'),
+            'type': 'string',
+            'required': True,
+        },
+        'password': {
+            'name': _('Password'),
+            'type': 'string',
+            'private': True,
+            'required': True,
+        },
+        'topic': {
+            'name': _('Target Queue'),
+            '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',
+        },
+        'qos': {
+            'name': _('QOS'),
+            'type': 'int',
+            'default': 0,
+            'min': 0,
+            'max': 2,
+        },
+        'version': {
+            'name': _('Version'),
+            'type': 'choice:string',
+            'values': HUMAN_MQTT_PROTOCOL_MAP,
+            'default': "v3.1.1",
+        },
+        'client_id': {
+            'name': _('Client ID'),
+            'type': 'string',
+        },
+        'session': {
+            'name': _('Use Session'),
+            'type': 'bool',
+            'default': False,
+        },
+    })
+
+    def __init__(self, targets=None, version=None, qos=None,
+                 client_id=None, session=None, **kwargs):
+        """
+        Initialize MQTT Object
+        """
+
+        super(NotifyMQTT, self).__init__(**kwargs)
+
+        # Initialize topics
+        self.topics = parse_list(targets)
+
+        if version is None:
+            self.version = self.template_args['version']['default']
+        else:
+            self.version = version
+
+        # Save our client id if specified
+        self.client_id = client_id
+
+        # Maintain our session (associated with our user id if set)
+        self.session = self.template_args['session']['default'] \
+            if session is None or not self.client_id \
+            else parse_bool(session)
+
+        # Set up our Quality of Service (QoS)
+        try:
+            self.qos = self.template_args['qos']['default'] \
+                if qos is None else int(qos)
+
+            if self.qos < self.template_args['qos']['min'] \
+                    or self.qos > self.template_args['qos']['max']:
+                # Let error get handle on exceptio higher up
+                raise ValueError("")
+
+        except (ValueError, TypeError):
+            msg = 'An invalid MQTT QOS ({}) was specified.'.format(qos)
+            self.logger.warning(msg)
+            raise TypeError(msg)
+
+        if not self.port:
+            # Assign port (if not otherwise set)
+            self.port = self.mqtt_secure_port \
+                if self.secure else self.mqtt_insecure_port
+
+        self.ca_certs = None
+        if self.secure:
+            # verify SSL key or abort
+            self.ca_certs = next(
+                (cert for cert in self.CA_CERTIFICATE_FILE_LOCATIONS
+                 if isfile(cert)), None)
+
+        if not self._enabled:
+            # Nothing more we can do
+            return
+
+        # Set up our MQTT Publisher
+        try:
+            # Get our protocol
+            self.mqtt_protocol = \
+                MQTT_PROTOCOL_MAP[re.sub(r'[^0-9]+', '', self.version)]
+
+        except (KeyError):
+            msg = 'An invalid MQTT Protocol version ' \
+                '({}) was specified.'.format(version)
+            self.logger.warning(msg)
+            raise TypeError(msg)
+
+        # Our MQTT Client Object
+        self.client = mqtt.Client(
+            client_id=self.client_id,
+            clean_session=not self.session, userdata=None,
+            protocol=self.mqtt_protocol, transport=self.mqtt_transport,
+        )
+
+        # Our maximum number of in-flight messages
+        self.client.max_inflight_messages_set(self.mqtt_inflight_messages)
+
+        # Toggled to False once our connection has been established at least
+        # once
+        self.__initial_connect = True
+
+    def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
+        """
+        Perform MQTT Notification
+        """
+
+        if not self._enabled:
+            self.logger.warning(
+                "MQTT Notifications are not supported by this system; "
+                "`pip install paho-mqtt`.")
+            return False
+
+        if len(self.topics) == 0:
+            # There were no services to notify
+            self.logger.warning('There were no MQTT topics to notify.')
+            return False
+
+        # For logging:
+        url = '{host}:{port}'.format(host=self.host, port=self.port)
+
+        try:
+            if self.__initial_connect:
+                # Our initial connection
+                if self.user:
+                    self.client.username_pw_set(
+                        self.user, password=self.password)
+
+                if self.secure:
+                    if self.ca_certs is None:
+                        self.logger.warning(
+                            'MQTT Secure comunication can not be verified; '
+                            'no local CA certificate file')
+                        return False
+
+                    self.client.tls_set(
+                        ca_certs=self.ca_certs, certfile=None, keyfile=None,
+                        cert_reqs=ssl.CERT_REQUIRED,
+                        tls_version=ssl.PROTOCOL_TLS,
+                        ciphers=None)
+
+                    # Set our TLS Verify Flag
+                    self.client.tls_insecure_set(self.verify_certificate)
+
+                # Establish our connection
+                if self.client.connect(
+                        self.host, port=self.port,
+                        keepalive=self.mqtt_keepalive) \
+                        != mqtt.MQTT_ERR_SUCCESS:
+                    self.logger.warning(
+                        'An MQTT connection could not be established for {}'.
+                        format(url))
+                    return False
+
+                # Start our client loop
+                self.client.loop_start()
+
+                # Throttle our start otherwise the starting handshaking doesnt
+                # work. I'm not sure if this is a bug or not, but with qos=0,
+                # and without this sleep(), the messages randomly fails to be
+                # delivered.
+                sleep(0.01)
+
+                # Toggle our flag since we never need to enter this area again
+                self.__initial_connect = False
+
+            # Create a copy of the subreddits list
+            topics = list(self.topics)
+
+            has_error = False
+            while len(topics) > 0 and not has_error:
+                # Retrieve our subreddit
+                topic = topics.pop()
+
+                # For logging:
+                url = '{host}:{port}/{topic}'.format(
+                    host=self.host,
+                    port=self.port,
+                    topic=topic)
+
+                # Always call throttle before any remote server i/o is made
+                self.throttle()
+
+                # handle a re-connection
+                if not self.client.is_connected() and \
+                        self.client.reconnect() != mqtt.MQTT_ERR_SUCCESS:
+                    self.logger.warning(
+                        'An MQTT connection could not be sustained for {}'.
+                        format(url))
+                    has_error = True
+                    break
+
+                # Some Debug Logging
+                self.logger.debug('MQTT POST URL: {} (cert_verify={})'.format(
+                    url, self.verify_certificate))
+                self.logger.debug('MQTT Payload: %s' % str(body))
+
+                result = self.client.publish(
+                    topic, payload=body, qos=self.qos, retain=False)
+
+                if result.rc != mqtt.MQTT_ERR_SUCCESS:
+                    # Toggle our status
+                    self.logger.warning(
+                        'An error (rc={}) occured when sending MQTT to {}'.
+                        format(result.rc, url))
+                    has_error = True
+                    break
+
+                elif not result.is_published():
+                    self.logger.debug(
+                        'Blocking until MQTT payload is published...')
+                    reference = datetime.now()
+                    while not has_error and not result.is_published():
+                        # Throttle
+                        sleep(self.mqtt_block_time_sec)
+
+                        # Our own throttle so we can abort eventually....
+                        elapsed = (datetime.now() - reference).total_seconds()
+                        if elapsed >= self.socket_read_timeout:
+                            self.logger.warning(
+                                'The MQTT message could not be delivered')
+                            has_error = True
+
+                # if we reach here; we're at the bottom of our loop
+                # we loop around and do the next topic now
+
+        except ConnectionError as e:
+            self.logger.warning(
+                'MQTT Connection Error received from {}'.format(url))
+            self.logger.debug('Socket Exception: %s' % str(e))
+            return False
+
+        except ssl.CertificateError as e:
+            self.logger.warning(
+                'MQTT SSL Certificate Error received from {}'.format(url))
+            self.logger.debug('Socket Exception: %s' % str(e))
+            return False
+
+        except ValueError as e:
+            # ValueError's are thrown from publish() call if there is a problem
+            self.logger.warning(
+                'MQTT Publishing error received: from {}'.format(url))
+            self.logger.debug('Socket Exception: %s' % str(e))
+            return False
+
+        return not has_error
+
+    def url(self, privacy=False, *args, **kwargs):
+        """
+        Returns the URL built dynamically based on specified arguments.
+        """
+
+        # Define any URL parameters
+        params = {
+            'version': self.version,
+            'qos': str(self.qos),
+            'session': 'yes' if self.session else 'no',
+        }
+
+        if self.client_id:
+            # Our client id is set if specified
+            params['client_id'] = self.client_id
+
+        # Extend our parameters
+        params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
+
+        # Determine Authentication
+        auth = ''
+        if self.user and self.password:
+            auth = '{user}:{password}@'.format(
+                user=NotifyMQTT.quote(self.user, safe=''),
+                password=self.pprint(
+                    self.password, privacy, mode=PrivacyMode.Secret, safe=''),
+            )
+        elif self.user:
+            auth = '{user}@'.format(
+                user=NotifyMQTT.quote(self.user, safe=''),
+            )
+
+        default_port = self.mqtt_secure_port \
+            if self.secure else self.mqtt_insecure_port
+
+        return '{schema}://{auth}{hostname}{port}/{targets}?{params}'.format(
+            schema=self.secure_protocol if self.secure else self.protocol,
+            auth=auth,
+            # never encode hostname since we're expecting it to be a valid one
+            hostname=self.host,
+            port='' if self.port is None or self.port == default_port
+                 else ':{}'.format(self.port),
+            targets=','.join(
+                [NotifyMQTT.quote(x, safe='/') for x in self.topics]),
+            params=NotifyMQTT.urlencode(params),
+        )
+
+    @staticmethod
+    def parse_url(url):
+        """
+        There are no parameters nessisary for this protocol; simply having
+        windows:// is all you need.  This function just makes sure that
+        is in place.
+
+        """
+
+        results = NotifyBase.parse_url(url)
+        if not results:
+            # We're done early as we couldn't load the results
+            return results
+
+        try:
+            # Acquire topic(s)
+            results['targets'] = parse_list(
+                NotifyMQTT.unquote(results['fullpath'].lstrip('/')))
+
+        except AttributeError:
+            # No 'fullpath' specified
+            results['targets'] = []
+
+        # The MQTT protocol version to use
+        if 'version' in results['qsd'] and len(results['qsd']['version']):
+            results['version'] = \
+                NotifyMQTT.unquote(results['qsd']['version'])
+
+        # The MQTT Client ID
+        if 'client_id' in results['qsd'] and len(results['qsd']['client_id']):
+            results['client_id'] = \
+                NotifyMQTT.unquote(results['qsd']['client_id'])
+
+        if 'session' in results['qsd'] and len(results['qsd']['session']):
+            results['session'] = parse_bool(results['qsd']['session'])
+
+        # The MQTT Quality of Service to use
+        if 'qos' in results['qsd'] and len(results['qsd']['qos']):
+            results['qos'] = \
+                NotifyMQTT.unquote(results['qsd']['qos'])
+
+        # The 'to' makes it easier to use yaml configuration
+        if 'to' in results['qsd'] and len(results['qsd']['to']):
+            results['targets'].extend(
+                NotifyMQTT.parse_list(results['qsd']['to']))
+
+        # return results
+        return results

dev-requirements.txt

@@ -16,3 +16,6 @@ slixmpp; python_version >= '3.7'
 
 # Provides growl:// support
 gntp
+
+# Provides mqtt:// support
+paho-mqtt

packaging/redhat/python-apprise.spec

@@ -50,7 +50,7 @@ it easy to access:
 Boxcar, ClickSend, DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, Gitter,
 Google Chat, Gotify, Growl, Home Assistant, IFTTT, Join, Kavenegar, KODI,
 Kumulos, LaMetric, MacOSX, Mailgun, Mattermost, Matrix, Microsoft Windows,
-Microsoft Teams, MessageBird, MSG91, MyAndroid, Nexmo, Nextcloud, Notica,
+Microsoft Teams, MessageBird, MQTT, MSG91, MyAndroid, Nexmo, Nextcloud, Notica,
 Notifico, Office365, OneSignal, Opsgenie, ParsePlatform, PopcornNotify, Prowl,
 Pushalot, PushBullet, Pushjet, Pushover, PushSafer, Reddit, Rocket.Chat,
 SendGrid, SimplePush, Sinch, Slack, SMTP2Go, Spontit, SparkPost, Super Toasty,

setup.py

@@ -72,8 +72,8 @@ setup(
     keywords='Push Notifications Alerts Email AWS SNS Boxcar ClickSend '
         'Dingtalk Discord Dbus Emby Faast FCM Flock Gitter Gnome Google Chat '
         'Gotify Growl Home Assistant IFTTT Join Kavenegar KODI Kumulos '
-        'LaMetric MacOS Mailgun Matrix Mattermost MessageBird MSG91 Nexmo '
+        'LaMetric MacOS Mailgun Matrix Mattermost MessageBird MQTT MSG91 '
-        'Nextcloud Notica Notifico Office365 OneSignal Opsgenie '
+        'Nexmo Nextcloud Notica Notifico Office365 OneSignal Opsgenie '
         'ParsePlatform PopcornNotify Prowl PushBullet Pushjet Pushed '
         'Pushover PushSafer Reddit Rocket.Chat Ryver SendGrid SimplePush '
         'Sinch Slack SMTP2Go SparkPost Spontit Stride Syslog Techulus '

test/test_mqtt_plugin.py

@@ -0,0 +1,318 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2021 Chris Caron <lead2gold@gmail.com>
+# 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 mock
+import re
+import sys
+import ssl
+import six
+import os
+import pytest
+
+# Rebuild our Apprise environment
+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
+
+# Disable logging for a cleaner testing output
+import logging
+logging.disable(logging.CRITICAL)
+
+
+@pytest.mark.skipif('paho' not in sys.modules, reason="requires paho-mqtt")
+def test_paho_mqtt_plugin_import_error(tmpdir):
+    """
+    API: NotifyMQTT 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 paho dependancy.  Since
+    # paho 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 paho (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("paho")
+    suite.join("__init__.py").write('')
+    module_name = 'paho'
+    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]
+
+    # 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.NotifyMQTT'])
+    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('mqtt://user:pass@localhost/my/topic')
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    # We can still retrieve our url back to us
+    assert obj.url().startswith('mqtt://user:pass@localhost/my/topic')
+    # Notifications are not possible
+    assert obj.notify(body="test") is False
+
+    # 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.NotifyMQTT'])
+    reload(sys.modules['apprise.plugins'])
+    reload(sys.modules['apprise.Apprise'])
+    reload(sys.modules['apprise'])
+
+
+@pytest.mark.skipif(
+    'paho' not in sys.modules, reason="requires paho-mqtt")
+@mock.patch('paho.mqtt.client.Client')
+def test_mqtt_plugin(mock_client):
+    """
+    API: NotifyMQTT Plugin()
+
+    """
+    # Speed up request rate for testing
+    apprise.plugins.NotifyBase.request_rate_per_sec = 0
+
+    # our call to publish() response object
+    publish_result = mock.Mock()
+    publish_result.rc = 0
+    publish_result.is_published.return_value = True
+
+    # Our mqtt.Client() object
+    _mock_client = mock.Mock()
+    _mock_client.connect.return_value = 0
+    _mock_client.reconnect.return_value = 0
+    _mock_client.is_connected.return_value = True
+    _mock_client.publish.return_value = publish_result
+    mock_client.return_value = _mock_client
+
+    # Instantiate our object
+    obj = apprise.Apprise.instantiate(
+        'mqtt://localhost:1234/my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.url().startswith('mqtt://localhost:1234/my/topic')
+    # Detect our defaults
+    assert re.search(r'qos=0', obj.url())
+    assert re.search(r'version=v3.1.1', obj.url())
+    # Send a good notification
+    assert obj.notify(body="test=test") is True
+
+    # leverage the to= argument to identify our topic
+    obj = apprise.Apprise.instantiate(
+        'mqtt://localhost?to=my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.url().startswith('mqtt://localhost/my/topic')
+    # Detect our defaults
+    assert re.search(r'qos=0', obj.url())
+    assert re.search(r'version=v3.1.1', obj.url())
+    # Send a good notification
+    assert obj.notify(body="test=test") is True
+
+    # Send a notification in a situation where our publish failed
+    publish_result.rc = 2
+    assert obj.notify(body="test=test") is False
+    # Toggle our response object back to what it should be
+    publish_result.rc = 0
+
+    # Test case where we provide an invalid/unsupported mqtt version
+    with pytest.raises(TypeError):
+        obj = apprise.Apprise.instantiate(
+            'mqtt://localhost?version=v1.0.0.0', suppress_exceptions=False)
+
+    # Test case where we provide an invalid/unsupported qos
+    with pytest.raises(TypeError):
+        obj = apprise.Apprise.instantiate(
+            'mqtt://localhost?qos=123', suppress_exceptions=False)
+    with pytest.raises(TypeError):
+        obj = apprise.Apprise.instantiate(
+            'mqtt://localhost?qos=invalid', suppress_exceptions=False)
+
+    # Test a bad URL
+    obj = apprise.Apprise.instantiate('mqtt://', suppress_exceptions=False)
+    assert obj is None
+
+    # Instantiate our object without any topics
+    # we also test that we can set our qos and version if we want from
+    # the URL
+    obj = apprise.Apprise.instantiate(
+        'mqtt://localhost?qos=1&version=v3.1', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.url().startswith('mqtt://localhost')
+    assert re.search(r'qos=1', obj.url())
+    assert re.search(r'version=v3.1', obj.url())
+    assert re.search(r'session=no', obj.url())
+    assert re.search(r'client_id=', obj.url()) is None
+
+    # Our notification will fail because we have no topics to notify
+    assert obj.notify(body="test=test") is False
+
+    # A Secure URL
+    obj = apprise.Apprise.instantiate(
+        'mqtts://user:pass@localhost/my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.url().startswith('mqtts://user:pass@localhost/my/topic')
+    assert obj.notify(body="test=test") is True
+
+    # Clear CA Certificates
+    ca_certs_backup = \
+        list(apprise.plugins.NotifyMQTT.CA_CERTIFICATE_FILE_LOCATIONS)
+    apprise.plugins.NotifyMQTT.CA_CERTIFICATE_FILE_LOCATIONS = []
+    obj = apprise.Apprise.instantiate(
+        'mqtts://user:pass@localhost/my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.url().startswith('mqtts://user:pass@localhost/my/topic')
+
+    # A notification is not possible now (without ca_certs)
+    assert obj.notify(body="test=test") is False
+
+    # Restore our certificates (for future tests)
+    apprise.plugins.NotifyMQTT.CA_CERTIFICATE_FILE_LOCATIONS = ca_certs_backup
+
+    # A single user (not password) + no verifying of host
+    obj = apprise.Apprise.instantiate(
+        'mqtts://user@localhost/my/topic,my/other/topic?verify=False',
+        suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.url().startswith('mqtts://user@localhost')
+    assert re.search(r'my/other/topic', obj.url())
+    assert re.search(r'my/topic', obj.url())
+    assert obj.notify(body="test=test") is True
+
+    # Session and client_id handling
+    obj = apprise.Apprise.instantiate(
+        'mqtts://user@localhost/my/topic?session=yes&client_id=apprise',
+        suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.url().startswith('mqtts://user@localhost')
+    assert re.search(r'my/topic', obj.url())
+    assert re.search(r'client_id=apprise', obj.url())
+    assert re.search(r'session=yes', obj.url())
+    assert obj.notify(body="test=test") is True
+
+    # handle case where we fail to connect
+    _mock_client.connect.return_value = 2
+    obj = apprise.Apprise.instantiate(
+        'mqtt://localhost/my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.notify(body="test=test") is False
+    # Restore our values
+    _mock_client.connect.return_value = 0
+
+    # handle case where we fail to reconnect
+    _mock_client.reconnect.return_value = 2
+    _mock_client.is_connected.return_value = False
+    obj = apprise.Apprise.instantiate(
+        'mqtt://localhost/my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.notify(body="test=test") is False
+    # Restore our values
+    _mock_client.reconnect.return_value = 0
+    _mock_client.is_connected.return_value = True
+
+    # handle case where we fail to publish()
+    publish_result.rc = 2
+    obj = apprise.Apprise.instantiate(
+        'mqtt://localhost/my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    assert obj.notify(body="test=test") is False
+    # Restore our values
+    publish_result.rc = 0
+    # Set another means of failing publish()
+    publish_result.is_published.return_value = False
+    assert obj.notify(body="test=test") is False
+    # Restore our values
+    publish_result.is_published.return_value = True
+    # Verify that was all we had to do
+    assert obj.notify(body="test=test") is True
+    # A slight variation on the same failure (but with recovery)
+    publish_result.is_published.return_value = None
+    publish_result.is_published.side_effect = (False, True)
+    # Our notification is still sent okay
+    assert obj.notify(body="test=test") is True
+
+    # Exception handling
+    obj = apprise.Apprise.instantiate(
+        'mqtt://localhost/my/topic', suppress_exceptions=False)
+    assert isinstance(obj, apprise.plugins.NotifyMQTT)
+    _mock_client.connect.return_value = None
+
+    if six.PY2:
+        # Python v2.7 does not support the ConnectionError exception
+        for side_effect in (ValueError, ssl.CertificateError):
+            _mock_client.connect.side_effect = side_effect
+            assert obj.notify(body="test=test") is False
+
+    else:
+        for side_effect in (
+                ValueError,
+                # Python 2.7 doesn't recognize ConnectionError so for that
+                # reason we stick a noqa entry here...
+                ConnectionError,  # noqa: F821
+                ssl.CertificateError):
+            _mock_client.connect.side_effect = side_effect
+            assert obj.notify(body="test=test") is False
+
+    # Restore our values
+    _mock_client.connect.side_effect = None
+    _mock_client.connect.return_value = 0