diff --git a/apprise/plugins/NotifyBase.py b/apprise/plugins/NotifyBase.py index a9ecf20d..8c9116ec 100644 --- a/apprise/plugins/NotifyBase.py +++ b/apprise/plugins/NotifyBase.py @@ -2,7 +2,7 @@ # # Base Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -243,14 +243,18 @@ class NotifyBase(object): return self.asset.app_url @staticmethod - def escape_html(html, convert_new_lines=False): + def escape_html(html, convert_new_lines=False, whitespace=True): """ Takes html text as input and escapes it so that it won't conflict with any xml/html wrapping characters. """ - escaped = _escape(html).\ - replace(u'\t', u' ').\ - replace(u' ', u' ') + escaped = _escape(html) + + if whitespace: + # Tidy up whitespace too + escaped = escaped\ + .replace(u'\t', u' ')\ + .replace(u' ', u' ') if convert_new_lines: return escaped.replace(u'\n', u'<br/>') @@ -335,7 +339,7 @@ class NotifyBase(object): return is_hostname(hostname) @staticmethod - def parse_url(url, verify_host=True, default_format=NotifyFormat.TEXT): + def parse_url(url, verify_host=True): """ Parses the URL and returns it broken apart into a dictionary. @@ -350,9 +354,6 @@ class NotifyBase(object): # if our URL ends with an 's', then assueme our secure flag is set. results['secure'] = (results['schema'][-1] == 's') - # Our default notification format - results['notify_format'] = default_format - # Support SSL Certificate 'verify' keyword. Default to being enabled results['verify'] = verify_host @@ -360,6 +361,15 @@ class NotifyBase(object): results['verify'] = parse_bool( results['qsd'].get('verify', True)) + # Allow overriding the default format + if 'format' in results['qsd']: + results['format'] = results['qsd'].get('format') + if results['format'] not in NOTIFY_FORMATS: + NotifyBase.logger.warning( + 'Unsupported format specified {}'.format( + results['format'])) + del results['format'] + # Password overrides if 'pass' in results['qsd']: results['password'] = results['qsd']['pass'] diff --git a/apprise/plugins/NotifyBoxcar.py b/apprise/plugins/NotifyBoxcar.py index 3589309c..1cc4435c 100644 --- a/apprise/plugins/NotifyBoxcar.py +++ b/apprise/plugins/NotifyBoxcar.py @@ -2,7 +2,7 @@ # # Boxcar Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -240,6 +240,9 @@ class NotifyBoxcar(NotifyBase): # Return; we're done return False + else: + self.logger.info('Sent Boxcar notification.') + except requests.RequestException as e: self.logger.warning( 'A Connection error occured sending Boxcar ' diff --git a/apprise/plugins/NotifyEmail.py b/apprise/plugins/NotifyEmail.py index ffc9068d..5552f9ea 100644 --- a/apprise/plugins/NotifyEmail.py +++ b/apprise/plugins/NotifyEmail.py @@ -2,7 +2,7 @@ # # Email Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -157,7 +157,7 @@ class NotifyEmail(NotifyBase): # Now we want to construct the To and From email # addresses from the URL provided - self.from_name = kwargs.get('name', NotifyBase.app_desc) + self.from_name = kwargs.get('name', self.app_desc) self.from_addr = kwargs.get('from', None) if not NotifyBase.is_email(self.to_addr): diff --git a/apprise/plugins/NotifyGrowl/NotifyGrowl.py b/apprise/plugins/NotifyGrowl/NotifyGrowl.py index 2607eda6..8632f525 100644 --- a/apprise/plugins/NotifyGrowl/NotifyGrowl.py +++ b/apprise/plugins/NotifyGrowl/NotifyGrowl.py @@ -2,7 +2,7 @@ # # Growl Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -173,9 +173,7 @@ class NotifyGrowl(NotifyBase): ) else: - self.logger.debug( - 'Growl notification sent successfully.' - ) + self.logger.info('Sent Growl notification.') except errors.BaseError as e: # Since Growl servers listen for UDP broadcasts, it's possible diff --git a/apprise/plugins/NotifyJSON.py b/apprise/plugins/NotifyJSON.py index 3513e161..4ba71ca4 100644 --- a/apprise/plugins/NotifyJSON.py +++ b/apprise/plugins/NotifyJSON.py @@ -2,7 +2,7 @@ # # JSON Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -118,6 +118,9 @@ class NotifyJSON(NotifyBase): # Return; we're done return False + else: + self.logger.info('Sent JSON notification.') + except requests.RequestException as e: self.logger.warning( 'A Connection error occured sending JSON ' diff --git a/apprise/plugins/NotifyJoin.py b/apprise/plugins/NotifyJoin.py index 6e4562af..20348c4d 100644 --- a/apprise/plugins/NotifyJoin.py +++ b/apprise/plugins/NotifyJoin.py @@ -2,7 +2,7 @@ # # Join Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -196,6 +196,9 @@ class NotifyJoin(NotifyBase): return_status = False + else: + self.logger.info('Sent Join notification to %s.' % device) + except requests.RequestException as e: self.logger.warning( 'A Connection error occured sending Join:%s ' diff --git a/apprise/plugins/NotifyProwl.py b/apprise/plugins/NotifyProwl.py index 68c7e305..e28a06e2 100644 --- a/apprise/plugins/NotifyProwl.py +++ b/apprise/plugins/NotifyProwl.py @@ -155,6 +155,7 @@ class NotifyProwl(NotifyBase): # Return; we're done return False + else: self.logger.info('Sent Prowl notification.') diff --git a/apprise/plugins/NotifyPushjet/NotifyPushjet.py b/apprise/plugins/NotifyPushjet/NotifyPushjet.py index 356da3fb..09baf668 100644 --- a/apprise/plugins/NotifyPushjet/NotifyPushjet.py +++ b/apprise/plugins/NotifyPushjet/NotifyPushjet.py @@ -2,7 +2,7 @@ # # Pushjet Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -67,6 +67,7 @@ class NotifyPushjet(NotifyBase): service = api.Service(secret_key=self.host) service.send(body, title) + self.logger.info('Sent Pushjet notification.') except (errors.PushjetError, ValueError) as e: self.logger.warning('Failed to send Pushjet notification.') diff --git a/apprise/plugins/NotifyPushover.py b/apprise/plugins/NotifyPushover.py index 223cf29b..f5ac59e9 100644 --- a/apprise/plugins/NotifyPushover.py +++ b/apprise/plugins/NotifyPushover.py @@ -2,7 +2,7 @@ # # Pushover Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -201,6 +201,10 @@ class NotifyPushover(NotifyBase): # Return; we're done has_error = True + else: + self.logger.info( + 'Sent Pushover notification to %s.' % device) + except requests.RequestException as e: self.logger.warning( 'A Connection error occured sending Pushover:%s ' % ( diff --git a/apprise/plugins/NotifySlack.py b/apprise/plugins/NotifySlack.py index 1622c132..d5543d81 100644 --- a/apprise/plugins/NotifySlack.py +++ b/apprise/plugins/NotifySlack.py @@ -2,7 +2,7 @@ # # Slack Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -263,6 +263,9 @@ class NotifySlack(NotifyBase): # Return; we're done notify_okay = False + else: + self.logger.info('Sent Slack notification.') + except requests.RequestException as e: self.logger.warning( 'A Connection error occured sending Slack:%s ' % ( diff --git a/apprise/plugins/NotifyTelegram.py b/apprise/plugins/NotifyTelegram.py index b6698434..3b859f8f 100644 --- a/apprise/plugins/NotifyTelegram.py +++ b/apprise/plugins/NotifyTelegram.py @@ -2,7 +2,7 @@ # # Telegram Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -40,17 +40,24 @@ # For example, a url might look like this: # https://api.telegram.org/bot123456789:alphanumeri_characters/getMe # +# Development API Reference:: +# - https://core.telegram.org/bots/api import requests import re +from os.path import basename + from json import loads from json import dumps from .NotifyBase import NotifyBase from .NotifyBase import HTTP_ERROR_MAP -from ..common import NotifyFormat from ..common import NotifyImageSize +from ..common import NotifyFormat from ..utils import compat_is_basestring +from ..utils import parse_bool + +TELEGRAM_IMAGE_XY = NotifyImageSize.XY_256 # Token required as part of the API request # allow the word 'bot' infront @@ -82,15 +89,15 @@ class NotifyTelegram(NotifyBase): # Telegram uses the http protocol with JSON requests notify_url = 'https://api.telegram.org/bot' - def __init__(self, bot_token, chat_ids, notify_format=NotifyFormat.HTML, - **kwargs): + def __init__(self, bot_token, chat_ids, notify_format=NotifyFormat.TEXT, + detect_bot_owner=True, include_image=True, **kwargs): """ Initialize Telegram Object """ super(NotifyTelegram, self).__init__( title_maxlen=250, body_maxlen=4096, notify_format=notify_format, - **kwargs) + image_size=TELEGRAM_IMAGE_XY, **kwargs) try: self.bot_token = bot_token.strip() @@ -124,10 +131,193 @@ class NotifyTelegram(NotifyBase): # Treat this as a channel too self.chat_ids.append(self.user) + if len(self.chat_ids) == 0 and detect_bot_owner: + _id = self.detect_bot_owner() + if _id: + # Store our id + self.chat_ids = [str(_id)] + if len(self.chat_ids) == 0: self.logger.warning('No chat_id(s) were specified.') raise TypeError('No chat_id(s) were specified.') + # Track whether or not we want to send an image with our notification + # or not. + self.include_image = include_image + + def send_image(self, chat_id, notify_type): + """ + Sends a sticker based on the specified notify type + + """ + + # The URL; we do not set headers because the api doesn't seem to like + # when we set one. + url = '%s%s/%s' % ( + self.notify_url, + self.bot_token, + 'sendPhoto' + ) + + path = self.image_path(notify_type) + if not path: + # No image to send + self.logger.debug( + 'Telegram Image does not exist for %s' % ( + notify_type)) + return None + + files = {'photo': (basename(path), open(path), 'rb')} + + payload = { + 'chat_id': chat_id, + } + + self.logger.debug( + 'Telegram Image POST URL: %s (cert_verify=%r)' % ( + url, self.verify_certificate)) + + try: + r = requests.post( + url, + files=files, + data=payload, + verify=self.verify_certificate, + ) + + if r.status_code != requests.codes.ok: + # We had a problem + try: + self.logger.warning( + 'Failed to post Telegram Image: ' + '%s (error=%s).' % ( + HTTP_ERROR_MAP[r.status_code], + r.status_code)) + + except KeyError: + self.logger.warning( + 'Failed to detect Telegram Image. (error=%s).' % ( + r.status_code)) + + # self.logger.debug('Response Details: %s' % r.raw.read()) + return False + + except requests.RequestException as e: + self.logger.warning( + 'A connection error occured posting Telegram Image.') + self.logger.debug('Socket Exception: %s' % str(e)) + return False + + return True + + def detect_bot_owner(self): + """ + Takes a bot and attempts to detect it's chat id from that + + """ + + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + } + + url = '%s%s/%s' % ( + self.notify_url, + self.bot_token, + 'getUpdates' + ) + + self.logger.debug( + 'Telegram User Detection POST URL: %s (cert_verify=%r)' % ( + url, self.verify_certificate)) + + try: + r = requests.post( + url, + headers=headers, + verify=self.verify_certificate, + ) + + if r.status_code != requests.codes.ok: + # We had a problem + + try: + # Try to get the error message if we can: + error_msg = loads(r.content)['description'] + + except: + error_msg = None + + try: + if error_msg: + self.logger.warning( + 'Failed to detect Telegram user: (%s) %s.' % ( + r.status_code, error_msg)) + + else: + self.logger.warning( + 'Failed to detect Telegram user: ' + '%s (error=%s).' % ( + HTTP_ERROR_MAP[r.status_code], + r.status_code)) + + except KeyError: + self.logger.warning( + 'Failed to detect Telegram user. (error=%s).' % ( + r.status_code)) + + # self.logger.debug('Response Details: %s' % r.raw.read()) + return 0 + + except requests.RequestException as e: + self.logger.warning( + 'A connection error occured detecting Telegram User.') + self.logger.debug('Socket Exception: %s' % str(e)) + return 0 + + # A Response might look something like this: + # { + # "ok":true, + # "result":[{ + # "update_id":645421321, + # "message":{ + # "message_id":1, + # "from":{ + # "id":532389719, + # "is_bot":false, + # "first_name":"Chris", + # "language_code":"en-US" + # }, + # "chat":{ + # "id":532389719, + # "first_name":"Chris", + # "type":"private" + # }, + # "date":1519694394, + # "text":"/start", + # "entities":[{"offset":0,"length":6,"type":"bot_command"}]}}] + + # Load our response and attempt to fetch our userid + response = loads(r.content) + if 'ok' in response and response['ok'] is True: + start = re.compile('^\s*\/start', re.I) + for _msg in iter(response['result']): + # Find /start + if not start.search(_msg['message']['text']): + continue + + _id = _msg['message']['from'].get('id', 0) + _user = _msg['message']['from'].get('first_name') + self.logger.info('Detected telegram user %s (userid=%d)' % ( + _user, _id)) + # Return our detected userid + return _id + + self.logger.warning( + 'Could not detect bot owner. Is it running (/start)?') + + return 0 + def notify(self, title, body, notify_type, **kwargs): """ Perform Telegram Notification @@ -149,19 +339,24 @@ class NotifyTelegram(NotifyBase): payload = {} - if self.notify_format == NotifyFormat.HTML: - # HTML - payload['parse_mode'] = 'HTML' - payload['text'] = '%s\r\n%s' % (title, body) + # HTML Spaces ( ) and tabs ( ) aren't supported + # See https://core.telegram.org/bots/api#html-style + title = re.sub(' ?', ' ', title, re.I) + body = re.sub(' ?', ' ', body, re.I) + # Tabs become 3 spaces + title = re.sub(' ?', ' ', title, re.I) + body = re.sub(' ?', ' ', body, re.I) - else: - # Text - # payload['parse_mode'] = 'Markdown' - payload['parse_mode'] = 'HTML' - payload['text'] = '%s\r\n%s' % ( - NotifyBase.escape_html(title), - NotifyBase.escape_html(body), - ) + # HTML + title = NotifyBase.escape_html(title, whitespace=False) + body = NotifyBase.escape_html(body, whitespace=False) + + payload['parse_mode'] = 'HTML' + + payload['text'] = '%s\r\n%s' % ( + title, + body, + ) # Create a copy of the chat_ids list chat_ids = list(self.chat_ids) @@ -183,9 +378,17 @@ class NotifyTelegram(NotifyBase): else: # ID - payload['chat_id'] = chat_id.group('idno') + payload['chat_id'] = int(chat_id.group('idno')) + + if self.include_image is True: + # Send an image + if self.send_image( + payload['chat_id'], notify_type) is not None: + # We sent a post (whether we were successful or not) + # we still hit the remote server... just throttle + # before our next hit server query + self.throttle() - self.logger.debug('Telegram POST URL: %s' % url) self.logger.debug('Telegram POST URL: %s (cert_verify=%r)' % ( url, self.verify_certificate, )) @@ -204,7 +407,7 @@ class NotifyTelegram(NotifyBase): try: # Try to get the error message if we can: - error_msg = loads(r.text)['description'] + error_msg = loads(r.content)['description'] except: error_msg = None @@ -236,9 +439,12 @@ class NotifyTelegram(NotifyBase): # Flag our error has_error = True + else: + self.logger.info('Sent Telegram notification.') + except requests.RequestException as e: self.logger.warning( - 'A Connection error occured sending Telegram:%s ' % ( + 'A connection error occured sending Telegram:%s ' % ( payload['chat_id']) + 'notification.' ) self.logger.debug('Socket Exception: %s' % str(e)) @@ -322,4 +528,8 @@ class NotifyTelegram(NotifyBase): # Store our chat ids results['chat_ids'] = chat_ids + # Include images with our message + results['include_image'] = \ + parse_bool(results['qsd'].get('image', False)) + return results diff --git a/apprise/plugins/NotifyToasty.py b/apprise/plugins/NotifyToasty.py index fed3b8e1..1aae6de1 100644 --- a/apprise/plugins/NotifyToasty.py +++ b/apprise/plugins/NotifyToasty.py @@ -2,7 +2,7 @@ # # (Super) Toasty Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -133,6 +133,10 @@ class NotifyToasty(NotifyBase): # Return; we're done has_error = True + else: + self.logger.info( + 'Sent Toasty notification to %s.' % device) + except requests.RequestException as e: self.logger.warning( 'A Connection error occured sending Toasty:%s ' % ( diff --git a/apprise/plugins/NotifyTwitter/NotifyTwitter.py b/apprise/plugins/NotifyTwitter/NotifyTwitter.py index 11168089..aaf01638 100644 --- a/apprise/plugins/NotifyTwitter/NotifyTwitter.py +++ b/apprise/plugins/NotifyTwitter/NotifyTwitter.py @@ -2,7 +2,7 @@ # # Twitter Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -102,6 +102,7 @@ class NotifyTwitter(NotifyBase): # Send our Direct Message api.send_direct_message(self.user, text=text) + self.logger.info('Sent Twitter DM notification.') except Exception as e: self.logger.warning( diff --git a/apprise/plugins/NotifyXML.py b/apprise/plugins/NotifyXML.py index 7e7285b6..c056ab89 100644 --- a/apprise/plugins/NotifyXML.py +++ b/apprise/plugins/NotifyXML.py @@ -2,7 +2,7 @@ # # XML Notify Wrapper # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -136,6 +136,9 @@ class NotifyXML(NotifyBase): # Return; we're done return False + else: + self.logger.info('Sent XML notification.') + except requests.RequestException as e: self.logger.warning( 'A Connection error occured sending XML ' diff --git a/test/test_rest_plugins.py b/test/test_rest_plugins.py index 2ed184ca..1eceeff7 100644 --- a/test/test_rest_plugins.py +++ b/test/test_rest_plugins.py @@ -2,7 +2,7 @@ # # REST Based Plugins - Unit Tests # -# Copyright (C) 2017 Chris Caron +# Copyright (C) 2017-2018 Chris Caron # # This file is part of apprise. # @@ -985,19 +985,15 @@ TEST_URLS = ( ('tgram://bottest@123456789:abcdefg_hijklmnop/lead2gold/', { 'instance': plugins.NotifyTelegram, }), - # Testing valid format - ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?format=text', { + # Testing image + ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?image=Yes', { 'instance': plugins.NotifyTelegram, }), - # Testing valid format - ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?format=html', { - 'instance': plugins.NotifyTelegram, - }), - # Testing invalid format (fall's back to text) + # Testing invalid format (fall's back to html) ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?format=invalid', { 'instance': plugins.NotifyTelegram, }), - # Testing empty format (falls back to text) + # Testing empty format (falls back to html) ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?format=', { 'instance': plugins.NotifyTelegram, }), @@ -1021,7 +1017,7 @@ TEST_URLS = ( 'response': False, 'requests_response_code': requests.codes.internal_server_error, }), - ('tgram://123456789:abcdefg_hijklmnop/lead2gold/', { + ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?image=Yes', { 'instance': plugins.NotifyTelegram, # force a failure without an image specified 'include_image': False, @@ -1055,17 +1051,26 @@ TEST_URLS = ( 'response': False, 'requests_response_code': 999, }), + # Test with image set + ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?image=Yes', { + 'instance': plugins.NotifyTelegram, + # throw a bizzare code forcing us to fail to look it up without + # having an image included + 'include_image': True, + 'response': False, + 'requests_response_code': 999, + }), ('tgram://123456789:abcdefg_hijklmnop/lead2gold/', { 'instance': plugins.NotifyTelegram, # Throws a series of connection and transfer exceptions when this flag # is set and tests that we gracfully handle them 'test_requests_exceptions': True, }), - ('tgram://123456789:abcdefg_hijklmnop/lead2gold/', { + ('tgram://123456789:abcdefg_hijklmnop/lead2gold/?image=Yes', { 'instance': plugins.NotifyTelegram, # Throws a series of connection and transfer exceptions when this flag # is set and tests that we gracfully handle them without images set - 'include_image': False, + 'include_image': True, 'test_requests_exceptions': True, }), @@ -1878,6 +1883,8 @@ def test_notify_telegram_plugin(mock_post, mock_get): mock_post.return_value = requests.Request() mock_post.return_value.status_code = requests.codes.ok mock_get.return_value.status_code = requests.codes.ok + mock_get.return_value.content = '{}' + mock_post.return_value.content = '{}' try: obj = plugins.NotifyTelegram(bot_token=None, chat_ids=chat_ids) @@ -1960,3 +1967,127 @@ def test_notify_telegram_plugin(mock_post, mock_get): title='title', body='body', notify_type=NotifyType.INFO) is False assert nimg_obj.notify( title='title', body='body', notify_type=NotifyType.INFO) is False + + # Bot Token Detection + # Just to make it clear to people reading this code and trying to learn + # what is going on. Apprise tries to detect the bot owner if you don't + # specify a user to message. The idea is to just default to messaging + # the bot owner himself (it makes it easier for people). So we're testing + # the creating of a Telegram Notification without providing a chat ID. + # We're testing the error handling of this bot detection section of the + # code + mock_post.return_value.content = dumps({ + "ok": True, + "result": [{ + "update_id": 645421321, + "message": { + "message_id": 1, + "from": { + "id": 532389719, + "is_bot": False, + "first_name": "Chris", + "language_code": "en-US" + }, + "chat": { + "id": 532389719, + "first_name": "Chris", + "type": "private" + }, + "date": 1519694394, + "text": "/start", + "entities": [{ + "offset": 0, + "length": 6, + "type": "bot_command", + }], + }}, + ], + }) + mock_post.return_value.status_code = requests.codes.ok + + obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=None) + assert(len(obj.chat_ids) == 1) + assert(obj.chat_ids[0] == '532389719') + + # Do the test again, but without the expected (parsed response) + mock_post.return_value.content = dumps({ + "ok": True, + "result": [{ + "message": { + "text": "/ignored.entry", + }}, + ], + }) + try: + obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=None) + # No chat_ids specified + assert(False) + + except TypeError: + # Exception should be thrown about the fact no token was specified + assert(True) + + # Test our bot detection with a internal server error + mock_post.return_value.status_code = requests.codes.internal_server_error + try: + obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=None) + # No chat_ids specified + assert(False) + + except TypeError: + # Exception should be thrown about the fact no token was specified + assert(True) + + # Test our bot detection with an unmappable html error + mock_post.return_value.status_code = 999 + try: + obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=None) + # No chat_ids specified + assert(False) + + except TypeError: + # Exception should be thrown about the fact no token was specified + assert(True) + + # Do it again but this time provide a failure message + mock_post.return_value.content = dumps({'description': 'Failure Message'}) + try: + obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=None) + # No chat_ids specified + assert(False) + + except TypeError: + # Exception should be thrown about the fact no token was specified + assert(True) + + # Do it again but this time provide a failure message and perform a + # notification without a bot detection by providing at least 1 chat id + obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=['@abcd']) + assert nimg_obj.notify( + title='title', body='body', notify_type=NotifyType.INFO) is False + + # Test our exception handling with bot detection + test_requests_exceptions = ( + requests.ConnectionError( + 0, 'requests.ConnectionError() not handled'), + requests.RequestException( + 0, 'requests.RequestException() not handled'), + requests.HTTPError( + 0, 'requests.HTTPError() not handled'), + requests.ReadTimeout( + 0, 'requests.ReadTimeout() not handled'), + requests.TooManyRedirects( + 0, 'requests.TooManyRedirects() not handled'), + ) + + # iterate over our exceptions and test them + for _exception in test_requests_exceptions: + mock_post.side_effect = _exception + try: + obj = plugins.NotifyTelegram(bot_token=bot_token, chat_ids=None) + # No chat_ids specified + assert(False) + + except TypeError: + # Exception should be thrown about the fact no token was specified + assert(True)