mirror of
https://github.com/caronc/apprise.git
synced 2024-11-21 23:53:23 +01:00
Telegram Attachment Support (#178)
This commit is contained in:
Chris Caron
GitHub
parent
cc8fae4c73
commit
4065108b08
@ -51,6 +51,7 @@
|
||||
# - https://core.telegram.org/bots/api
|
||||
import requests
|
||||
import re
|
||||
import os
|
||||
|
||||
from json import loads
|
||||
from json import dumps
|
||||
@ -63,6 +64,7 @@ from ..utils import parse_bool
|
||||
from ..utils import parse_list
|
||||
from ..utils import validate_regex
|
||||
from ..AppriseLocale import gettext_lazy as _
|
||||
from ..attachment.AttachBase import AttachBase
|
||||
|
||||
TELEGRAM_IMAGE_XY = NotifyImageSize.XY_256
|
||||
|
||||
@ -100,12 +102,71 @@ class NotifyTelegram(NotifyBase):
|
||||
# The maximum allowable characters allowed in the body per message
|
||||
body_maxlen = 4096
|
||||
|
||||
# Telegram is limited to sending a maximum of 100 requests per second.
|
||||
request_rate_per_sec = 0.001
|
||||
|
||||
# Define object templates
|
||||
templates = (
|
||||
'{schema}://{bot_token}',
|
||||
'{schema}://{bot_token}/{targets}',
|
||||
)
|
||||
|
||||
# Telegram Attachment Support
|
||||
mime_lookup = (
|
||||
# This list is intentionally ordered so that it can be scanned
|
||||
# from top to bottom. The last entry is a catch-all
|
||||
|
||||
# Animations are documented to only support gif or H.264/MPEG-4
|
||||
# Source: https://core.telegram.org/bots/api#sendanimation
|
||||
{
|
||||
'regex': re.compile(r'^(image/gif|video/H264)', re.I),
|
||||
'function_name': 'sendAnimation',
|
||||
'key': 'animation',
|
||||
},
|
||||
|
||||
# This entry is intentially placed below the sendAnimiation allowing
|
||||
# it to catch gif files. This then becomes a catch all to remaining
|
||||
# image types.
|
||||
# Source: https://core.telegram.org/bots/api#sendphoto
|
||||
{
|
||||
'regex': re.compile(r'^image/.*', re.I),
|
||||
'function_name': 'sendPhoto',
|
||||
'key': 'photo',
|
||||
},
|
||||
|
||||
# Video is documented to only support .mp4
|
||||
# Source: https://core.telegram.org/bots/api#sendvideo
|
||||
{
|
||||
'regex': re.compile(r'^video/mp4', re.I),
|
||||
'function_name': 'sendVideo',
|
||||
'key': 'video',
|
||||
},
|
||||
|
||||
# Voice supports ogg
|
||||
# Source: https://core.telegram.org/bots/api#sendvoice
|
||||
{
|
||||
'regex': re.compile(r'^(application|audio)/ogg', re.I),
|
||||
'function_name': 'sendVoice',
|
||||
'key': 'voice',
|
||||
},
|
||||
|
||||
# Audio supports mp3 and m4a only
|
||||
# Source: https://core.telegram.org/bots/api#sendaudio
|
||||
{
|
||||
'regex': re.compile(r'^audio/(mpeg|mp4a-latm)', re.I),
|
||||
'function_name': 'sendAudio',
|
||||
'key': 'audio',
|
||||
},
|
||||
|
||||
# Catch All (all other types)
|
||||
# Source: https://core.telegram.org/bots/api#senddocument
|
||||
{
|
||||
'regex': re.compile(r'.*', re.I),
|
||||
'function_name': 'sendDocument',
|
||||
'key': 'document',
|
||||
},
|
||||
)
|
||||
|
||||
# Define our template tokens
|
||||
template_tokens = dict(NotifyBase.template_tokens, **{
|
||||
'bot_token': {
|
||||
@ -189,82 +250,101 @@ class NotifyTelegram(NotifyBase):
|
||||
# or not.
|
||||
self.include_image = include_image
|
||||
|
||||
def send_image(self, chat_id, notify_type):
|
||||
def send_media(self, chat_id, notify_type, attach=None):
|
||||
"""
|
||||
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.
|
||||
# Prepare our Headers
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
}
|
||||
|
||||
# Our function name and payload are determined on the path
|
||||
function_name = 'SendPhoto'
|
||||
key = 'photo'
|
||||
path = None
|
||||
|
||||
if isinstance(attach, AttachBase):
|
||||
# Store our path to our file
|
||||
path = attach.path
|
||||
file_name = attach.name
|
||||
mimetype = attach.mimetype
|
||||
|
||||
if not path:
|
||||
# Could not load attachment
|
||||
return False
|
||||
|
||||
# Process our attachment
|
||||
function_name, key = \
|
||||
next(((x['function_name'], x['key']) for x in self.mime_lookup
|
||||
if x['regex'].match(mimetype))) # pragma: no cover
|
||||
|
||||
else:
|
||||
attach = self.image_path(notify_type) if attach is None else attach
|
||||
if attach is None:
|
||||
# Nothing specified to send
|
||||
return True
|
||||
|
||||
# Take on specified attachent as path
|
||||
path = attach
|
||||
file_name = os.path.basename(path)
|
||||
|
||||
url = '%s%s/%s' % (
|
||||
self.notify_url,
|
||||
self.bot_token,
|
||||
'sendPhoto'
|
||||
function_name,
|
||||
)
|
||||
|
||||
# Acquire our image path if configured to do so; we don't bother
|
||||
# checking to see if selfinclude_image is set here because the
|
||||
# send_image() function itself (this function) checks this flag
|
||||
# already
|
||||
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))
|
||||
|
||||
# No need to fail; we may have been configured this way through
|
||||
# the apprise.AssetObject()
|
||||
return True
|
||||
# Always call throttle before any remote server i/o is made;
|
||||
# Telegram throttles to occur before sending the image so that
|
||||
# content can arrive together.
|
||||
self.throttle()
|
||||
|
||||
try:
|
||||
with open(path, 'rb') as f:
|
||||
# Configure file payload (for upload)
|
||||
files = {
|
||||
'photo': f,
|
||||
}
|
||||
|
||||
payload = {
|
||||
'chat_id': chat_id,
|
||||
}
|
||||
files = {key: (file_name, f)}
|
||||
payload = {'chat_id': chat_id}
|
||||
|
||||
self.logger.debug(
|
||||
'Telegram image POST URL: %s (cert_verify=%r)' % (
|
||||
'Telegram attachment POST URL: %s (cert_verify=%r)' % (
|
||||
url, self.verify_certificate))
|
||||
|
||||
try:
|
||||
r = requests.post(
|
||||
url,
|
||||
files=files,
|
||||
data=payload,
|
||||
verify=self.verify_certificate,
|
||||
)
|
||||
r = requests.post(
|
||||
url,
|
||||
headers=headers,
|
||||
files=files,
|
||||
data=payload,
|
||||
verify=self.verify_certificate,
|
||||
)
|
||||
|
||||
if r.status_code != requests.codes.ok:
|
||||
# We had a problem
|
||||
status_str = NotifyTelegram\
|
||||
.http_response_code_lookup(r.status_code)
|
||||
if r.status_code != requests.codes.ok:
|
||||
# We had a problem
|
||||
status_str = NotifyTelegram\
|
||||
.http_response_code_lookup(r.status_code)
|
||||
|
||||
self.logger.warning(
|
||||
'Failed to send Telegram image: '
|
||||
'{}{}error={}.'.format(
|
||||
status_str,
|
||||
', ' if status_str else '',
|
||||
r.status_code))
|
||||
|
||||
self.logger.debug(
|
||||
'Response Details:\r\n{}'.format(r.content))
|
||||
|
||||
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))
|
||||
'Failed to send Telegram attachment: '
|
||||
'{}{}error={}.'.format(
|
||||
status_str,
|
||||
', ' if status_str else '',
|
||||
r.status_code))
|
||||
|
||||
self.logger.debug(
|
||||
'Response Details:\r\n{}'.format(r.content))
|
||||
|
||||
return False
|
||||
|
||||
return True
|
||||
# Content was sent successfully if we got here
|
||||
return True
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A connection error occured posting Telegram '
|
||||
'attachment.')
|
||||
self.logger.debug('Socket Exception: %s' % str(e))
|
||||
|
||||
except (IOError, OSError):
|
||||
# IOError is present for backwards compatibility with Python
|
||||
@ -364,26 +444,20 @@ class NotifyTelegram(NotifyBase):
|
||||
|
||||
# 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(r'^\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)?')
|
||||
if 'ok' in response and response['ok'] is True \
|
||||
and 'result' in response and len(response['result']):
|
||||
entry = response['result'][0]
|
||||
_id = entry['message']['from'].get('id', 0)
|
||||
_user = entry['message']['from'].get('first_name')
|
||||
self.logger.info('Detected telegram user %s (userid=%d)' % (
|
||||
_user, _id))
|
||||
# Return our detected userid
|
||||
return _id
|
||||
|
||||
return 0
|
||||
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
|
||||
**kwargs):
|
||||
"""
|
||||
Perform Telegram Notification
|
||||
"""
|
||||
@ -476,15 +550,20 @@ class NotifyTelegram(NotifyBase):
|
||||
# ID
|
||||
payload['chat_id'] = int(chat_id.group('idno'))
|
||||
|
||||
if self.include_image is True:
|
||||
# Define our path
|
||||
if not self.send_media(payload['chat_id'], notify_type):
|
||||
# We failed to send the image associated with our
|
||||
notify_type
|
||||
self.logger.warning(
|
||||
'Failed to send Telegram type image to {}.',
|
||||
payload['chat_id'])
|
||||
|
||||
# Always call throttle before any remote server i/o is made;
|
||||
# Telegram throttles to occur before sending the image so that
|
||||
# content can arrive together.
|
||||
self.throttle()
|
||||
|
||||
if self.include_image is True:
|
||||
# Send an image
|
||||
self.send_image(payload['chat_id'], notify_type)
|
||||
|
||||
self.logger.debug('Telegram POST URL: %s (cert_verify=%r)' % (
|
||||
url, self.verify_certificate,
|
||||
))
|
||||
@ -524,9 +603,6 @@ class NotifyTelegram(NotifyBase):
|
||||
has_error = True
|
||||
continue
|
||||
|
||||
else:
|
||||
self.logger.info('Sent Telegram notification.')
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A connection error occured sending Telegram:%s ' % (
|
||||
@ -538,6 +614,22 @@ class NotifyTelegram(NotifyBase):
|
||||
has_error = True
|
||||
continue
|
||||
|
||||
self.logger.info('Sent Telegram notification.')
|
||||
|
||||
if attach:
|
||||
# Send our attachments now (if specified and if it exists)
|
||||
for attachment in attach:
|
||||
sent_attachment = self.send_media(
|
||||
payload['chat_id'], notify_type, attach=attachment)
|
||||
|
||||
if not sent_attachment:
|
||||
# We failed; don't continue
|
||||
has_error = True
|
||||
break
|
||||
|
||||
self.logger.info(
|
||||
'Sent Telegram attachment: {}.'.format(attachment))
|
||||
|
||||
return not has_error
|
||||
|
||||
def url(self, privacy=False, *args, **kwargs):
|
||||
|
||||
@ -5114,203 +5114,6 @@ def test_notify_rocketchat_plugin(mock_post, mock_get):
|
||||
assert obj.logout() is False
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_notify_telegram_plugin(mock_post, mock_get):
|
||||
"""
|
||||
API: NotifyTelegram() Extra Checks
|
||||
|
||||
"""
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyBase.request_rate_per_sec = 0
|
||||
|
||||
# Bot Token
|
||||
bot_token = '123456789:abcdefg_hijklmnop'
|
||||
invalid_bot_token = 'abcd:123'
|
||||
|
||||
# Chat ID
|
||||
chat_ids = 'l2g, lead2gold'
|
||||
|
||||
# Prepare Mock
|
||||
mock_get.return_value = requests.Request()
|
||||
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 = '{}'
|
||||
|
||||
# Exception should be thrown about the fact no bot token was specified
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=None, targets=chat_ids)
|
||||
|
||||
# Exception should be thrown about the fact an invalid bot token was
|
||||
# specifed
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=invalid_bot_token, targets=chat_ids)
|
||||
|
||||
obj = plugins.NotifyTelegram(
|
||||
bot_token=bot_token, targets=chat_ids, include_image=True)
|
||||
assert isinstance(obj, plugins.NotifyTelegram) is True
|
||||
assert len(obj.targets) == 2
|
||||
|
||||
# Test Image Sending Exceptions
|
||||
mock_get.side_effect = IOError()
|
||||
mock_post.side_effect = IOError()
|
||||
obj.send_image(obj.targets[0], NotifyType.INFO)
|
||||
|
||||
# Restore their entries
|
||||
mock_get.side_effect = None
|
||||
mock_post.side_effect = None
|
||||
mock_get.return_value.content = '{}'
|
||||
mock_post.return_value.content = '{}'
|
||||
|
||||
# test url call
|
||||
assert isinstance(obj.url(), six.string_types) is True
|
||||
|
||||
# test privacy version of url
|
||||
assert isinstance(obj.url(privacy=True), six.string_types) is True
|
||||
assert obj.url(privacy=True).startswith('tgram://1...p/') is True
|
||||
|
||||
# Test that we can load the string we generate back:
|
||||
obj = plugins.NotifyTelegram(**plugins.NotifyTelegram.parse_url(obj.url()))
|
||||
assert isinstance(obj, plugins.NotifyTelegram) is True
|
||||
|
||||
# Prepare Mock to fail
|
||||
response = mock.Mock()
|
||||
response.status_code = requests.codes.internal_server_error
|
||||
|
||||
# a error response
|
||||
response.content = dumps({
|
||||
'description': 'test',
|
||||
})
|
||||
mock_get.return_value = response
|
||||
mock_post.return_value = response
|
||||
|
||||
# No image asset
|
||||
nimg_obj = plugins.NotifyTelegram(bot_token=bot_token, targets=chat_ids)
|
||||
nimg_obj.asset = AppriseAsset(image_path_mask=False, image_url_mask=False)
|
||||
|
||||
# Test that our default settings over-ride base settings since they are
|
||||
# not the same as the one specified in the base; this check merely
|
||||
# ensures our plugin inheritance is working properly
|
||||
assert obj.body_maxlen == plugins.NotifyTelegram.body_maxlen
|
||||
|
||||
# We don't override the title maxlen so we should be set to the same
|
||||
# as our parent class in this case
|
||||
assert obj.title_maxlen == plugins.NotifyBase.title_maxlen
|
||||
|
||||
# This tests erroneous messages involving multiple chat ids
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
assert nimg_obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
|
||||
# This tests erroneous messages involving a single chat id
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets='l2g')
|
||||
nimg_obj = plugins.NotifyTelegram(bot_token=bot_token, targets='l2g')
|
||||
nimg_obj.asset = AppriseAsset(image_path_mask=False, image_url_mask=False)
|
||||
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
assert nimg_obj.notify(
|
||||
body='body', title='title', 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, targets=None)
|
||||
assert len(obj.targets) == 1
|
||||
assert obj.targets[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",
|
||||
}},
|
||||
],
|
||||
})
|
||||
|
||||
# Exception should be thrown about the fact no bot token was specified
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# Detect the bot with a bad response
|
||||
mock_post.return_value.content = dumps({})
|
||||
obj.detect_bot_owner()
|
||||
|
||||
# Test our bot detection with a internal server error
|
||||
mock_post.return_value.status_code = requests.codes.internal_server_error
|
||||
|
||||
# Exception should be thrown over internal server error caused
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# Test our bot detection with an unmappable html error
|
||||
mock_post.return_value.status_code = 999
|
||||
# Exception should be thrown over invali internal error no
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# Do it again but this time provide a failure message
|
||||
mock_post.return_value.content = dumps({'description': 'Failure Message'})
|
||||
|
||||
# Exception should be thrown about the fact no bot token was specified
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# 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, targets=['@abcd'])
|
||||
assert nimg_obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
|
||||
# iterate over our exceptions and test them
|
||||
for _exception in REQUEST_EXCEPTIONS:
|
||||
mock_post.side_effect = _exception
|
||||
|
||||
# No chat_ids specified
|
||||
with pytest.raises(TypeError):
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
|
||||
def test_notify_overflow_truncate():
|
||||
"""
|
||||
API: Overflow Truncate Functionality Testing
|
||||
|
||||
252
test/test_telegram.py
Normal file
252
test/test_telegram.py
Normal file
@ -0,0 +1,252 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# 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 os
|
||||
import six
|
||||
import pytest
|
||||
import mock
|
||||
import requests
|
||||
from json import dumps
|
||||
from apprise import AppriseAttachment
|
||||
from apprise import AppriseAsset
|
||||
from apprise import NotifyType
|
||||
from apprise import plugins
|
||||
|
||||
# Disable logging for a cleaner testing output
|
||||
import logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
# Attachment Directory
|
||||
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_notify_telegram_plugin(mock_post, mock_get, tmpdir):
|
||||
"""
|
||||
API: NotifyTelegram() Tests
|
||||
|
||||
"""
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyBase.request_rate_per_sec = 0
|
||||
|
||||
# Bot Token
|
||||
bot_token = '123456789:abcdefg_hijklmnop'
|
||||
invalid_bot_token = 'abcd:123'
|
||||
|
||||
# Chat ID
|
||||
chat_ids = 'l2g, lead2gold'
|
||||
|
||||
# Prepare Mock
|
||||
mock_get.return_value = requests.Request()
|
||||
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 = '{}'
|
||||
|
||||
# Exception should be thrown about the fact no bot token was specified
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=None, targets=chat_ids)
|
||||
|
||||
# Exception should be thrown about the fact an invalid bot token was
|
||||
# specifed
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=invalid_bot_token, targets=chat_ids)
|
||||
|
||||
obj = plugins.NotifyTelegram(
|
||||
bot_token=bot_token, targets=chat_ids, include_image=True)
|
||||
assert isinstance(obj, plugins.NotifyTelegram) is True
|
||||
assert len(obj.targets) == 2
|
||||
|
||||
# Test Image Sending Exceptions
|
||||
mock_post.side_effect = IOError()
|
||||
assert not obj.send_media(obj.targets[0], NotifyType.INFO)
|
||||
|
||||
# Test our other objects
|
||||
mock_post.side_effect = requests.HTTPError
|
||||
assert not obj.send_media(obj.targets[0], NotifyType.INFO)
|
||||
|
||||
# Restore their entries
|
||||
mock_get.side_effect = None
|
||||
mock_post.side_effect = None
|
||||
mock_get.return_value.content = '{}'
|
||||
mock_post.return_value.content = '{}'
|
||||
|
||||
# test url call
|
||||
assert isinstance(obj.url(), six.string_types) is True
|
||||
|
||||
# test privacy version of url
|
||||
assert isinstance(obj.url(privacy=True), six.string_types) is True
|
||||
assert obj.url(privacy=True).startswith('tgram://1...p/') is True
|
||||
|
||||
# Test that we can load the string we generate back:
|
||||
obj = plugins.NotifyTelegram(**plugins.NotifyTelegram.parse_url(obj.url()))
|
||||
assert isinstance(obj, plugins.NotifyTelegram) is True
|
||||
|
||||
# Prepare Mock to fail
|
||||
response = mock.Mock()
|
||||
response.status_code = requests.codes.internal_server_error
|
||||
|
||||
# a error response
|
||||
response.content = dumps({
|
||||
'description': 'test',
|
||||
})
|
||||
mock_get.return_value = response
|
||||
mock_post.return_value = response
|
||||
|
||||
# No image asset
|
||||
nimg_obj = plugins.NotifyTelegram(bot_token=bot_token, targets=chat_ids)
|
||||
nimg_obj.asset = AppriseAsset(image_path_mask=False, image_url_mask=False)
|
||||
|
||||
# Test that our default settings over-ride base settings since they are
|
||||
# not the same as the one specified in the base; this check merely
|
||||
# ensures our plugin inheritance is working properly
|
||||
assert obj.body_maxlen == plugins.NotifyTelegram.body_maxlen
|
||||
|
||||
# We don't override the title maxlen so we should be set to the same
|
||||
# as our parent class in this case
|
||||
assert obj.title_maxlen == plugins.NotifyBase.title_maxlen
|
||||
|
||||
# This tests erroneous messages involving multiple chat ids
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
assert nimg_obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
|
||||
# This tests erroneous messages involving a single chat id
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets='l2g')
|
||||
nimg_obj = plugins.NotifyTelegram(bot_token=bot_token, targets='l2g')
|
||||
nimg_obj.asset = AppriseAsset(image_path_mask=False, image_url_mask=False)
|
||||
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
assert nimg_obj.notify(
|
||||
body='body', title='title', 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
|
||||
|
||||
# Test sending attachments
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets='12345')
|
||||
assert len(obj.targets) == 1
|
||||
assert obj.targets[0] == '12345'
|
||||
|
||||
path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif')
|
||||
attach = AppriseAttachment(path)
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO,
|
||||
attach=attach) is True
|
||||
|
||||
path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg')
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO,
|
||||
attach=path) is False
|
||||
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
assert len(obj.targets) == 1
|
||||
assert obj.targets[0] == '532389719'
|
||||
|
||||
# Do the test again, but without the expected (parsed response)
|
||||
mock_post.return_value.content = dumps({
|
||||
"ok": True,
|
||||
"result": [],
|
||||
})
|
||||
|
||||
# Exception should be thrown about the fact no bot token was specified
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# Detect the bot with a bad response
|
||||
mock_post.return_value.content = dumps({})
|
||||
obj.detect_bot_owner()
|
||||
|
||||
# Test our bot detection with a internal server error
|
||||
mock_post.return_value.status_code = requests.codes.internal_server_error
|
||||
|
||||
# Exception should be thrown over internal server error caused
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# Test our bot detection with an unmappable html error
|
||||
mock_post.return_value.status_code = 999
|
||||
# Exception should be thrown over invali internal error no
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# Do it again but this time provide a failure message
|
||||
mock_post.return_value.content = dumps({'description': 'Failure Message'})
|
||||
|
||||
# Exception should be thrown about the fact no bot token was specified
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
|
||||
# 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, targets=['@abcd'])
|
||||
assert nimg_obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
|
||||
# iterate over our exceptions and test them
|
||||
mock_post.side_effect = requests.HTTPError
|
||||
|
||||
# No chat_ids specified
|
||||
with pytest.raises(TypeError):
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 233 KiB After Width: | Height: | Size: 242 KiB |
Loading…
Reference in New Issue
Block a user