Revolt Support (#1057)

CONTRIBUTIONS.md

@@ -30,3 +30,6 @@ The contributors have been listed in chronological order:
 
 * Joey Espinosa <@particledecay>
   * Apr 3rd 2022 - Added Ntfy Support
+
+* Kate Ward <https://kate.pet>
+  * 6th Feb 2024 - Add Revolt Support

KEYWORDS

@@ -71,6 +71,7 @@ PushSafer
 Pushy
 PushDeer
 Reddit
+Revolt
 Rocket.Chat
 RSyslog
 Ryver

README.md

@@ -108,6 +108,7 @@ The table below identifies the services this tool supports and some example serv
 | [Pushy](https://github.com/caronc/apprise/wiki/Notify_pushy)  | pushy://  | (TCP) 443  | pushy://apikey/DEVICE<br />pushy://apikey/DEVICE1/DEVICE2/DEVICEN<br />pushy://apikey/TOPIC<br />pushy://apikey/TOPIC1/TOPIC2/TOPICN
 | [PushDeer](https://github.com/caronc/apprise/wiki/Notify_pushdeer) | pushdeer:// or pushdeers:// | (TCP) 80 or 443 | pushdeer://pushKey<br />pushdeer://hostname/pushKey<br />pushdeer://hostname:port/pushKey
 | [Reddit](https://github.com/caronc/apprise/wiki/Notify_reddit) | reddit:// | (TCP) 443   | reddit://user:password@app_id/app_secret/subreddit<br />reddit://user:password@app_id/app_secret/sub1/sub2/subN
+| [Revolt](https://github.com/caronc/apprise/wiki/Notify_Revolt) | revolt:// | (TCP) 443   | revolt://bot_token/channel_id |
 | [Rocket.Chat](https://github.com/caronc/apprise/wiki/Notify_rocketchat) | rocket:// or rockets://  | (TCP) 80 or 443   | rocket://user:password@hostname/RoomID/Channel<br />rockets://user:password@hostname:443/#Channel1/#Channel1/RoomID<br />rocket://user:password@hostname/#Channel<br />rocket://webhook@hostname<br />rockets://webhook@hostname/@User/#Channel
 | [RSyslog](https://github.com/caronc/apprise/wiki/Notify_rsyslog) | rsyslog://  | (UDP) 514 | rsyslog://hostname<br />rsyslog://hostname/Facility
 | [Ryver](https://github.com/caronc/apprise/wiki/Notify_ryver) | ryver://  | (TCP) 443   | ryver://Organization/Token<br />ryver://botname@Organization/Token

apprise/plugins/NotifyRevolt.py

@@ -0,0 +1,415 @@
+# -*- coding: utf-8 -*-
+# BSD 2-Clause License
+#
+# Apprise - Push Notification Library.
+# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice,
+#    this list of conditions and the following disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+# Youll need your own Revolt Bot and a Channel Id for the notifications to
+# be sent in since Revolt does not support webhooks yet.
+#
+#  This plugin will simply work using the url of:
+#     revolt://BOT_TOKEN/CHANNEL_ID
+#
+# API Documentation:
+#    - https://api.revolt.chat/swagger/index.html
+#
+
+import requests
+from json import dumps
+from datetime import timedelta
+from datetime import datetime
+from datetime import timezone
+
+from .NotifyBase import NotifyBase
+from ..common import NotifyImageSize
+from ..common import NotifyFormat
+from ..common import NotifyType
+from ..utils import parse_bool
+from ..utils import validate_regex
+from ..AppriseLocale import gettext_lazy as _
+
+
+class NotifyRevolt(NotifyBase):
+    """
+    A wrapper for Revolt Notifications
+
+    """
+    # The default descriptive name associated with the Notification
+    service_name = 'Revolt'
+
+    # The services URL
+    service_url = 'https://api.revolt.chat/'
+
+    # The default secure protocol
+    secure_protocol = 'revolt'
+
+    # A URL that takes you to the setup/help of the specific protocol
+    setup_url = 'https://github.com/caronc/apprise/wiki/Notify_revolt'
+
+    # Revolt Channel Message
+    notify_url = 'https://api.revolt.chat/'
+
+    # Revolt supports attachments but don't implemenet for now
+    attachment_support = False
+
+    # Allows the user to specify the NotifyImageSize object
+    image_size = NotifyImageSize.XY_256
+
+    # Revolt is kind enough to return how many more requests we're allowed to
+    # continue to make within it's header response as:
+    # X-RateLimit-Reset: The epoc time (in seconds) we can expect our
+    #                    rate-limit to be reset.
+    # X-RateLimit-Remaining: an integer identifying how many requests we're
+    #                        still allow to make.
+    request_rate_per_sec = 3
+
+    # Taken right from google.auth.helpers:
+    clock_skew = timedelta(seconds=10)
+
+    # The maximum allowable characters allowed in the body per message
+    body_maxlen = 2000
+
+    # Title Maximum Length
+    title_maxlen = 100
+
+    # Define object templates
+    templates = (
+        '{schema}://{bot_token}/{channel_id}',
+    )
+
+    # Defile out template tokens
+    template_tokens = dict(NotifyBase.template_tokens, **{
+        'bot_token': {
+            'name': _('Bot Token'),
+            'type': 'string',
+            'private': True,
+            'required': True,
+        },
+        'channel_id': {
+            'name': _('Channel Id'),
+            'type': 'string',
+            'private': True,
+            'required': True,
+        },
+    })
+
+    # Define our template arguments
+    template_args = dict(NotifyBase.template_args, **{
+        'channel_id': {
+            'alias_of': 'channel_id',
+        },
+        'bot_token': {
+            'alias_of': 'bot_token',
+        },
+        'embed_img': {
+            'name': _('Embed Image Url'),
+            'type': 'string'
+        },
+        'embed_url': {
+            'name': _('Embed Url'),
+            'type': 'string'
+        },
+        'custom_img': {
+            'name': _('Custom Embed Url'),
+            'type': 'bool',
+            'default': False
+        }
+    })
+
+    def __init__(self, bot_token, channel_id, embed_img=None, embed_url=None,
+                 custom_img=None, **kwargs):
+        super().__init__(**kwargs)
+
+        # Bot Token
+        self.bot_token = validate_regex(bot_token)
+        if not self.bot_token:
+            msg = 'An invalid Revolt Bot Token ' \
+                '({}) was specified.'.format(bot_token)
+            self.logger.warning(msg)
+            raise TypeError(msg)
+
+        # Channel Id
+        self.channel_id = validate_regex(channel_id)
+        if not self.channel_id:
+            msg = 'An invalid Revolt Channel Id' \
+                '({}) was specified.'.format(channel_id)
+            self.logger.warning(msg)
+            raise TypeError(msg)
+
+        # Use custom image for embed image
+        self.custom_img = parse_bool(custom_img) \
+            if custom_img is not None \
+            else self.template_args['custom_img']['default']
+
+        # Image for Embed
+        self.embed_img = embed_img
+
+        # Url for embed title
+        self.embed_url = embed_url
+
+        # For Tracking Purposes
+        self.ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None)
+
+        # Default to 1.0
+        self.ratelimit_remaining = 1.0
+
+        return
+
+    def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
+        """
+        Perform Revolt Notification
+
+        """
+
+        payload = {}
+
+        # Acquire image_url
+        image_url = self.image_url(notify_type)
+
+        if self.custom_img and (image_url or self.embed_url):
+            image_url = self.embed_url if self.embed_url else image_url
+
+        if body:
+            if self.notify_format == NotifyFormat.MARKDOWN:
+                if len(title) > 100:
+                    msg = 'Title length must be less than 100 when ' \
+                        'embeds are enabled (is %s)' % len(title)
+                    self.logger.warning(msg)
+                    title = title[0:100]
+                payload['embeds'] = [{
+                    'title': title,
+                    'description': body,
+
+                    # Our color associated with our notification
+                    'colour': self.color(notify_type, int)
+                }]
+
+                if self.embed_img:
+                    payload['embeds'][0]['icon_url'] = image_url
+
+                if self.embed_url:
+                    payload['embeds'][0]['url'] = self.embed_url
+            else:
+                payload['content'] = \
+                    body if not title else "{}\n{}".format(title, body)
+
+            if not self._send(payload):
+                # Failed to send message
+                return False
+        return True
+
+    def _send(self, payload, rate_limit=1, **kwargs):
+        """
+        Wrapper to the requests (post) object
+
+        """
+
+        headers = {
+            'User-Agent': self.app_id,
+            'X-Bot-Token': self.bot_token,
+            'Content-Type': 'application/json; charset=utf-8'
+        }
+
+        notify_url = '{0}channels/{1}/messages'.format(
+            self.notify_url,
+            self.channel_id
+        )
+
+        self.logger.debug('Revolt POST URL: %s (cert_verify=%r)' % (
+            notify_url, self.verify_certificate
+        ))
+        self.logger.debug('Revolt Payload: %s' % str(payload))
+
+        # By default set wait to None
+        wait = None
+
+        if self.ratelimit_remaining <= 0.0:
+            # Determine how long we should wait for or if we should wait at
+            # all. This isn't fool-proof because we can't be sure the client
+            # time (calling this script) is completely synced up with the
+            # Discord server.  One would hope we're on NTP and our clocks are
+            # the same allowing this to role smoothly:
+
+            now = datetime.now(timezone.utc).replace(tzinfo=None)
+            if now < self.ratelimit_reset:
+                # We need to throttle for the difference in seconds
+                wait = abs(
+                    (self.ratelimit_reset - now + self.clock_skew)
+                    .total_seconds())
+
+        # Always call throttle before any remote server i/o is made;
+        self.throttle(wait=wait)
+
+        try:
+            r = requests.post(
+                notify_url,
+                data=dumps(payload),
+                headers=headers,
+                verify=self.verify_certificate,
+                timeout=self.request_timeout
+            )
+
+            # Handle rate limiting (if specified)
+            try:
+                # Store our rate limiting (if provided)
+                self.ratelimit_remaining = \
+                    float(r.headers.get(
+                        'X-RateLimit-Remaining'))
+                self.ratelimit_reset = datetime.fromtimestamp(
+                    int(r.headers.get('X-RateLimit-Reset')),
+                    timezone.utc).replace(tzinfo=None)
+
+            except (TypeError, ValueError):
+                # This is returned if we could not retrieve this
+                # information gracefully accept this state and move on
+                pass
+
+            if r.status_code not in (
+                    requests.codes.ok, requests.codes.no_content):
+
+                # We had a problem
+                status_str = \
+                    NotifyBase.http_response_code_lookup(r.status_code)
+
+                if r.status_code == requests.codes.too_many_requests \
+                        and rate_limit > 0:
+
+                    # handle rate limiting
+                    self.logger.warning(
+                        'Revolt rate limiting in effect; '
+                        'blocking for %.2f second(s)',
+                        self.ratelimit_remaining)
+
+                    # Try one more time before failing
+                    return self._send(
+                        payload=payload,
+                        rate_limit=rate_limit - 1, **kwargs)
+
+                self.logger.warning(
+                    'Failed to send to Revolt notification: '
+                    '{}{}error={}.'.format(
+                        status_str,
+                        ', ' if status_str else '',
+                        r.status_code))
+
+                self.logger.debug('Response Details:\r\n{}'.format(r.content))
+
+                # Return; we're done
+                return False
+
+            else:
+                self.logger.info('Sent Revolt notification.')
+
+        except requests.RequestException as e:
+            self.logger.warning(
+                'A Connection error occurred posting to Revolt.')
+            self.logger.debug('Socket Exception: %s' % str(e))
+            return False
+
+        return True
+
+    def url(self, privacy=False, *args, **kwargs):
+        """
+        Returns the URL built dynamically based on specified arguments.
+
+        """
+
+        params = {}
+
+        if self.embed_img:
+            params['embed_img'] = self.embed_img
+
+        if self.embed_url:
+            params['embed_url'] = self.embed_url
+
+        params.update(self.url_parameters(privacy=privacy, *args, **kwargs))
+
+        return '{schema}://{bot_token}/{channel_id}/?{params}'.format(
+            schema=self.secure_protocol,
+            bot_token=self.pprint(self.bot_token, privacy, safe=''),
+            channel_id=self.pprint(self.channel_id, privacy, safe=''),
+            params=NotifyRevolt.urlencode(params),
+        )
+
+    @staticmethod
+    def parse_url(url):
+        """
+        Parses the URL and returns enough arguments that can allow
+        us to re-instantiate this object.
+
+        Syntax:
+          revolt://bot_token/channel_id
+
+        """
+        results = NotifyBase.parse_url(url, verify_host=False)
+        if not results:
+            # We're done early as we couldn't load the results
+            return results
+
+        # Store our bot token
+        bot_token = NotifyRevolt.unquote(results['host'])
+
+        # Now fetch the channel id
+        try:
+            channel_id = \
+                NotifyRevolt.split_path(results['fullpath'])[0]
+
+        except IndexError:
+            # Force some bad values that will get caught
+            # in parsing later
+            channel_id = None
+
+        results['bot_token'] = bot_token
+        results['channel_id'] = channel_id
+
+        # Text To Speech
+        results['tts'] = parse_bool(results['qsd'].get('tts', False))
+
+        # Support channel id on the URL string (if specified)
+        if 'channel_id' in results['qsd']:
+            results['channel_id'] = \
+                NotifyRevolt.unquote(results['qsd']['channel_id'])
+
+        # Support bot token on the URL string (if specified)
+        if 'bot_token' in results['qsd']:
+            results['bot_token'] = \
+                NotifyRevolt.unquote(results['qsd']['bot_token'])
+
+        # Extract avatar url if it was specified
+        if 'embed_img' in results['qsd']:
+            results['embed_img'] = \
+                NotifyRevolt.unquote(results['qsd']['embed_img'])
+
+        if 'custom_img' in results['qsd']:
+            results['custom_img'] = \
+                NotifyRevolt.unquote(results['qsd']['custom_img'])
+
+        elif 'embed_url' in results['qsd']:
+            results['embed_url'] = \
+                NotifyRevolt.unquote(results['qsd']['embed_url'])
+            # Markdown is implied
+            results['format'] = NotifyFormat.MARKDOWN
+
+        return results

packaging/redhat/python-apprise.spec

@@ -47,9 +47,9 @@ Mattermost,Matrix, MessageBird, Microsoft Windows, Microsoft Teams, Misskey,
 MQTT, MSG91, MyAndroid, Nexmo, Nextcloud, NextcloudTalk, Notica, Notifiarr,
 Notifico, ntfy, Office365, OneSignal, Opsgenie, PagerDuty, PagerTree,
 ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, PushMe,
-Pushover, PushSafer, Pushy, PushDeer, Reddit, Rocket.Chat, RSyslog, SendGrid,
+Pushover, PushSafer, Pushy, PushDeer, Revolt, Reddit, Rocket.Chat, RSyslog,
-ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go,
+SendGrid, ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMS Manager,
-SparkPost, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog,
+SMTP2Go, SparkPost, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog,
 Techulus Push, Telegram, Threema Gateway, Twilio, Twitter, Twist, XBMC,
 Voipms, Vonage, WeCom Bot, WhatsApp, Webex Teams}
 

test/test_plugin_revolt.py

@@ -0,0 +1,459 @@
+# -*- coding: utf-8 -*-
+# BSD 2-Clause License
+#
+# Apprise - Push Notification Library.
+# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice,
+#    this list of conditions and the following disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+#    this list of conditions and the following disclaimer in the documentation
+#    and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+import os
+from unittest import mock
+from datetime import datetime, timedelta
+from datetime import timezone
+import pytest
+import requests
+
+from apprise.plugins.NotifyRevolt import NotifyRevolt
+from helpers import AppriseURLTester
+from apprise import Apprise
+from apprise import NotifyType
+from apprise import NotifyFormat
+from apprise.common import OverflowMode
+
+from random import choice
+from string import ascii_uppercase as str_alpha
+from string import digits as str_num
+
+# Disable logging for a cleaner testing output
+import logging
+logging.disable(logging.CRITICAL)
+
+# Attachment Directory
+TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
+
+# Our Testing URLs
+apprise_url_tests = (
+    ('revolt://', {
+        'instance': TypeError,
+    }),
+    # An invalid url
+    ('revolt://:@/', {
+        'instance': TypeError,
+    }),
+    # No channel_id specified
+    ('revolt://%s' % ('i' * 24), {
+        'instance': TypeError,
+    }),
+    # channel_id specified on url
+    ('revolt://?channel_id=%s' % ('i' * 24), {
+        'instance': TypeError,
+    }),
+    # Provide both a bot token and a channel id
+    ('revolt://%s/%s' % ('i' * 24, 't' * 64), {
+        'instance': NotifyRevolt,
+        'requests_response_code': requests.codes.no_content,
+    }),
+    # Provide a temporary username
+    ('revolt://l2g@%s/%s' % ('i' * 24, 't' * 64), {
+        'instance': NotifyRevolt,
+        'requests_response_code': requests.codes.no_content,
+    }),
+    ('revolt://l2g@_?bot_token=%s&channel_id=%s' % ('i' * 24, 't' * 64), {
+        'instance': NotifyRevolt,
+        'requests_response_code': requests.codes.no_content,
+    }),
+    # test custom_img= field
+    ('revolt://%s/%s?format=markdown&custom_img=Yes' % (
+        'i' * 24, 't' * 64), {
+            'instance': NotifyRevolt,
+            'requests_response_code': requests.codes.no_content,
+    }),
+    ('revolt://%s/%s?format=markdown&custom_img=No' % (
+        'i' * 24, 't' * 64), {
+            'instance': NotifyRevolt,
+            'requests_response_code': requests.codes.no_content,
+    }),
+    # different format support
+    ('revolt://%s/%s?format=markdown' % ('i' * 24, 't' * 64), {
+        'instance': NotifyRevolt,
+        'requests_response_code': requests.codes.no_content,
+    }),
+    ('revolt://%s/%s?format=text' % ('i' * 24, 't' * 64), {
+        'instance': NotifyRevolt,
+        'requests_response_code': requests.codes.no_content,
+    }),
+    # Test with embed_url (title link)
+    ('revolt://%s/%s?hmarkdown=true&embed_url=http://localhost' % (
+        'i' * 24, 't' * 64), {
+            'instance': NotifyRevolt,
+            'requests_response_code': requests.codes.no_content,
+    }),
+    # Test with avatar URL
+    ('revolt://%s/%s?embed_img=http://localhost/test.jpg' % (
+        'i' * 24, 't' * 64), {
+            'instance': NotifyRevolt,
+            'requests_response_code': requests.codes.no_content,
+    }),
+    # Test without image set
+    ('revolt://%s/%s' % ('i' * 24, 't' * 64), {
+        'instance': NotifyRevolt,
+        'requests_response_code': requests.codes.no_content,
+        # don't include an image by default
+        'embed_img': False,
+    }),
+    ('revolt://%s/%s/' % ('a' * 24, 'b' * 64), {
+        'instance': NotifyRevolt,
+        # force a failure
+        'response': False,
+        'requests_response_code': requests.codes.internal_server_error,
+    }),
+    ('revolt://%s/%s/' % ('a' * 24, 'b' * 64), {
+        'instance': NotifyRevolt,
+        # throw a bizzare code forcing us to fail to look it up
+        'response': False,
+        'requests_response_code': 999,
+    }),
+    ('revolt://%s/%s/' % ('a' * 24, 'b' * 64), {
+        'instance': NotifyRevolt,
+        # Throws a series of connection and transfer exceptions when this flag
+        # is set and tests that we gracfully handle them
+        'test_requests_exceptions': True,
+    }),
+)
+
+
+def test_plugin_revolt_urls():
+    """
+    NotifyRevolt() Apprise URLs
+
+    """
+
+    # Run our general tests
+    AppriseURLTester(tests=apprise_url_tests).run_all()
+
+
+@mock.patch('requests.post')
+def test_plugin_revolt_notifications(mock_post):
+    """
+    NotifyRevolt() Notifications/Ping Support
+
+    """
+
+    # Initialize some generic (but valid) tokens
+    bot_token = 'A' * 24
+    channel_id = 'B' * 64
+
+    # Prepare Mock
+    mock_post.return_value = requests.Request()
+    mock_post.return_value.status_code = requests.codes.ok
+
+    # Test our header parsing when not lead with a header
+    body = """
+    # Heading
+    @everyone and @admin, wake and meet our new user <@123>; <@&456>"
+    """
+
+    results = NotifyRevolt.parse_url(
+        f'revolt://{bot_token}/{channel_id}/?format=markdown')
+
+    assert isinstance(results, dict)
+    assert results['user'] is None
+    assert results['bot_token'] == bot_token
+    assert results['channel_id'] == channel_id
+    assert results['password'] is None
+    assert results['port'] is None
+    assert results['host'] == bot_token
+    assert results['fullpath'] == f'/{channel_id}/'
+    assert results['path'] == f'/{channel_id}/'
+    assert results['query'] is None
+    assert results['schema'] == 'revolt'
+    assert results['url'] == f'revolt://{bot_token}/{channel_id}/'
+
+    instance = NotifyRevolt(**results)
+    assert isinstance(instance, NotifyRevolt)
+
+    response = instance.send(body=body)
+    assert response is True
+    assert mock_post.call_count == 1
+
+    # Reset our object
+    mock_post.reset_mock()
+
+    results = NotifyRevolt.parse_url(
+        f'revolt://{bot_token}/{channel_id}/?format=text')
+
+    assert isinstance(results, dict)
+    assert results['user'] is None
+    assert results['bot_token'] == bot_token
+    assert results['channel_id'] == channel_id
+    assert results['password'] is None
+    assert results['port'] is None
+    assert results['host'] == bot_token
+    assert results['fullpath'] == f'/{channel_id}/'
+    assert results['path'] == f'/{channel_id}/'
+    assert results['query'] is None
+    assert results['schema'] == 'revolt'
+    assert results['url'] == f'revolt://{bot_token}/{channel_id}/'
+
+    instance = NotifyRevolt(**results)
+    assert isinstance(instance, NotifyRevolt)
+
+    response = instance.send(body=body)
+    assert response is True
+    assert mock_post.call_count == 1
+
+
+@mock.patch('requests.post')
+def test_plugin_revolt_general(mock_post):
+    """
+    NotifyRevolt() General Checks
+
+    """
+
+    # Turn off clock skew for local testing
+    NotifyRevolt.clock_skew = timedelta(seconds=0)
+    # Epoch time:
+    epoch = datetime.fromtimestamp(0, timezone.utc)
+
+    # Initialize some generic (but valid) tokens
+    bot_token = 'A' * 24
+    channel_id = 'B' * 64
+
+    # Prepare Mock
+    mock_post.return_value = requests.Request()
+    mock_post.return_value.status_code = requests.codes.ok
+    mock_post.return_value.content = ''
+    mock_post.return_value.headers = {
+        'X-RateLimit-Reset': (
+            datetime.now(timezone.utc) - epoch).total_seconds(),
+        'X-RateLimit-Remaining': 1,
+    }
+
+    # Invalid bot_token
+    with pytest.raises(TypeError):
+        NotifyRevolt(bot_token=None, channel_id=channel_id)
+    # Invalid bot_token (whitespace)
+    with pytest.raises(TypeError):
+        NotifyRevolt(bot_token="  ", channel_id=channel_id)
+
+    # Invalid channel_id
+    with pytest.raises(TypeError):
+        NotifyRevolt(bot_token=bot_token, channel_id=None)
+    # Invalid channel_id (whitespace)
+    with pytest.raises(TypeError):
+        NotifyRevolt(bot_token=bot_token, channel_id="   ")
+
+    obj = NotifyRevolt(
+        bot_token=bot_token,
+        channel_id=channel_id)
+    assert obj.ratelimit_remaining == 1
+
+    # Test that we get a string response
+    assert isinstance(obj.url(), str) is True
+
+    # This call includes an image with it's payload:
+    assert obj.notify(
+        body='body', title='title', notify_type=NotifyType.INFO) is True
+
+    # Force a case where there are no more remaining posts allowed
+    mock_post.return_value.headers = {
+        'X-RateLimit-Reset': (
+            datetime.now(timezone.utc) - epoch).total_seconds(),
+        'X-RateLimit-Remaining': 0,
+    }
+
+    # This call includes an image with it's payload:
+    assert obj.notify(
+        body='body', title='title', notify_type=NotifyType.INFO) is True
+
+    # behind the scenes, it should cause us to update our rate limit
+    assert obj.send(body="test") is True
+    assert obj.ratelimit_remaining == 0
+
+    # This should cause us to block
+    mock_post.return_value.headers = {
+        'X-RateLimit-Reset': (
+            datetime.now(timezone.utc) - epoch).total_seconds(),
+        'X-RateLimit-Remaining': 10,
+    }
+    assert obj.send(body="test") is True
+    assert obj.ratelimit_remaining == 10
+
+    # Reset our variable back to 1
+    mock_post.return_value.headers = {
+        'X-RateLimit-Reset': (
+            datetime.now(timezone.utc) - epoch).total_seconds(),
+        'X-RateLimit-Remaining': 1,
+    }
+    # Handle cases where our epoch time is wrong
+    del mock_post.return_value.headers['X-RateLimit-Reset']
+    assert obj.send(body="test") is True
+
+    # Return our object, but place it in the future forcing us to block
+    mock_post.return_value.headers = {
+        'X-RateLimit-Reset': (
+            datetime.now(timezone.utc) - epoch).total_seconds() + 1,
+        'X-RateLimit-Remaining': 0,
+    }
+
+    obj.ratelimit_remaining = 0
+    assert obj.send(body="test") is True
+
+    # Test 429 error response
+    mock_post.return_value.status_code = requests.codes.too_many_requests
+
+    # The below will attempt a second transmission and fail (because we didn't
+    # set up a second post request to pass) :)
+    assert obj.send(body="test") is False
+
+    # Return our object, but place it in the future forcing us to block
+    mock_post.return_value.status_code = requests.codes.ok
+    mock_post.return_value.headers = {
+        'X-RateLimit-Reset': (
+            datetime.now(timezone.utc) - epoch).total_seconds() - 1,
+        'X-RateLimit-Remaining': 0,
+    }
+    assert obj.send(body="test") is True
+
+    # Return our limits to always work
+    obj.ratelimit_remaining = 1
+
+    # Return our headers to normal
+    mock_post.return_value.headers = {
+        'X-RateLimit-Reset': (
+            datetime.now(timezone.utc) - epoch).total_seconds(),
+        'X-RateLimit-Remaining': 1,
+    }
+
+    # This call includes an image with it's payload:
+    assert obj.notify(
+        body='body', title='title', notify_type=NotifyType.INFO) is True
+
+    # Create an apprise instance
+    a = Apprise()
+
+    # Our processing is slightly different when we aren't using markdown
+    # as we do not pre-parse content during our notifications
+    assert a.add(
+        'revolt://{bot_token}/{channel_id}/'
+        '?format=markdown'.format(
+            bot_token=bot_token,
+            channel_id=channel_id)) is True
+
+    # Toggle our logo availability
+    a.asset.image_url_logo = None
+    assert a.notify(
+        body='body', title='title', notify_type=NotifyType.INFO) is True
+
+
+@mock.patch('requests.post')
+def test_plugin_revolt_overflow(mock_post):
+    """
+    NotifyRevolt() Overflow Checks
+
+    """
+
+    # Initialize some generic (but valid) tokens
+    bot_token = 'A' * 24
+    channel_id = 'B' * 64
+
+    # Prepare Mock
+    mock_post.return_value = requests.Request()
+    mock_post.return_value.status_code = requests.codes.ok
+
+    # Some variables we use to control the data we work with
+    body_len = 2005
+    title_len = 110
+
+    # Number of characters per line
+    row = 24
+
+    # Create a large body and title with random data
+    body = ''.join(choice(str_alpha + str_num + ' ') for _ in range(body_len))
+    body = '\r\n'.join([body[i: i + row] for i in range(0, len(body), row)])
+
+    # Create our title using random data
+    title = ''.join(choice(str_alpha + str_num) for _ in range(title_len))
+
+    results = NotifyRevolt.parse_url(
+        f'revolt://{bot_token}/{channel_id}/?overflow=split')
+
+    assert isinstance(results, dict)
+    assert results['user'] is None
+    assert results['bot_token'] == bot_token
+    assert results['channel_id'] == channel_id
+    assert results['password'] is None
+    assert results['port'] is None
+    assert results['host'] == bot_token
+    assert results['fullpath'] == f'/{channel_id}/'
+    assert results['path'] == f'/{channel_id}/'
+    assert results['query'] is None
+    assert results['schema'] == 'revolt'
+    assert results['url'] == f'revolt://{bot_token}/{channel_id}/'
+
+    instance = NotifyRevolt(**results)
+    assert isinstance(instance, NotifyRevolt)
+
+    results = instance._apply_overflow(
+        body, title=title, overflow=OverflowMode.SPLIT)
+    # Split into 2
+    assert len(results) == 2
+    assert len(results[0]['title']) <= instance.title_maxlen
+    assert len(results[0]['body']) <= instance.body_maxlen
+
+
+@mock.patch('requests.post')
+def test_plugin_revolt_markdown_extra(mock_post):
+    """
+    NotifyRevolt() Markdown Extra Checks
+
+    """
+
+    # Initialize some generic (but valid) tokens
+    bot_token = 'A' * 24
+    channel_id = 'B' * 64
+
+    # Prepare Mock
+    mock_post.return_value = requests.Request()
+    mock_post.return_value.status_code = requests.codes.ok
+
+    # Reset our apprise object
+    a = Apprise()
+
+    # We want to further test our markdown support to accomodate bug rased on
+    # 2022.10.25; see https://github.com/caronc/apprise/issues/717
+    assert a.add(
+        'revolt://{bot_token}/{channel_id}/'
+        '?format=markdown'.format(
+            bot_token=bot_token,
+            channel_id=channel_id)) is True
+
+    test_markdown = "[green-blue](https://google.com)"
+
+    # This call includes an image with it's payload:
+    assert a.notify(body=test_markdown, title='title',
+                    notify_type=NotifyType.INFO,
+                    body_format=NotifyFormat.TEXT) is True
+
+    assert a.notify(
+        body='body', title='title', notify_type=NotifyType.INFO) is True