diff --git a/CONTRIBUTIONS.md b/CONTRIBUTIONS.md index 0fd3a2be..f54ec1cf 100644 --- a/CONTRIBUTIONS.md +++ b/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 + * 6th Feb 2024 - Add Revolt Support diff --git a/KEYWORDS b/KEYWORDS index 20db7817..b23a1981 100644 --- a/KEYWORDS +++ b/KEYWORDS @@ -71,6 +71,7 @@ PushSafer Pushy PushDeer Reddit +Revolt Rocket.Chat RSyslog Ryver diff --git a/README.md b/README.md index 83efe989..33524ee1 100644 --- a/README.md +++ b/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
pushy://apikey/DEVICE1/DEVICE2/DEVICEN
pushy://apikey/TOPIC
pushy://apikey/TOPIC1/TOPIC2/TOPICN | [PushDeer](https://github.com/caronc/apprise/wiki/Notify_pushdeer) | pushdeer:// or pushdeers:// | (TCP) 80 or 443 | pushdeer://pushKey
pushdeer://hostname/pushKey
pushdeer://hostname:port/pushKey | [Reddit](https://github.com/caronc/apprise/wiki/Notify_reddit) | reddit:// | (TCP) 443 | reddit://user:password@app_id/app_secret/subreddit
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
rockets://user:password@hostname:443/#Channel1/#Channel1/RoomID
rocket://user:password@hostname/#Channel
rocket://webhook@hostname
rockets://webhook@hostname/@User/#Channel | [RSyslog](https://github.com/caronc/apprise/wiki/Notify_rsyslog) | rsyslog:// | (UDP) 514 | rsyslog://hostname
rsyslog://hostname/Facility | [Ryver](https://github.com/caronc/apprise/wiki/Notify_ryver) | ryver:// | (TCP) 443 | ryver://Organization/Token
ryver://botname@Organization/Token diff --git a/apprise/plugins/NotifyRevolt.py b/apprise/plugins/NotifyRevolt.py new file mode 100644 index 00000000..24cfeee6 --- /dev/null +++ b/apprise/plugins/NotifyRevolt.py @@ -0,0 +1,415 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2024, Chris Caron +# +# 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 diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index bc3d51bb..4e6b74a1 100644 --- a/packaging/redhat/python-apprise.spec +++ b/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, -ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go, -SparkPost, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog, +Pushover, PushSafer, Pushy, PushDeer, Revolt, Reddit, Rocket.Chat, RSyslog, +SendGrid, ServerChan, Signal, SimplePush, Sinch, Slack, SMSEagle, SMS Manager, +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} diff --git a/test/test_plugin_revolt.py b/test/test_plugin_revolt.py new file mode 100644 index 00000000..1701b782 --- /dev/null +++ b/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 +# +# 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