From e997c11382730b5b8b563047900f05b7c5bbae56 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Wed, 30 Dec 2020 15:55:16 -0500 Subject: [PATCH] Added Opsgenie Support (#337) --- README.md | 3 +- apprise/plugins/NotifyOpsgenie.py | 601 +++++++++++++++++++++++++++ apprise/utils.py | 39 +- packaging/redhat/python-apprise.spec | 5 +- setup.py | 9 +- test/test_rest_plugins.py | 96 ++++- 6 files changed, 742 insertions(+), 11 deletions(-) create mode 100644 apprise/plugins/NotifyOpsgenie.py diff --git a/README.md b/README.md index d6d1e95e..4b262312 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,8 @@ The table below identifies the services this tool supports and some example serv | [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/ | [Office 365](https://github.com/caronc/apprise/wiki/Notify_office365) | o365:// | (TCP) 443 | o365://TenantID:AccountEmail/ClientID/ClientSecret
o365://TenantID:AccountEmail/ClientID/ClientSecret/TargetEmail
o365://TenantID:AccountEmail/ClientID/ClientSecret/TargetEmail1/TargetEmail2/TargetEmailN -| [OneSignal](https://github.com/caronc/apprise/wiki/Notify_onesignal) | onesignal:// | (TCP) 443 | onesignal://AppID@APIKey/PlayerID
onesignal://TemplateID:AppID@APIKey/UserID
onesignal://AppID@APIKey/IncludeSegment
onesignal://AppID@APIKey/Email +| [OneSignal](https://github.com/caronc/apprise/wiki/Notify_onesignal) | onesignal:// | (TCP) 443 | onesignal://AppID@APIKey/PlayerID
onesignal://TemplateID:AppID@APIKey/UserID
onesignal://AppID@APIKey/#IncludeSegment
onesignal://AppID@APIKey/Email +| [Opsgenie](https://github.com/caronc/apprise/wiki/Notify_opsgenie) | opsgenie:// | (TCP) 443 | opsgenie://APIKey
opsgenie://APIKey/UserID
opsgenie://APIKey/#Team
opsgenie://APIKey/\*Schedule
opsgenie://APIKey/^Escalation | [ParsePlatform](https://github.com/caronc/apprise/wiki/Notify_parseplatform) | parsep:// or parseps:// | (TCP) 80 or 443 | parsep://AppID:MasterKey@Hostname
parseps://AppID:MasterKey@Hostname | [PopcornNotify](https://github.com/caronc/apprise/wiki/Notify_popcornnotify) | popcorn:// | (TCP) 443 | popcorn://ApiKey/ToPhoneNo
popcorn://ApiKey/ToPhoneNo1/ToPhoneNo2/ToPhoneNoN/
popcorn://ApiKey/ToEmail
popcorn://ApiKey/ToEmail1/ToEmail2/ToEmailN/
popcorn://ApiKey/ToPhoneNo1/ToEmail1/ToPhoneNoN/ToEmailN | [Prowl](https://github.com/caronc/apprise/wiki/Notify_prowl) | prowl:// | (TCP) 443 | prowl://apikey
prowl://apikey/providerkey diff --git a/apprise/plugins/NotifyOpsgenie.py b/apprise/plugins/NotifyOpsgenie.py new file mode 100644 index 00000000..da63a1d8 --- /dev/null +++ b/apprise/plugins/NotifyOpsgenie.py @@ -0,0 +1,601 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2020 Chris Caron +# All rights reserved. +# +# This code is licensed under the MIT License. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files(the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions : +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +# Signup @ https://www.opsgenie.com +# +# Generate your Integration API Key +# https://app.opsgenie.com/settings/integration/add/API/ + +# Knowing this, you can build your Opsgenie URL as follows: +# opsgenie://{apikey}/ +# opsgenie://{apikey}/@{user} +# opsgenie://{apikey}/*{schedule} +# opsgenie://{apikey}/^{escalation} +# opsgenie://{apikey}/#{team} +# +# You can mix and match what you want to notify freely +# opsgenie://{apikey}/@{user}/#{team}/*{schedule}/^{escalation} +# +# If no target prefix is specified, then it is assumed to be a user. +# +# API Documentation: https://docs.opsgenie.com/docs/alert-api +# API Integration Docs: https://docs.opsgenie.com/docs/api-integration + +import requests +from json import dumps + +from .NotifyBase import NotifyBase +from ..common import NotifyType +from ..utils import validate_regex +from ..utils import is_uuid +from ..utils import parse_list +from ..utils import parse_bool +from ..AppriseLocale import gettext_lazy as _ + + +class OpsgenieCategory(NotifyBase): + """ + We define the different category types that we can notify + """ + USER = 'user' + SCHEDULE = 'schedule' + ESCALATION = 'escalation' + TEAM = 'team' + + +OPSGENIE_CATEGORIES = ( + OpsgenieCategory.USER, + OpsgenieCategory.SCHEDULE, + OpsgenieCategory.ESCALATION, + OpsgenieCategory.TEAM, +) + + +# Regions +class OpsgenieRegion(object): + US = 'us' + EU = 'eu' + + +# Opsgenie APIs +OPSGENIE_API_LOOKUP = { + OpsgenieRegion.US: 'https://api.opsgenie.com/v2/alerts', + OpsgenieRegion.EU: 'https://api.eu.opsgenie.com/v2/alerts', +} + +# A List of our regions we can use for verification +OPSGENIE_REGIONS = ( + OpsgenieRegion.US, + OpsgenieRegion.EU, +) + + +# Priorities +class OpsgeniePriority(object): + LOW = 1 + MODERATE = 2 + NORMAL = 3 + HIGH = 4 + EMERGENCY = 5 + + +OPSGENIE_PRIORITIES = ( + OpsgeniePriority.LOW, + OpsgeniePriority.MODERATE, + OpsgeniePriority.NORMAL, + OpsgeniePriority.HIGH, + OpsgeniePriority.EMERGENCY, +) + + +class NotifyOpsgenie(NotifyBase): + """ + A wrapper for Opsgenie Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'Opsgenie' + + # The services URL + service_url = 'https://opsgenie.com/' + + # All notification requests are secure + secure_protocol = 'opsgenie' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_opsgenie' + + # The maximum length of the body + body_maxlen = 15000 + + # If we don't have the specified min length, then we don't bother using + # the body directive + opsgenie_body_minlen = 130 + + # The default region to use if one isn't otherwise specified + opsgenie_default_region = OpsgenieRegion.US + + # The maximum allowable targets within a notification + maximum_batch_size = 50 + + # Define object templates + templates = ( + '{schema}://{apikey}', + '{schema}://{apikey}/{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'apikey': { + 'name': _('API Key'), + 'type': 'string', + 'private': True, + 'required': True, + }, + 'target_escalation': { + 'name': _('Target Escalation'), + 'prefix': '^', + 'type': 'string', + 'map_to': 'targets', + }, + 'target_schedule': { + 'name': _('Target Schedule'), + 'type': 'string', + 'prefix': '*', + 'map_to': 'targets', + }, + 'target_user': { + 'name': _('Target User'), + 'type': 'string', + 'prefix': '@', + 'map_to': 'targets', + }, + 'target_team': { + 'name': _('Target Team'), + 'type': 'string', + 'prefix': '#', + 'map_to': 'targets', + }, + 'targets': { + 'name': _('Targets '), + 'type': 'list:string', + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + 'region': { + 'name': _('Region Name'), + 'type': 'choice:string', + 'values': OPSGENIE_REGIONS, + 'default': OpsgenieRegion.US, + 'map_to': 'region_name', + }, + 'batch': { + 'name': _('Batch Mode'), + 'type': 'bool', + 'default': False, + }, + 'priority': { + 'name': _('Priority'), + 'type': 'choice:int', + 'values': OPSGENIE_PRIORITIES, + 'default': OpsgeniePriority.NORMAL, + }, + 'entity': { + 'name': _('Entity'), + 'type': 'string', + }, + 'alias': { + 'name': _('Alias'), + 'type': 'string', + }, + 'tags': { + 'name': _('Tags'), + 'type': 'string', + }, + 'to': { + 'alias_of': 'targets', + }, + }) + + # Map of key-value pairs to use as custom properties of the alert. + template_kwargs = { + 'details': { + 'name': _('Details'), + 'prefix': '+', + }, + } + + def __init__(self, apikey, targets, region_name=None, details=None, + priority=None, alias=None, entity=None, batch=False, + tags=None, **kwargs): + """ + Initialize Opsgenie Object + """ + super(NotifyOpsgenie, self).__init__(**kwargs) + + # API Key (associated with project) + self.apikey = validate_regex(apikey) + if not self.apikey: + msg = 'An invalid Opsgenie API Key ' \ + '({}) was specified.'.format(apikey) + self.logger.warning(msg) + raise TypeError(msg) + + # The Priority of the message + if priority not in OPSGENIE_PRIORITIES: + self.priority = OpsgeniePriority.NORMAL + + else: + self.priority = priority + + # Store our region + try: + self.region_name = self.opsgenie_default_region \ + if region_name is None else region_name.lower() + + if self.region_name not in OPSGENIE_REGIONS: + # allow the outer except to handle this common response + raise + except: + # Invalid region specified + msg = 'The Opsgenie region specified ({}) is invalid.' \ + .format(region_name) + self.logger.warning(msg) + raise TypeError(msg) + + self.details = {} + if details: + # Store our extra details + self.details.update(details) + + # Prepare Batch Mode Flag + self.batch_size = self.maximum_batch_size if batch else 1 + + # Assign our tags (if defined) + self.__tags = parse_list(tags) + + # Assign our entity (if defined) + self.entity = entity + + # Assign our alias (if defined) + self.alias = alias + + # Initialize our Targets + self.targets = [] + + # Sort our targets + for _target in parse_list(targets): + target = _target.strip() + if len(target) < 2: + self.logger.debug('Ignoring Opsgenie Entry: %s' % target) + continue + + if target.startswith(NotifyOpsgenie.template_tokens + ['target_team']['prefix']): + + self.targets.append( + {'type': OpsgenieCategory.TEAM, 'id': target[1:]} + if is_uuid(target[1:]) else + {'type': OpsgenieCategory.TEAM, 'name': target[1:]}) + + elif target.startswith(NotifyOpsgenie.template_tokens + ['target_schedule']['prefix']): + + self.targets.append( + {'type': OpsgenieCategory.SCHEDULE, 'id': target[1:]} + if is_uuid(target[1:]) else + {'type': OpsgenieCategory.SCHEDULE, 'name': target[1:]}) + + elif target.startswith(NotifyOpsgenie.template_tokens + ['target_escalation']['prefix']): + + self.targets.append( + {'type': OpsgenieCategory.ESCALATION, 'id': target[1:]} + if is_uuid(target[1:]) else + {'type': OpsgenieCategory.ESCALATION, 'name': target[1:]}) + + elif target.startswith(NotifyOpsgenie.template_tokens + ['target_user']['prefix']): + + self.targets.append( + {'type': OpsgenieCategory.USER, 'id': target[1:]} + if is_uuid(target[1:]) else + {'type': OpsgenieCategory.USER, 'username': target[1:]}) + + else: + # Ambiguious entry; treat it as a user but not before + # displaying a warning to the end user first: + self.logger.debug( + 'Treating ambigious Opsgenie target %s as a user', target) + self.targets.append( + {'type': OpsgenieCategory.USER, 'id': target} + if is_uuid(target) else + {'type': OpsgenieCategory.USER, 'username': target}) + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform Opsgenie Notification + """ + + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + 'Authorization': 'GenieKey {}'.format(self.apikey), + } + + # Prepare our URL as it's based on our hostname + notify_url = OPSGENIE_API_LOOKUP[self.region_name] + + # Initialize our has_error flag + has_error = False + + # We want to manually set the title onto the body if specified + title_body = body if not title else '{}: {}'.format(title, body) + + # Create a copy ouf our details object + details = self.details.copy() + if 'type' not in details: + details['type'] = notify_type + + # Prepare our payload + payload = { + 'source': self.app_desc, + 'message': title_body, + 'description': body, + 'details': details, + 'priority': 'P{}'.format(self.priority), + } + + # Use our body directive if we exceed the minimum message + # limitation + if len(payload['message']) > self.opsgenie_body_minlen: + payload['message'] = '{}...'.format( + body[:self.opsgenie_body_minlen - 3]) + + if self.__tags: + payload['tags'] = self.__tags + + if self.entity: + payload['entity'] = self.entity + + if self.alias: + payload['alias'] = self.alias + + length = len(self.targets) if self.targets else 1 + for index in range(0, length, self.batch_size): + if self.targets: + # If there were no targets identified, then we simply + # just iterate once without the responders set + payload['responders'] = \ + self.targets[index:index + self.batch_size] + + # Some Debug Logging + self.logger.debug( + 'Opsgenie POST URL: {} (cert_verify={})'.format( + notify_url, self.verify_certificate)) + self.logger.debug('Opsgenie Payload: {}' .format(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + try: + r = requests.post( + notify_url, + data=dumps(payload), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code not in ( + requests.codes.accepted, requests.codes.ok): + status_str = \ + NotifyBase.http_response_code_lookup( + r.status_code) + + self.logger.warning( + 'Failed to send Opsgenie notification:' + '{}{}error={}.'.format( + status_str, + ', ' if status_str else '', + r.status_code)) + + self.logger.debug( + 'Response Details:\r\n{}'.format(r.content)) + + # Mark our failure + has_error = True + continue + + # If we reach here; the message was sent + self.logger.info('Sent Opsgenie notification') + self.logger.debug( + 'Response Details:\r\n{}'.format(r.content)) + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending Opsgenie ' + 'notification.') + self.logger.debug('Socket Exception: %s' % str(e)) + # Mark our failure + has_error = True + + return not has_error + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + _map = { + OpsgeniePriority.LOW: 'low', + OpsgeniePriority.MODERATE: 'moderate', + OpsgeniePriority.NORMAL: 'normal', + OpsgeniePriority.HIGH: 'high', + OpsgeniePriority.EMERGENCY: 'emergency', + } + + # Define any URL parameters + params = { + 'region': self.region_name, + 'priority': + _map[OpsgeniePriority.NORMAL] if self.priority not in _map + else _map[self.priority], + 'batch': 'yes' if self.batch_size > 1 else 'no', + } + + # Assign our entity value (if defined) + if self.entity: + params['entity'] = self.entity + + # Assign our alias value (if defined) + if self.alias: + params['alias'] = self.alias + + # Assign our tags (if specifed) + if self.__tags: + params['tags'] = ','.join(self.__tags) + + # Append our details into our parameters + params.update({'+{}'.format(k): v for k, v in self.details.items()}) + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + # A map allows us to map our target types so they can be correctly + # placed back into your URL below. Hence map the 'user' -> '@' + __map = { + OpsgenieCategory.USER: + NotifyOpsgenie.template_tokens['target_user']['prefix'], + OpsgenieCategory.SCHEDULE: + NotifyOpsgenie.template_tokens['target_schedule']['prefix'], + OpsgenieCategory.ESCALATION: + NotifyOpsgenie.template_tokens['target_escalation']['prefix'], + OpsgenieCategory.TEAM: + NotifyOpsgenie.template_tokens['target_team']['prefix'], + } + + return '{schema}://{apikey}/{targets}/?{params}'.format( + schema=self.secure_protocol, + apikey=self.pprint(self.apikey, privacy, safe=''), + targets='/'.join( + [NotifyOpsgenie.quote('{}{}'.format( + __map[x['type']], + x.get('id', x.get('name', x.get('username'))))) + for x in self.targets]), + params=NotifyOpsgenie.urlencode(params)) + + @staticmethod + def parse_url(url): + """ + Parses the URL and returns enough arguments that can allow + us to re-instantiate this object. + + """ + results = NotifyBase.parse_url(url, verify_host=False) + if not results: + # We're done early as we couldn't load the results + return results + + # The API Key is stored in the hostname + results['apikey'] = NotifyOpsgenie.unquote(results['host']) + + # Get our Targets + results['targets'] = NotifyOpsgenie.split_path(results['fullpath']) + + # Add our Meta Detail keys + results['details'] = {NotifyBase.unquote(x): NotifyBase.unquote(y) + for x, y in results['qsd+'].items()} + + if 'priority' in results['qsd'] and len(results['qsd']['priority']): + _map = { + # Letter Assignnments + 'l': OpsgeniePriority.LOW, + 'm': OpsgeniePriority.MODERATE, + 'n': OpsgeniePriority.NORMAL, + 'h': OpsgeniePriority.HIGH, + 'e': OpsgeniePriority.EMERGENCY, + 'lo': OpsgeniePriority.LOW, + 'me': OpsgeniePriority.MODERATE, + 'no': OpsgeniePriority.NORMAL, + 'hi': OpsgeniePriority.HIGH, + 'em': OpsgeniePriority.EMERGENCY, + # Support 3rd Party API Documented Scale + '1': OpsgeniePriority.LOW, + '2': OpsgeniePriority.MODERATE, + '3': OpsgeniePriority.NORMAL, + '4': OpsgeniePriority.HIGH, + '5': OpsgeniePriority.EMERGENCY, + 'p1': OpsgeniePriority.LOW, + 'p2': OpsgeniePriority.MODERATE, + 'p3': OpsgeniePriority.NORMAL, + 'p4': OpsgeniePriority.HIGH, + 'p5': OpsgeniePriority.EMERGENCY, + } + try: + results['priority'] = \ + _map[results['qsd']['priority'][0:2].lower()] + + except KeyError: + # No priority was set + pass + + # Get Batch Boolean (if set) + results['batch'] = \ + parse_bool( + results['qsd'].get( + 'batch', + NotifyOpsgenie.template_args['batch']['default'])) + + if 'apikey' in results['qsd'] and len(results['qsd']['apikey']): + results['apikey'] = \ + NotifyOpsgenie.unquote(results['qsd']['apikey']) + + if 'tags' in results['qsd'] and len(results['qsd']['tags']): + # Extract our tags + results['tags'] = \ + parse_list(NotifyOpsgenie.unquote(results['qsd']['tags'])) + + if 'region' in results['qsd'] and len(results['qsd']['region']): + # Extract our region + results['region_name'] = \ + NotifyOpsgenie.unquote(results['qsd']['region']) + + if 'entity' in results['qsd'] and len(results['qsd']['entity']): + # Extract optional entity field + results['entity'] = \ + NotifyOpsgenie.unquote(results['qsd']['entity']) + + if 'alias' in results['qsd'] and len(results['qsd']['alias']): + # Extract optional alias field + results['alias'] = \ + NotifyOpsgenie.unquote(results['qsd']['alias']) + + # Handle 'to' email address + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'].append(results['qsd']['to']) + + return results diff --git a/apprise/utils.py b/apprise/utils.py index 4b4833be..3d5de6f4 100644 --- a/apprise/utils.py +++ b/apprise/utils.py @@ -138,6 +138,11 @@ EMAIL_DETECTION_RE = re.compile( r'[^@\s,]+@[^\s,]+)', re.IGNORECASE) +# Used to prepare our UUID regex matching +UUID4_RE = re.compile( + r'[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}', + re.IGNORECASE) + # validate_regex() utilizes this mapping to track and re-use pre-complied # regular expressions REGEX_VALIDATE_LOOKUP = {} @@ -247,6 +252,27 @@ def is_hostname(hostname, ipv4=True, ipv6=True): return hostname +def is_uuid(uuid): + """Determine if the specified entry is uuid v4 string + + Args: + address (str): The string you want to check. + + Returns: + bool: Returns False if the specified element is not a uuid otherwise + it returns True + """ + + try: + match = UUID4_RE.match(uuid) + + except TypeError: + # not parseable content + return False + + return True if match else False + + def is_email(address): """Determine if the specified entry is an email address @@ -254,8 +280,17 @@ def is_email(address): address (str): The string you want to check. Returns: - bool: Returns True if the address specified is an email address - and False if it isn't. + bool: Returns False if the address specified is not an email address + and a dictionary of the parsed email if it is as: + { + 'name': 'Parse Name' + 'email': 'user@domain.com' + 'full_email': 'label+user@domain.com' + 'label': 'label' + 'user': 'user', + 'domain': 'domain.com' + } + """ try: diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 1e2f5ed8..ed31acc7 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -50,10 +50,11 @@ it easy to access: Boxcar, ClickSend, Discord, E-Mail, Emby, Faast, FCM, Flock, Gitter, Gotify, Growl, IFTTT, Join, Kavenegar, KODI, Kumulos, LaMetric, MacOSX, Mailgun, MatterMost, Matrix, Microsoft Windows, Microsoft Teams, MessageBird, MSG91, -MyAndroid, Nexmo, Nextcloud, Notica, Notifico, Office365, OneSignal, +MyAndroid, Nexmo, Nextcloud, Notica, Notifico, Office365, OneSignal, Opsgenie, ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, Pushover, PushSafer, Rocket.Chat, SendGrid, SimplePush, Sinch, Slack, Spontit, -SparkPost, Super Toasty, Stride, Syslog, Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP, Webex Teams} +SparkPost, Super Toasty, Stride, Syslog, Techulus Push, Telegram, Twilio, +Twitter, Twist, XBMC, XMPP, Webex Teams} Name: python-%{pypi_name} Version: 0.8.9 diff --git a/setup.py b/setup.py index a8a1050c..a315f7aa 100755 --- a/setup.py +++ b/setup.py @@ -73,10 +73,11 @@ setup( 'Discord Dbus Emby Faast FCM Flock Gitter Gnome Gotify Growl IFTTT ' 'Join Kavenegar KODI Kumulos LaMetric MacOS Mailgun Matrix Mattermost ' 'MessageBird MSG91 Nexmo Nextcloud Notica Notifico Office365 ' - 'OneSignal ParsePlatform PopcornNotify Prowl PushBullet Pushjet ' - 'Pushed Pushover PushSafer Rocket.Chat Ryver SendGrid SimplePush ' - 'Sinch Slack SparkPost Spontit Stride Syslog Techulus Telegram ' - 'Twilio Twist Twitter XBMC MSTeams Microsoft Windows Webex CLI API', + 'OneSignal Opsgenie ParsePlatform PopcornNotify Prowl PushBullet ' + 'Pushjet Pushed Pushover PushSafer Rocket.Chat Ryver SendGrid ' + 'SimplePush Sinch Slack SparkPost Spontit Stride Syslog Techulus ' + 'Telegram Twilio Twist Twitter XBMC MSTeams Microsoft Windows Webex ' + 'CLI API', author='Chris Caron', author_email='lead2gold@gmail.com', packages=find_packages(), diff --git a/test/test_rest_plugins.py b/test/test_rest_plugins.py index 534551a7..5393b519 100644 --- a/test/test_rest_plugins.py +++ b/test/test_rest_plugins.py @@ -2383,19 +2383,111 @@ TEST_URLS = ( # Test Kwargs 'instance': plugins.NotifyOneSignal, }), - ('onesignal://appid@apikey/#topic1/device/', { + ('onesignal://appid@apikey/#segment/playerid/', { 'instance': plugins.NotifyOneSignal, # throw a bizzare code forcing us to fail to look it up 'response': False, 'requests_response_code': 999, }), - ('onesignal://appid@apikey/#topic1/device/', { + ('onesignal://appid@apikey/#segment/playerid/', { 'instance': plugins.NotifyOneSignal, # Throws a series of connection and transfer exceptions when this flag # is set and tests that we gracfully handle them 'test_requests_exceptions': True, }), + ################################## + # NotifyOpsgenie + ################################## + ('opsgenie://', { + # We failed to identify any valid authentication + 'instance': TypeError, + }), + ('opsgenie://:@/', { + # We failed to identify any valid authentication + 'instance': TypeError, + }), + ('opsgenie://%20%20/', { + # invalid apikey specified + 'instance': TypeError, + }), + ('opsgenie://apikey/user/?region=xx', { + # invalid region id + 'instance': TypeError, + }), + ('opsgenie://apikey/', { + # No targets specified; this is allowed + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/user', { + # Valid user + 'instance': plugins.NotifyOpsgenie, + 'privacy_url': 'opsgenie://a...y/%40user', + }), + ('opsgenie://apikey/@user?region=eu', { + # European Region + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/@user?entity=A%20Entity', { + # Assign an entity + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/@user?alias=An%20Alias', { + # Assign an alias + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/@user?priority=p3', { + # Assign our priority + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/?tags=comma,separated', { + # Test our our 'tags' (tag is reserved in Apprise) but not 'tags' + # Also test the fact we do not need to define a target + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/@user?priority=invalid', { + # Invalid priority (loads using default) + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/user@email.com/#team/*sche/^esc/%20/a', { + # Valid user (email), valid schedule, Escalated ID, + # an invalid entry (%20), and too short of an entry (a) + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/{}/@{}/#{}/*{}/^{}/'.format( + UUID4, UUID4, UUID4, UUID4, UUID4), { + # similar to the above, except we use the UUID's + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey?to=#team,user&+key=value&+type=override', { + # Test to= and details (key/value pair) also override 'type' + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/#team/@user/?batch=yes', { + # Test batch= + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/#team/@user/?batch=no', { + # Test batch= + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://?apikey=abc&to=user', { + # Test Kwargs + 'instance': plugins.NotifyOpsgenie, + }), + ('opsgenie://apikey/#team/user/', { + 'instance': plugins.NotifyOpsgenie, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + }), + ('opsgenie://apikey/#topic1/device/', { + 'instance': plugins.NotifyOpsgenie, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracfully handle them + 'test_requests_exceptions': True, + }), + ################################## # NotifyParsePlatform ##################################