From b28cd4cdff1a6a09101cc235366cbcbe5238225a Mon Sep 17 00:00:00 2001 From: Joey Espinosa Date: Sun, 3 Apr 2022 22:00:44 -0400 Subject: [PATCH] Added ntfy support (#524) --- README.md | 1 + apprise/plugins/NotifyNtfy.py | 594 +++++++++++++++++++++++++++ packaging/redhat/python-apprise.spec | 10 +- setup.py | 2 +- test/test_plugin_ntfy.py | 305 ++++++++++++++ 5 files changed, 906 insertions(+), 6 deletions(-) create mode 100644 apprise/plugins/NotifyNtfy.py create mode 100644 test/test_plugin_ntfy.py diff --git a/README.md b/README.md index 8170e08c..b7f8bbd0 100644 --- a/README.md +++ b/README.md @@ -62,6 +62,7 @@ The table below identifies the services this tool supports and some example serv | [NextcloudTalk](https://github.com/caronc/apprise/wiki/Notify_nextcloudtalk) | nctalk:// or nctalks:// | (TCP) 80 or 443 | nctalk://user:pass@host/RoomId
nctalks://user:pass@host/RoomId1/RoomId2/RoomIdN | [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/ +| [ntfy](https://github.com/caronc/apprise/wiki/Notify_ntfy) | ntfy:// | (TCP) 80 or 443 | ntfy://topic/
ntfys://topic/ | [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 | [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 diff --git a/apprise/plugins/NotifyNtfy.py b/apprise/plugins/NotifyNtfy.py new file mode 100644 index 00000000..c691612a --- /dev/null +++ b/apprise/plugins/NotifyNtfy.py @@ -0,0 +1,594 @@ +# MIT License + +# Copyright (c) 2022 Joey Espinosa <@particledecay> + +# 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. +# +# Examples: +# ntfys://my-topic +# ntfy://ntfy.local.domain/my-topic +# ntfys://ntfy.local.domain:8080/my-topic +# ntfy://ntfy.local.domain/?priority=max +import re +import requests +import six +from json import loads +from os.path import basename + +from .NotifyBase import NotifyBase +from ..common import NotifyType +from ..AppriseLocale import gettext_lazy as _ +from ..utils import parse_list +from ..utils import is_hostname +from ..utils import is_ipaddr +from ..utils import validate_regex +from ..URLBase import PrivacyMode + + +class NtfyMode(object): + """ + Define ntfy Notification Modes + """ + # App posts upstream to the developer API on ntfy's website + CLOUD = "cloud" + + # Running a dedicated private ntfy Server + PRIVATE = "private" + + +NTFY_MODES = ( + NtfyMode.CLOUD, + NtfyMode.PRIVATE, +) + + +class NtfyPriority(object): + """ + Ntfy Priority Definitions + """ + MAX = 'max' + HIGH = 'high' + NORMAL = 'default' + LOW = 'low' + MIN = 'min' + + +NTFY_PRIORITIES = ( + NtfyPriority.MAX, + NtfyPriority.HIGH, + NtfyPriority.NORMAL, + NtfyPriority.LOW, + NtfyPriority.MIN, +) + + +class NotifyNtfy(NotifyBase): + """ + A wrapper for ntfy Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'ntfy' + + # The services URL + service_url = 'https://ntfy.sh/' + + # Insecure protocol (for those self hosted requests) + protocol = 'ntfy' + + # The default protocol + secure_protocol = 'ntfys' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_ntfy' + + # Default upstream/cloud host if none is defined + cloud_notify_url = 'https://ntfy.sh' + + # Message time to live (if remote client isn't around to receive it) + time_to_live = 2419200 + + # if our hostname matches the following we automatically enforce + # cloud mode + __auto_cloud_host = re.compile(r'ntfy\.sh', re.IGNORECASE) + + # Define object templates + templates = ( + '{schema}://{topic}', + '{schema}://{host}/{targets}', + '{schema}://{host}:{port}/{targets}', + '{schema}://{user}@{host}/{targets}', + '{schema}://{user}@{host}:{port}/{targets}', + '{schema}://{user}:{password}@{host}/{targets}', + '{schema}://{user}:{password}@{host}:{port}/{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'host': { + 'name': _('Hostname'), + 'type': 'string', + }, + 'port': { + 'name': _('Port'), + 'type': 'int', + 'min': 1, + 'max': 65535, + }, + 'user': { + 'name': _('Username'), + 'type': 'string', + }, + 'password': { + 'name': _('Password'), + 'type': 'string', + 'private': True, + }, + 'topic': { + 'name': _('Topic'), + 'type': 'string', + 'map_to': 'targets', + 'regex': (r'^[a-z0-9_-]{1,64}$', 'i') + }, + 'targets': { + 'name': _('Targets'), + 'type': 'list:string', + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + 'attach': { + 'name': _('Attach'), + 'type': 'string', + }, + 'filename': { + 'name': _('Attach Filename'), + 'type': 'string', + }, + 'click': { + 'name': _('Click'), + 'type': 'string', + }, + 'delay': { + 'name': _('Delay'), + 'type': 'string', + }, + 'email': { + 'name': _('Email'), + 'type': 'string', + }, + 'priority': { + 'name': _('Priority'), + 'type': 'choice:string', + 'values': NTFY_PRIORITIES, + 'default': NtfyPriority.NORMAL, + }, + 'tags': { + 'name': _('Tags'), + 'type': 'string', + }, + 'mode': { + 'name': _('Mode'), + 'type': 'choice:string', + 'values': NTFY_MODES, + 'default': NtfyMode.PRIVATE, + }, + 'to': { + 'alias_of': 'targets', + }, + }) + + def __init__(self, targets=None, attach=None, filename=None, click=None, + delay=None, email=None, priority=None, tags=None, mode=None, + **kwargs): + """ + Initialize ntfy Object + """ + super(NotifyNtfy, self).__init__(**kwargs) + + # Prepare our mode + self.mode = mode.strip().lower() \ + if isinstance(mode, six.string_types) \ + else self.template_args['mode']['default'] + + if self.mode not in NTFY_MODES: + msg = 'An invalid ntfy Mode ({}) was specified.'.format(mode) + self.logger.warning(msg) + raise TypeError(msg) + + # Attach a file (URL supported) + self.attach = attach + + # Our filename (if defined) + self.filename = filename + + # A clickthrough option for notifications + self.click = click + + # Time delay for notifications (various string formats) + self.delay = delay + + # An email to forward notifications to + self.email = email + + # The priority of the message + + if priority is None: + self.priority = self.template_args['priority']['default'] + else: + self.priority = priority + + if self.priority not in NTFY_PRIORITIES: + msg = 'An invalid ntfy Priority ({}) was specified.'.format( + priority) + self.logger.warning(msg) + raise TypeError(msg) + + # Any optional tags to attach to the notification + self.__tags = parse_list(tags) + + # Build list of topics + topics = parse_list(targets) + self.topics = [] + for _topic in topics: + topic = validate_regex( + _topic, *self.template_tokens['topic']['regex']) + if not topic: + self.logger.warning( + 'A specified ntfy topic ({}) is invalid and will be ' + 'ignored'.format(_topic)) + continue + self.topics.append(topic) + return + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform ntfy Notification + """ + + # error tracking (used for function return) + has_error = False + + if not len(self.topics): + # We have nothing to notify; we're done + self.logger.warning('There are no ntfy topics to notify') + return False + + # Prepare our headers + headers = { + 'User-Agent': self.app_id, + } + + if self.priority != NtfyPriority.NORMAL: + headers['X-Priority'] = self.priority + + if title: + headers['X-Title'] = title + + if self.attach is not None: + headers['X-Attach'] = self.attach + if self.filename is not None: + headers['X-Filename'] = self.filename + + if self.click is not None: + headers['X-Click'] = self.click + + if self.delay is not None: + headers['X-Delay'] = self.delay + + if self.email is not None: + headers['X-Email'] = self.email + + if self.__tags: + headers['X-Tags'] = ",".join(self.__tags) + + # Prepare our payload + payload = body + + auth = None + if self.mode == NtfyMode.CLOUD: + # Cloud Service + template_url = self.cloud_notify_url + + else: # NotifyNtfy.PRVATE + # Allow more settings to be applied now + if self.user: + auth = (self.user, self.password) + + # Prepare our ntfy Template URL + schema = 'https' if self.secure else 'http' + + template_url = '%s://%s' % (schema, self.host) + if isinstance(self.port, int): + template_url += ':%d' % self.port + + template_url += '/{topic}' + + # Create a copy of the subreddits list + topics = list(self.topics) + while len(topics) > 0: + # Retrieve our topic + topic = topics.pop() + + # Create our Posting URL per topic provided + url = template_url.format(topic=topic) + self.logger.debug('ntfy POST URL: %s (cert_verify=%r)' % ( + url, self.verify_certificate, + )) + self.logger.debug('ntfy Payload: %s' % str(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.post( + url, + data=payload, + headers=headers, + auth=auth, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + if r.status_code != requests.codes.ok: + # We had a problem + status_str = \ + NotifyBase.http_response_code_lookup(r.status_code) + + # set up our status code to use + status_code = r.status_code + + try: + # Update our status response if we can + json_response = loads(r.content) + status_str = json_response.get('error', status_str) + status_code = \ + int(json_response.get('code', status_code)) + + except (AttributeError, TypeError, ValueError): + # ValueError = r.content is Unparsable + # TypeError = r.content is None + # AttributeError = r is None + + # We could not parse JSON response. + # We will just use the status we already have. + pass + + self.logger.warning( + "Failed to send ntfy notification to topic '{}': " + '{}{}error={}.'.format( + topic, + status_str, + ', ' if status_str else '', + status_code)) + + self.logger.debug( + 'Response Details:\r\n{}'.format(r.content)) + + # Mark our failure + has_error = True + + else: + self.logger.info( + "Sent ntfy notification to '{}'.".format(url)) + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending ntfy:%s ' % ( + url) + '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. + """ + + default_port = 443 if self.secure else 80 + + params = { + 'priority': self.priority, + 'mode': self.mode, + } + + if self.attach is not None: + params['attach'] = self.attach + + if self.click is not None: + params['click'] = self.click + + if self.delay is not None: + params['delay'] = self.delay + + if self.email is not None: + params['email'] = self.email + + if self.__tags: + params['tags'] = ','.join(self.__tags) + + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + # Determine Authentication + auth = '' + if self.user and self.password: + auth = '{user}:{password}@'.format( + user=NotifyNtfy.quote(self.user, safe=''), + password=self.pprint( + self.password, privacy, mode=PrivacyMode.Secret, safe=''), + ) + elif self.user: + auth = '{user}@'.format( + user=NotifyNtfy.quote(self.user, safe=''), + ) + + if self.mode == NtfyMode.PRIVATE: + return '{schema}://{auth}{host}{port}/{targets}?{params}'.format( + schema=self.secure_protocol if self.secure else self.protocol, + auth=auth, + host=self.host, + port='' if self.port is None or self.port == default_port + else ':{}'.format(self.port), + targets='/'.join( + [NotifyNtfy.quote(x, safe='') for x in self.topics]), + params=NotifyNtfy.urlencode(params) + ) + + else: # Cloud mode + return '{schema}://{targets}?{params}'.format( + schema=self.secure_protocol, + targets='/'.join( + [NotifyNtfy.quote(x, safe='') for x in self.topics]), + params=NotifyNtfy.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 + + if 'priority' in results['qsd'] and len(results['qsd']['priority']): + _map = { + # Supported lookups + 'mi': NtfyPriority.MIN, + '1': NtfyPriority.MIN, + 'l': NtfyPriority.LOW, + '2': NtfyPriority.LOW, + 'n': NtfyPriority.NORMAL, # support normal keyword + 'd': NtfyPriority.NORMAL, # default keyword + '3': NtfyPriority.NORMAL, + 'h': NtfyPriority.HIGH, + '4': NtfyPriority.HIGH, + 'ma': NtfyPriority.MAX, + '5': NtfyPriority.MAX, + } + try: + # pretty-format (and update short-format) + results['priority'] = \ + _map[results['qsd']['priority'][0:2].lower()] + + except KeyError: + # Pass along what was set so it can be handed during + # initialization + results['priority'] = str(results['qsd']['priority']) + pass + + if 'attach' in results['qsd'] and len(results['qsd']['attach']): + results['attach'] = NotifyNtfy.unquote(results['qsd']['attach']) + _results = NotifyBase.parse_url(results['attach']) + if _results: + results['filename'] = \ + None if _results['fullpath'] \ + else basename(_results['fullpath']) + + if 'filename' in results['qsd'] and \ + len(results['qsd']['filename']): + results['filename'] = \ + basename(NotifyNtfy.unquote(results['qsd']['filename'])) + + if 'click' in results['qsd'] and len(results['qsd']['click']): + results['click'] = NotifyNtfy.unquote(results['qsd']['click']) + + if 'delay' in results['qsd'] and len(results['qsd']['delay']): + results['delay'] = NotifyNtfy.unquote(results['qsd']['delay']) + + if 'email' in results['qsd'] and len(results['qsd']['email']): + results['email'] = NotifyNtfy.unquote(results['qsd']['email']) + + if 'tags' in results['qsd'] and len(results['qsd']['tags']): + results['tags'] = \ + parse_list(NotifyNtfy.unquote(results['qsd']['tags'])) + + # Acquire our targets/topics + results['targets'] = NotifyNtfy.split_path(results['fullpath']) + + # The 'to' makes it easier to use yaml configuration + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'] += \ + NotifyNtfy.parse_list(results['qsd']['to']) + + # Mode override + if 'mode' in results['qsd'] and results['qsd']['mode']: + results['mode'] = NotifyNtfy.unquote( + results['qsd']['mode'].strip().lower()) + + else: + # We can try to detect the mode based on the validity of the + # hostname. + # + # This isn't a surfire way to do things though; it's best to + # specify the mode= flag + results['mode'] = NtfyMode.PRIVATE \ + if ((is_hostname(results['host']) + or is_ipaddr(results['host'])) and results['targets']) \ + else NtfyMode.CLOUD + + if results['mode'] == NtfyMode.CLOUD: + # Store first entry as it can be a topic too in this case + # But only if we also rule it out not being the words + # ntfy.sh itself, something that starts wiht an non-alpha numeric + # character: + if not NotifyNtfy.__auto_cloud_host.search(results['host']): + # Add it to the front of the list for consistency + results['targets'].insert(0, results['host']) + + elif results['mode'] == NtfyMode.PRIVATE and \ + not (is_hostname(results['host'] or + is_ipaddr(results['host']))): + # Invalid Host for NtfyMode.PRIVATE + return None + + return results + + @staticmethod + def parse_native_url(url): + """ + Support https://ntfy.sh/topic + """ + + # Quick lookup for users who want to just paste + # the ntfy.sh url directly into Apprise + result = re.match( + r'^(http|ntfy)s?://ntfy\.sh' + r'(?P/[^?]+)?' + r'(?P\?.+)?$', url, re.I) + + if result: + mode = 'mode=%s' % NtfyMode.CLOUD + return NotifyNtfy.parse_url( + '{schema}://{topics}{params}'.format( + schema=NotifyNtfy.secure_protocol, + topics=result.group('topics') + if result.group('topics') else '', + params='?%s' % mode + if not result.group('params') + else result.group('params') + '&%s' % mode)) + + return None diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index 517df92e..57cb0b44 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -51,11 +51,11 @@ Apprise API, AWS SES, AWS SNS, Boxcar, ClickSend, DAPNET, DingTalk, Discord, E-M Emby, Faast, FCM, Flock, Gitter, Google Chat, Gotify, Growl, Home Assistant, IFTTT, Join, Kavenegar, KODI, Kumulos, LaMetric, MacOSX, Mailgun, Mattermost, Matrix, Microsoft Windows, Microsoft Teams, MessageBird, MQTT, MSG91, MyAndroid, -Nexmo, Nextcloud, NextcloudTalk, Notica, Notifico, Office365, OneSignal, Opsgenie, -ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, Pushover, -PushSafer, Reddit, Rocket.Chat, SendGrid, ServerChan, SimplePush, Sinch, Slack, -SMTP2Go, Spontit, SparkPost, Super Toasty, Streamlabs, Stride, Syslog, Techulus Push, -Telegram, Twilio, Twitter, Twist, XBMC, XMPP, Webex Teams} +Nexmo, Nextcloud, NextcloudTalk, Notica, Notifico, ntfy, Office365, OneSignal, +Opsgenie, ParsePlatform, PopcornNotify, Prowl, Pushalot, PushBullet, Pushjet, +Pushover, PushSafer, Reddit, Rocket.Chat, SendGrid, ServerChan, SimplePush, Sinch, +Slack, SMTP2Go, Spontit, SparkPost, Super Toasty, Streamlabs, Stride, Syslog, +Techulus Push, Telegram, Twilio, Twitter, Twist, XBMC, XMPP, Webex Teams} Name: python-%{pypi_name} Version: 0.9.7 diff --git a/setup.py b/setup.py index 568441db..9b30e921 100755 --- a/setup.py +++ b/setup.py @@ -73,7 +73,7 @@ setup( 'DAPNET Dingtalk Discord Dbus Emby Faast FCM Flock Gitter Gnome ' 'Google Chat Gotify Growl Home Assistant IFTTT Join Kavenegar KODI ' 'Kumulos LaMetric MacOS Mailgun Matrix Mattermost MessageBird MQTT ' - 'MSG91 Nexmo Nextcloud NextcloudTalk Notica Notifico Office365 ' + 'MSG91 Nexmo Nextcloud NextcloudTalk Notica Notifico Ntfy Office365 ' 'OneSignal Opsgenie ParsePlatform PopcornNotify Prowl PushBullet ' 'Pushjet Pushed Pushover PushSafer Reddit Rocket.Chat Ryver SendGrid ' 'ServerChan SimplePush Sinch Slack SMTP2Go SparkPost Spontit ' diff --git a/test/test_plugin_ntfy.py b/test/test_plugin_ntfy.py new file mode 100644 index 00000000..c5ed53c8 --- /dev/null +++ b/test/test_plugin_ntfy.py @@ -0,0 +1,305 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2022 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. +import json +import mock +import requests +from apprise import plugins +from apprise import NotifyType +from helpers import AppriseURLTester + +# Disable logging for a cleaner testing output +import logging +logging.disable(logging.CRITICAL) + +# For testing our return response +GOOD_RESPONSE_TEXT = { + 'code': '0', + 'error': 'success', +} + +# Our Testing URLs +apprise_url_tests = ( + ('ntfy://', { + # Initializes okay (as cloud mode) but has no topics to notify + 'instance': plugins.NotifyNtfy, + # invalid topics specified (nothing to notify) + # as a result the response type will be false + 'requests_response_text': GOOD_RESPONSE_TEXT, + 'response': False, + }), + ('ntfys://', { + # Initializes okay (as cloud mode) but has no topics to notify + 'instance': plugins.NotifyNtfy, + # invalid topics specified (nothing to notify) + # as a result the response type will be false + 'requests_response_text': GOOD_RESPONSE_TEXT, + 'response': False, + }), + ('ntfy://:@/', { + # Initializes okay (as cloud mode) but has no topics to notify + 'instance': plugins.NotifyNtfy, + # invalid topics specified (nothing to notify) + # as a result the response type will be false + 'requests_response_text': GOOD_RESPONSE_TEXT, + 'response': False, + }), + # No topics + ('ntfy://user:pass@localhost', { + 'instance': plugins.NotifyNtfy, + # invalid topics specified (nothing to notify) + # as a result the response type will be false + 'requests_response_text': GOOD_RESPONSE_TEXT, + 'response': False, + }), + # No valid topics + ('ntfy://user:pass@localhost/#/!/@', { + 'instance': plugins.NotifyNtfy, + # invalid topics specified (nothing to notify) + # as a result the response type will be false + 'requests_response_text': GOOD_RESPONSE_TEXT, + 'response': False, + }), + # user/pass combos + ('ntfy://user@localhost/topic/', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Ntfy cloud mode (enforced) + ('ntfy://ntfy.sh/topic1/topic2/', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # No user/pass combo + ('ntfy://localhost/topic1/topic2/', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # A Email Testing + ('ntfy://localhost/topic1/?email=user@gmail.com', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Tags + ('ntfy://localhost/topic1/?tags=tag1,tag2,tag3', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Delay + ('ntfy://localhost/topic1/?delay=3600', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Title + ('ntfy://localhost/topic1/?title=A%20Great%20Title', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Click + ('ntfy://localhost/topic1/?click=yes', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Email + ('ntfy://localhost/topic1/?email=user@example.com', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Attach + ('ntfy://localhost/topic1/?attach=http://example.com/file.jpg', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Attach with filename over-ride + ('ntfy://localhost/topic1/' + '?attach=http://example.com/file.jpg&filename=smoke.jpg', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT}), + # Attach with bad url + ('ntfy://localhost/topic1/?attach=http://-%20', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Priority + ('ntfy://localhost/topic1/?priority=default', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Priority higher + ('ntfy://localhost/topic1/?priority=high', { + 'instance': plugins.NotifyNtfy, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Invalid Priority + ('ntfy://localhost/topic1/?priority=invalid', { + 'instance': TypeError, + }), + # A topic and port identifier + ('ntfy://user:pass@localhost:8080/topic/', { + 'instance': plugins.NotifyNtfy, + # The response text is expected to be the following on a success + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # A topic (using the to=) + ('ntfys://user:pass@localhost?to=topic', { + 'instance': plugins.NotifyNtfy, + # The response text is expected to be the following on a success + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + ('https://just/a/random/host/that/means/nothing', { + # Nothing transpires from this + 'instance': None + }), + # reference the ntfy.sh url + ('https://ntfy.sh?to=topic', { + 'instance': plugins.NotifyNtfy, + # The response text is expected to be the following on a success + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Several topics + ('ntfy://user:pass@topic1/topic2/topic3/?mode=cloud', { + 'instance': plugins.NotifyNtfy, + # The response text is expected to be the following on a success + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + # Several topics (but do not add ntfy.sh) + ('ntfy://user:pass@ntfy.sh/topic1/topic2/?mode=cloud', { + 'instance': plugins.NotifyNtfy, + # The response text is expected to be the following on a success + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + ('ntfys://user:web/token@localhost/topic/?mode=invalid', { + # Invalid mode + 'instance': TypeError, + }), + # Invalid hostname on localhost/private mode + ('ntfys://user:web@-_/topic1/topic2/?mode=private', { + 'instance': None, + }), + ('ntfy://user:pass@localhost:8081/topic/topic2', { + 'instance': plugins.NotifyNtfy, + # force a failure using basic mode + 'response': False, + 'requests_response_code': requests.codes.internal_server_error, + }), + ('ntfy://user:pass@localhost:8082/topic', { + 'instance': plugins.NotifyNtfy, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), + ('ntfy://user:pass@localhost:8083/topic1/topic2/', { + 'instance': plugins.NotifyNtfy, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracfully handle them + 'test_requests_exceptions': True, + 'requests_response_text': GOOD_RESPONSE_TEXT, + }), +) + + +def test_plugin_ntfy_chat_urls(): + """ + NotifyNtfy() Apprise URLs + + """ + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +def test_plugin_custom_ntfy_edge_cases(mock_post): + """ + NotifyNtfy() Edge Cases + + """ + # Disable Throttling to speed testing + plugins.NotifyBase.request_rate_per_sec = 0 + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + response.content = json.dumps(GOOD_RESPONSE_TEXT) + + # Prepare Mock + mock_post.return_value = response + + results = plugins.NotifyNtfy.parse_url( + 'ntfys://abc---,topic2,~~,,?priority=max&tags=smile,de') + + assert isinstance(results, dict) + assert results['user'] is None + assert results['password'] is None + assert results['port'] is None + assert results['host'] == 'abc---,topic2,~~,,' + assert results['fullpath'] is None + assert results['path'] is None + assert results['query'] is None + assert results['schema'] == 'ntfys' + assert results['url'] == 'ntfys://abc---,topic2,~~,,' + assert isinstance(results['qsd:'], dict) is True + assert results['qsd']['priority'] == 'max' + assert results['qsd']['tags'] == 'smile,de' + + instance = plugins.NotifyNtfy(**results) + assert isinstance(instance, plugins.NotifyNtfy) + assert len(instance.topics) == 2 + assert 'abc---' in instance.topics + assert 'topic2' in instance.topics + + results = plugins.NotifyNtfy.parse_url( + 'ntfy://localhost/topic1/' + '?attach=http://example.com/file.jpg&filename=smoke.jpg') + + assert isinstance(results, dict) + assert results['user'] is None + assert results['password'] is None + assert results['port'] is None + assert results['host'] == 'localhost' + assert results['fullpath'] == '/topic1' + assert results['path'] == '/' + assert results['query'] == 'topic1' + assert results['schema'] == 'ntfy' + assert results['url'] == 'ntfy://localhost/topic1' + assert results['attach'] == 'http://example.com/file.jpg' + assert results['filename'] == 'smoke.jpg' + + instance = plugins.NotifyNtfy(**results) + assert isinstance(instance, plugins.NotifyNtfy) + assert len(instance.topics) == 1 + assert 'topic1' in instance.topics + + assert instance.notify( + body='body', title='title', notify_type=NotifyType.INFO) is True + + # Test our call count + assert mock_post.call_count == 1 + assert mock_post.call_args_list[0][0][0] == \ + 'http://localhost/topic1' + assert mock_post.call_args_list[0][1]['headers'].get('X-Attach') == \ + 'http://example.com/file.jpg' + assert mock_post.call_args_list[0][1]['headers'].get('X-Filename') == \ + 'smoke.jpg'