From de4701f78077bcffad021ecbc453627452b358ee Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 19 Aug 2023 14:26:44 -0400 Subject: [PATCH] Gitter Support Removed (available through Matrix) (#924) --- KEYWORDS | 1 - README.md | 1 - apprise/plugins/NotifyDiscord.py | 2 +- apprise/plugins/NotifyGitter.py | 433 ------------------------- apprise/plugins/NotifyParsePlatform.py | 2 - apprise/plugins/NotifyReddit.py | 2 +- packaging/redhat/python-apprise.spec | 2 +- test/test_apprise_utils.py | 9 - test/test_plugin_gitter.py | 293 ----------------- 9 files changed, 3 insertions(+), 742 deletions(-) delete mode 100644 apprise/plugins/NotifyGitter.py delete mode 100644 test/test_plugin_gitter.py diff --git a/KEYWORDS b/KEYWORDS index 245e023b..9a726c73 100644 --- a/KEYWORDS +++ b/KEYWORDS @@ -19,7 +19,6 @@ Faast FCM Flock Form -Gitter Gnome Google Chat Gotify diff --git a/README.md b/README.md index c7054603..4af9a7dd 100644 --- a/README.md +++ b/README.md @@ -66,7 +66,6 @@ The table below identifies the services this tool supports and some example serv | [Faast](https://github.com/caronc/apprise/wiki/Notify_faast) | faast:// | (TCP) 443 | faast://authorizationtoken | [FCM](https://github.com/caronc/apprise/wiki/Notify_fcm) | fcm:// | (TCP) 443 | fcm://project@apikey/DEVICE_ID
fcm://project@apikey/#TOPIC
fcm://project@apikey/DEVICE_ID1/#topic1/#topic2/DEVICE_ID2/ | [Flock](https://github.com/caronc/apprise/wiki/Notify_flock) | flock:// | (TCP) 443 | flock://token
flock://botname@token
flock://app_token/u:userid
flock://app_token/g:channel_id
flock://app_token/u:userid/g:channel_id -| [Gitter](https://github.com/caronc/apprise/wiki/Notify_gitter) | gitter:// | (TCP) 443 | gitter://token/room
gitter://token/room1/room2/roomN | [Google Chat](https://github.com/caronc/apprise/wiki/Notify_googlechat) | gchat:// | (TCP) 443 | gchat://workspace/key/token | [Gotify](https://github.com/caronc/apprise/wiki/Notify_gotify) | gotify:// or gotifys:// | (TCP) 80 or 443 | gotify://hostname/token
gotifys://hostname/token?priority=high | [Growl](https://github.com/caronc/apprise/wiki/Notify_growl) | growl:// | (UDP) 23053 | growl://hostname
growl://hostname:portno
growl://password@hostname
growl://password@hostname:port
**Note**: you can also use the get parameter _version_ which can allow the growl request to behave using the older v1.x protocol. An example would look like: growl://hostname?version=1 diff --git a/apprise/plugins/NotifyDiscord.py b/apprise/plugins/NotifyDiscord.py index d64fce96..78dd6319 100644 --- a/apprise/plugins/NotifyDiscord.py +++ b/apprise/plugins/NotifyDiscord.py @@ -414,7 +414,7 @@ class NotifyDiscord(NotifyBase): # 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 - # Gitter server. One would hope we're on NTP and our clocks are + # 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) diff --git a/apprise/plugins/NotifyGitter.py b/apprise/plugins/NotifyGitter.py deleted file mode 100644 index 32e0013c..00000000 --- a/apprise/plugins/NotifyGitter.py +++ /dev/null @@ -1,433 +0,0 @@ -# -*- coding: utf-8 -*- -# BSD 3-Clause License -# -# Apprise - Push Notification Library. -# Copyright (c) 2023, 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. -# -# 3. Neither the name of the copyright holder nor the names of its -# contributors may be used to endorse or promote products derived from -# this software without specific prior written permission. -# -# 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. - -# Once you visit: https://developer.gitter.im/apps you'll get a personal -# access token that will look something like this: -# b5647881d563fm846dfbb2c27d1fe8f669b8f026 - -# Don't worry about generating an app; this token is all you need to form -# you're URL with. The syntax is as follows: -# gitter://{token}/{channel} - -# Hence a URL might look like the following: -# gitter://b5647881d563fm846dfbb2c27d1fe8f669b8f026/apprise - -# Note: You must have joined the channel to send a message to it! - -# Official API reference: https://developer.gitter.im/docs/user-resource - -import re -import requests -from json import loads -from json import dumps -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_list -from ..utils import parse_bool -from ..utils import validate_regex -from ..AppriseLocale import gettext_lazy as _ - -# API Gitter URL -GITTER_API_URL = 'https://api.gitter.im/v1' - -# Used to break path apart into list of targets -TARGET_LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+') - - -class NotifyGitter(NotifyBase): - """ - A wrapper for Gitter Notifications - """ - - # The default descriptive name associated with the Notification - service_name = 'Gitter' - - # The services URL - service_url = 'https://gitter.im/' - - # All notification requests are secure - secure_protocol = 'gitter' - - # A URL that takes you to the setup/help of the specific protocol - setup_url = 'https://github.com/caronc/apprise/wiki/Notify_gitter' - - # Allows the user to specify the NotifyImageSize object - image_size = NotifyImageSize.XY_32 - - # Gitter does not support a title - title_maxlen = 0 - - # Gitter 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 = 0 - - # For Tracking Purposes - ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None) - - # Default to 1 - ratelimit_remaining = 1 - - # Default Notification Format - notify_format = NotifyFormat.MARKDOWN - - # Define object templates - templates = ( - '{schema}://{token}/{targets}/', - ) - - # Define our template tokens - template_tokens = dict(NotifyBase.template_tokens, **{ - 'token': { - 'name': _('Token'), - 'type': 'string', - 'private': True, - 'required': True, - 'regex': (r'^[a-z0-9]{40}$', 'i'), - }, - 'target_room': { - 'name': _('Target Room'), - 'type': 'string', - 'map_to': 'targets', - }, - 'targets': { - 'name': _('Targets'), - 'type': 'list:string', - 'required': True, - }, - }) - - # Define our template arguments - template_args = dict(NotifyBase.template_args, **{ - 'image': { - 'name': _('Include Image'), - 'type': 'bool', - 'default': False, - 'map_to': 'include_image', - }, - 'to': { - 'alias_of': 'targets', - }, - }) - - def __init__(self, token, targets, include_image=False, **kwargs): - """ - Initialize Gitter Object - """ - super().__init__(**kwargs) - - # Secret Key (associated with project) - self.token = validate_regex( - token, *self.template_tokens['token']['regex']) - if not self.token: - msg = 'An invalid Gitter API Token ' \ - '({}) was specified.'.format(token) - self.logger.warning(msg) - raise TypeError(msg) - - # Parse our targets - self.targets = parse_list(targets) - if not self.targets: - msg = 'There are no valid Gitter targets to notify.' - self.logger.warning(msg) - raise TypeError(msg) - - # Used to track maping of rooms to their numeric id lookup for - # messaging - self._room_mapping = None - - # Track whether or not we want to send an image with our notification - # or not. - self.include_image = include_image - - return - - def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): - """ - Perform Gitter Notification - """ - - # error tracking (used for function return) - has_error = False - - # Set up our image for display if configured to do so - image_url = None if not self.include_image \ - else self.image_url(notify_type) - - if image_url: - body = '![alt]({})\n{}'.format(image_url, body) - - if self._room_mapping is None: - # Populate our room mapping - self._room_mapping = {} - postokay, response = self._fetch(url='rooms') - if not postokay: - return False - - # Response generally looks like this: - # [ - # { - # noindex: False, - # oneToOne: False, - # avatarUrl: 'https://path/to/avatar/url', - # url: '/apprise-notifications/community', - # public: True, - # tags: [], - # lurk: False, - # uri: 'apprise-notifications/community', - # lastAccessTime: '2019-03-25T00:12:28.144Z', - # topic: '', - # roomMember: True, - # groupId: '5c981cecd73408ce4fbbad2f', - # githubType: 'REPO_CHANNEL', - # unreadItems: 0, - # mentions: 0, - # security: 'PUBLIC', - # userCount: 1, - # id: '5c981cecd73408ce4fbbad31', - # name: 'apprise/community' - # } - # ] - for entry in response: - self._room_mapping[entry['name'].lower().split('/')[0]] = { - # The ID of the room - 'id': entry['id'], - - # A descriptive name (useful for logging) - 'uri': entry['uri'], - } - - # Create a copy of the targets list - targets = list(self.targets) - while len(targets): - target = targets.pop(0).lower() - - if target not in self._room_mapping: - self.logger.warning( - 'Failed to locate Gitter room {}'.format(target)) - - # Flag our error - has_error = True - continue - - # prepare our payload - payload = { - 'text': body, - } - - # Our Notification URL - notify_url = 'rooms/{}/chatMessages'.format( - self._room_mapping[target]['id']) - - # Perform our query - postokay, response = self._fetch( - notify_url, payload=dumps(payload), method='POST') - - if not postokay: - # Flag our error - has_error = True - - return not has_error - - def _fetch(self, url, payload=None, method='GET'): - """ - Wrapper to request object - - """ - - # Prepare our headers: - headers = { - 'User-Agent': self.app_id, - 'Accept': 'application/json', - 'Authorization': 'Bearer ' + self.token, - } - if payload: - # Only set our header payload if it's defined - headers['Content-Type'] = 'application/json' - - # Default content response object - content = {} - - # Update our URL - url = '{}/{}'.format(GITTER_API_URL, url) - - # Some Debug Logging - self.logger.debug('Gitter {} URL: {} (cert_verify={})'.format( - method, - url, self.verify_certificate)) - if payload: - self.logger.debug('Gitter Payload: {}' .format(payload)) - - # By default set wait to None - wait = None - - if self.ratelimit_remaining <= 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 - # Gitter 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 - # We add 0.5 seconds to the end just to allow a grace - # period. - wait = (self.ratelimit_reset - now).total_seconds() + 0.5 - - # Always call throttle before any remote server i/o is made - self.throttle(wait=wait) - - # fetch function - fn = requests.post if method == 'POST' else requests.get - try: - r = fn( - url, - data=payload, - headers=headers, - verify=self.verify_certificate, - timeout=self.request_timeout, - ) - - if r.status_code != requests.codes.ok: - # We had a problem - status_str = \ - NotifyGitter.http_response_code_lookup(r.status_code) - - self.logger.warning( - 'Failed to send Gitter {} to {}: ' - '{}error={}.'.format( - method, - url, - ', ' if status_str else '', - r.status_code)) - - self.logger.debug( - 'Response Details:\r\n{}'.format(r.content)) - - # Mark our failure - return (False, content) - - try: - content = loads(r.content) - - except (AttributeError, TypeError, ValueError): - # ValueError = r.content is Unparsable - # TypeError = r.content is None - # AttributeError = r is None - content = {} - - try: - self.ratelimit_remaining = \ - int(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 - - except requests.RequestException as e: - self.logger.warning( - 'Exception received when sending Gitter {} to {}: '. - format(method, url)) - self.logger.debug('Socket Exception: %s' % str(e)) - - # Mark our failure - return (False, content) - - return (True, content) - - def url(self, privacy=False, *args, **kwargs): - """ - Returns the URL built dynamically based on specified arguments. - """ - - # Define any URL parameters - params = { - 'image': 'yes' if self.include_image else 'no', - } - - # Extend our parameters - params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) - - return '{schema}://{token}/{targets}/?{params}'.format( - schema=self.secure_protocol, - token=self.pprint(self.token, privacy, safe=''), - targets='/'.join( - [NotifyGitter.quote(x, safe='') for x in self.targets]), - params=NotifyGitter.urlencode(params)) - - def __len__(self): - """ - Returns the number of targets associated with this notification - """ - return len(self.targets) - - @staticmethod - def parse_url(url): - """ - Parses the URL and returns enough arguments that can allow - us to re-instantiate this object. - - """ - results = NotifyBase.parse_url(url, verify_host=False) - if not results: - # We're done early as we couldn't load the results - return results - - results['token'] = NotifyGitter.unquote(results['host']) - - # Get our entries; split_path() looks after unquoting content for us - # by default - results['targets'] = NotifyGitter.split_path(results['fullpath']) - - # Support the 'to' variable so that we can support targets this way too - # The 'to' makes it easier to use yaml configuration - if 'to' in results['qsd'] and len(results['qsd']['to']): - results['targets'] += NotifyGitter.parse_list(results['qsd']['to']) - - # Include images with our message - results['include_image'] = \ - parse_bool(results['qsd'].get('image', False)) - - return results diff --git a/apprise/plugins/NotifyParsePlatform.py b/apprise/plugins/NotifyParsePlatform.py index 69efb61c..4fcb0760 100644 --- a/apprise/plugins/NotifyParsePlatform.py +++ b/apprise/plugins/NotifyParsePlatform.py @@ -30,8 +30,6 @@ # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. -# Official API reference: https://developer.gitter.im/docs/user-resource - import re import requests from json import dumps diff --git a/apprise/plugins/NotifyReddit.py b/apprise/plugins/NotifyReddit.py index 37816a48..69f12737 100644 --- a/apprise/plugins/NotifyReddit.py +++ b/apprise/plugins/NotifyReddit.py @@ -547,7 +547,7 @@ class NotifyReddit(NotifyBase): # 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 - # Gitter server. One would hope we're on NTP and our clocks are + # Reddit 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) diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 3e2f6e4d..27a69f43 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -44,7 +44,7 @@ notification services that are out there. Apprise opens the door and makes it easy to access: Apprise API, AWS SES, AWS SNS, Bark, Boxcar, Burst SMS, BulkSMS, ClickSend, -DAPNET, DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, Gitter, Google Chat, +DAPNET, DingTalk, Discord, E-Mail, Emby, Faast, FCM, Flock, Google Chat, Gotify, Growl, Guilded, Home Assistant, IFTTT, Join, Kavenegar, KODI, Kumulos, LaMetric, Line, MacOSX, Mailgun, Mastodon, Mattermost, Matrix, MessageBird, Microsoft Windows, Microsoft Teams, Misskey, MQTT, MSG91, MyAndroid, Nexmo, diff --git a/test/test_apprise_utils.py b/test/test_apprise_utils.py index 14cb6a5f..ad0789cb 100644 --- a/test/test_apprise_utils.py +++ b/test/test_apprise_utils.py @@ -2705,15 +2705,6 @@ def test_cwe312_url(): 'http://user@localhost?secret=secret-.12345') == \ 'http://user@localhost?secret=s...5' - # Now test other:// private data - assert utils.cwe312_url( - 'gitter://b5637831f563aa846bb5b2c27d8fe8f633b8f026/apprise') == \ - 'gitter://b...6/apprise' - assert utils.cwe312_url( - 'gitter://b5637831f563aa846bb5b2c27d8fe8f633b8f026' - '/apprise/?pass=abc123') == \ - 'gitter://b...6/apprise/?pass=a...3' - assert utils.cwe312_url( 'slack://mybot@xoxb-43598234231-3248932482278-BZK5Wj15B9mPh1RkShJoCZ44' '/lead2gold@gmail.com') == 'slack://mybot@x...4/l...m' diff --git a/test/test_plugin_gitter.py b/test/test_plugin_gitter.py deleted file mode 100644 index 41f560d1..00000000 --- a/test/test_plugin_gitter.py +++ /dev/null @@ -1,293 +0,0 @@ -# -*- coding: utf-8 -*- -# BSD 3-Clause License -# -# Apprise - Push Notification Library. -# Copyright (c) 2023, 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. -# -# 3. Neither the name of the copyright holder nor the names of its -# contributors may be used to endorse or promote products derived from -# this software without specific prior written permission. -# -# 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 pytest -from unittest import mock - -import requests - -from apprise.plugins.NotifyGitter import NotifyGitter -from helpers import AppriseURLTester - -from json import dumps -from datetime import datetime -from datetime import timezone - -# Disable logging for a cleaner testing output -import logging -logging.disable(logging.CRITICAL) - -# Our Testing URLs -apprise_url_tests = ( - ################################## - # NotifyGitter - ################################## - ('gitter://', { - 'instance': TypeError, - }), - ('gitter://:@/', { - 'instance': TypeError, - }), - # Invalid Token Length - ('gitter://%s' % ('a' * 12), { - 'instance': TypeError, - }), - # Token specified but no channel - ('gitter://%s' % ('a' * 40), { - 'instance': TypeError, - }), - # Token + channel - ('gitter://%s/apprise' % ('b' * 40), { - 'instance': NotifyGitter, - 'response': False, - }), - # include image in post - ('gitter://%s/apprise?image=Yes' % ('c' * 40), { - 'instance': NotifyGitter, - 'response': False, - - # Our expected url(privacy=True) startswith() response: - 'privacy_url': 'gitter://c...c/apprise', - }), - # Don't include image in post (this is the default anyway) - ('gitter://%s/apprise?image=Yes' % ('d' * 40), { - 'instance': NotifyGitter, - 'response': False, - # don't include an image by default - 'include_image': False, - }), - # Don't include image in post (this is the default anyway) - ('gitter://%s/apprise?image=No' % ('e' * 40), { - 'instance': NotifyGitter, - 'response': False, - }), - ('gitter://%s/apprise' % ('f' * 40), { - 'instance': NotifyGitter, - # force a failure - 'response': False, - 'requests_response_code': requests.codes.internal_server_error, - }), - ('gitter://%s/apprise' % ('g' * 40), { - 'instance': NotifyGitter, - # throw a bizzare code forcing us to fail to look it up - 'response': False, - 'requests_response_code': 999, - }), - ('gitter://%s/apprise' % ('h' * 40), { - 'instance': NotifyGitter, - # 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_gitter_urls(): - """ - NotifyGitter() Apprise URLs - - """ - - # Run our general tests - AppriseURLTester(tests=apprise_url_tests).run_all() - - -@mock.patch('requests.get') -@mock.patch('requests.post') -def test_plugin_gitter_general(mock_post, mock_get): - """ - NotifyGitter() General Tests - - """ - - # Generate a valid token (40 characters) - token = 'a' * 40 - - response_obj = [ - { - 'noindex': False, - 'oneToOne': False, - 'avatarUrl': 'https://path/to/avatar/url', - 'url': '/apprise-notifications/community', - 'public': True, - 'tags': [], - 'lurk': False, - 'uri': 'apprise-notifications/community', - 'lastAccessTime': '2019-03-25T00:12:28.144Z', - 'topic': '', - 'roomMember': True, - 'groupId': '5c981cecd73408ce4fbbad2f', - 'githubType': 'REPO_CHANNEL', - 'unreadItems': 0, - 'mentions': 0, - 'security': 'PUBLIC', - 'userCount': 1, - 'id': '5c981cecd73408ce4fbbad31', - 'name': 'apprise/community', - }, - ] - - # Epoch time: - epoch = datetime.fromtimestamp(0, timezone.utc) - - request = mock.Mock() - request.content = dumps(response_obj) - request.status_code = requests.codes.ok - request.headers = { - 'X-RateLimit-Reset': ( - datetime.now(timezone.utc) - epoch).total_seconds(), - 'X-RateLimit-Remaining': 1, - } - - # Prepare Mock - mock_get.return_value = request - mock_post.return_value = request - - # Variation Initializations - obj = NotifyGitter(token=token, targets='apprise') - assert isinstance(obj, NotifyGitter) is True - assert isinstance(obj.url(), str) is True - - # apprise room was found - assert obj.send(body="test") is True - - # Change our status code and try again - request.status_code = 403 - assert obj.send(body="test") is False - assert obj.ratelimit_remaining == 1 - - # Return the status - request.status_code = requests.codes.ok - # Force a reset - request.headers['X-RateLimit-Remaining'] = 0 - # 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 - request.headers['X-RateLimit-Remaining'] = 10 - assert obj.send(body="test") is True - assert obj.ratelimit_remaining == 10 - - # Handle cases where we simply couldn't get this field - del request.headers['X-RateLimit-Remaining'] - assert obj.send(body="test") is True - # It remains set to the last value - assert obj.ratelimit_remaining == 10 - - # Reset our variable back to 1 - request.headers['X-RateLimit-Remaining'] = 1 - - # Handle cases where our epoch time is wrong - del request.headers['X-RateLimit-Reset'] - assert obj.send(body="test") is True - - # Return our object, but place it in the future forcing us to block - request.headers['X-RateLimit-Reset'] = \ - (datetime.now(timezone.utc) - epoch).total_seconds() + 1 - request.headers['X-RateLimit-Remaining'] = 0 - obj.ratelimit_remaining = 0 - assert obj.send(body="test") is True - - # Return our object, but place it in the future forcing us to block - request.headers['X-RateLimit-Reset'] = \ - (datetime.now(timezone.utc) - epoch).total_seconds() - 1 - request.headers['X-RateLimit-Remaining'] = 0 - obj.ratelimit_remaining = 0 - assert obj.send(body="test") is True - - # Return our limits to always work - request.headers['X-RateLimit-Reset'] = \ - (datetime.now(timezone.utc) - epoch).total_seconds() - request.headers['X-RateLimit-Remaining'] = 1 - obj.ratelimit_remaining = 1 - - # Cause content response to be None - request.content = None - assert obj.send(body="test") is True - - # Invalid JSON - request.content = '{' - assert obj.send(body="test") is True - - # Return it to a parseable string - request.content = '{}' - - # Support the 'to' as a target - results = NotifyGitter.parse_url( - 'gitter://{}?to={}'.format(token, 'apprise')) - assert isinstance(results, dict) is True - assert 'apprise' in results['targets'] - - # cause a json parsing issue now - response_obj = None - assert obj.send(body="test") is True - - response_obj = '{' - assert obj.send(body="test") is True - - # Variation Initializations - obj = NotifyGitter(token=token, targets='apprise') - assert isinstance(obj, NotifyGitter) is True - assert isinstance(obj.url(), str) is True - # apprise room was not found - assert obj.send(body="test") is False - - # Test exception handling - mock_post.side_effect = \ - requests.ConnectionError(0, 'requests.ConnectionError()') - - # Create temporary _room_mapping object so we will find the apprise - # channel on our second call to send() - obj._room_mapping = { - 'apprise': { - 'id': '5c981cecd73408ce4fbbad31', - 'uri': 'apprise-notifications/community', - } - } - assert obj.send(body='test body', title='test title') is False - - -def test_plugin_gitter_edge_cases(): - """ - NotifyGitter() Edge Cases - - """ - # Define our channels - targets = ['apprise'] - - # Initializes the plugin with an invalid token - with pytest.raises(TypeError): - NotifyGitter(token=None, targets=targets) - # Whitespace also acts as an invalid token value - with pytest.raises(TypeError): - NotifyGitter(token=" ", targets=targets)