mirror of
https://github.com/caronc/apprise.git
synced 2024-11-25 09:33:50 +01:00
Improved Apprise encoding and text format support (#566)
This commit is contained in:
Chris Caron
GitHub
parent
5d14259227
commit
a05b042c6d
@ -484,22 +484,43 @@ class Apprise(object):
|
||||
|
||||
if len(self) == 0:
|
||||
# Nothing to notify
|
||||
raise TypeError("No service(s) to notify")
|
||||
msg = "There are service(s) to notify"
|
||||
logger.error(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
if not (title or body):
|
||||
raise TypeError("No message content specified to deliver")
|
||||
msg = "No message content specified to deliver"
|
||||
logger.error(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
if six.PY2:
|
||||
# Python 2.7.x Unicode Character Handling
|
||||
# Ensure we're working with utf-8
|
||||
if isinstance(title, unicode): # noqa: F821
|
||||
title = title.encode('utf-8')
|
||||
try:
|
||||
if six.PY2:
|
||||
# Python 2.7 encoding support isn't the greatest, so we try
|
||||
# to ensure that we're ALWAYS dealing with unicode characters
|
||||
# prior to entrying the next part. This is especially required
|
||||
# for Markdown support
|
||||
if title and isinstance(title, str): # noqa: F821
|
||||
title = title.decode(self.asset.encoding)
|
||||
|
||||
if isinstance(body, unicode): # noqa: F821
|
||||
body = body.encode('utf-8')
|
||||
if body and isinstance(body, str): # noqa: F821
|
||||
body = body.decode(self.asset.encoding)
|
||||
|
||||
else: # Python 3+
|
||||
if title and isinstance(title, bytes): # noqa: F821
|
||||
title = title.decode(self.asset.encoding)
|
||||
|
||||
if body and isinstance(body, bytes): # noqa: F821
|
||||
body = body.decode(self.asset.encoding)
|
||||
|
||||
except UnicodeDecodeError:
|
||||
msg = 'The content passed into Apprise was not of encoding ' \
|
||||
'type: {}'.format(self.asset.encoding)
|
||||
logger.error(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
# Tracks conversions
|
||||
conversion_map = dict()
|
||||
conversion_body_map = dict()
|
||||
conversion_title_map = dict()
|
||||
|
||||
# Prepare attachments if required
|
||||
if attach is not None and not isinstance(attach, AppriseAttachment):
|
||||
@ -519,9 +540,13 @@ class Apprise(object):
|
||||
# If our code reaches here, we either did not define a tag (it
|
||||
# was set to None), or we did define a tag and the logic above
|
||||
# determined we need to notify the service it's associated with
|
||||
if server.notify_format not in conversion_map:
|
||||
conversion_map[server.notify_format] = convert_between(
|
||||
body_format, server.notify_format, body)
|
||||
if server.notify_format not in conversion_body_map:
|
||||
# Perform Conversion
|
||||
(conversion_title_map[server.notify_format],
|
||||
conversion_body_map[server.notify_format]) = \
|
||||
convert_between(
|
||||
body_format, server.notify_format, body=body,
|
||||
title=title)
|
||||
|
||||
if interpret_escapes:
|
||||
#
|
||||
@ -531,8 +556,13 @@ class Apprise(object):
|
||||
try:
|
||||
# Added overhead required due to Python 3 Encoding Bug
|
||||
# identified here: https://bugs.python.org/issue21331
|
||||
conversion_map[server.notify_format] = \
|
||||
conversion_map[server.notify_format]\
|
||||
conversion_body_map[server.notify_format] = \
|
||||
conversion_body_map[server.notify_format]\
|
||||
.encode('ascii', 'backslashreplace')\
|
||||
.decode('unicode-escape')
|
||||
|
||||
conversion_title_map[server.notify_format] = \
|
||||
conversion_title_map[server.notify_format]\
|
||||
.encode('ascii', 'backslashreplace')\
|
||||
.decode('unicode-escape')
|
||||
|
||||
@ -540,39 +570,43 @@ class Apprise(object):
|
||||
# This occurs using a very old verion of Python 2.7
|
||||
# such as the one that ships with CentOS/RedHat 7.x
|
||||
# (v2.7.5).
|
||||
conversion_map[server.notify_format] = \
|
||||
conversion_map[server.notify_format] \
|
||||
conversion_body_map[server.notify_format] = \
|
||||
conversion_body_map[server.notify_format] \
|
||||
.decode('string_escape')
|
||||
|
||||
conversion_title_map[server.notify_format] = \
|
||||
conversion_title_map[server.notify_format] \
|
||||
.decode('string_escape')
|
||||
|
||||
except AttributeError:
|
||||
# Must be of string type
|
||||
logger.error('Failed to escape message body')
|
||||
raise TypeError
|
||||
msg = 'Failed to escape message body'
|
||||
logger.error(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
if title:
|
||||
try:
|
||||
# Added overhead required due to Python 3 Encoding
|
||||
# Bug identified here:
|
||||
# https://bugs.python.org/issue21331
|
||||
title = title\
|
||||
.encode('ascii', 'backslashreplace')\
|
||||
.decode('unicode-escape')
|
||||
if six.PY2:
|
||||
# Python 2.7 strings must be encoded as utf-8 for
|
||||
# consistency across all platforms
|
||||
if conversion_title_map[server.notify_format] and \
|
||||
isinstance(
|
||||
conversion_title_map[server.notify_format],
|
||||
unicode): # noqa: F821
|
||||
conversion_title_map[server.notify_format] = \
|
||||
conversion_title_map[server.notify_format]\
|
||||
.encode('utf-8')
|
||||
|
||||
except UnicodeDecodeError: # pragma: no cover
|
||||
# This occurs using a very old verion of Python 2.7
|
||||
# such as the one that ships with CentOS/RedHat 7.x
|
||||
# (v2.7.5).
|
||||
title = title.decode('string_escape')
|
||||
|
||||
except AttributeError:
|
||||
# Must be of string type
|
||||
logger.error('Failed to escape message title')
|
||||
raise TypeError
|
||||
if conversion_body_map[server.notify_format] and \
|
||||
isinstance(
|
||||
conversion_body_map[server.notify_format],
|
||||
unicode): # noqa: F821
|
||||
conversion_body_map[server.notify_format] = \
|
||||
conversion_body_map[server.notify_format]\
|
||||
.encode('utf-8')
|
||||
|
||||
yield handler(
|
||||
server,
|
||||
body=conversion_map[server.notify_format],
|
||||
title=title,
|
||||
body=conversion_body_map[server.notify_format],
|
||||
title=conversion_title_map[server.notify_format],
|
||||
notify_type=notify_type,
|
||||
attach=attach
|
||||
)
|
||||
|
||||
@ -29,6 +29,7 @@ from os.path import join
|
||||
from os.path import dirname
|
||||
from os.path import isfile
|
||||
from os.path import abspath
|
||||
from locale import getpreferredencoding
|
||||
from .common import NotifyType
|
||||
|
||||
|
||||
@ -110,6 +111,9 @@ class AppriseAsset(object):
|
||||
# to a new line.
|
||||
interpret_escapes = False
|
||||
|
||||
# Defines the encoding of the content passed into Apprise
|
||||
encoding = getpreferredencoding()
|
||||
|
||||
# For more detail see CWE-312 @
|
||||
# https://cwe.mitre.org/data/definitions/312.html
|
||||
#
|
||||
|
||||
@ -36,25 +36,51 @@ else:
|
||||
from html.parser import HTMLParser
|
||||
|
||||
|
||||
def convert_between(from_format, to_format, body):
|
||||
def convert_between(from_format, to_format, body, title=None):
|
||||
"""
|
||||
Converts between different notification formats. If no conversion exists,
|
||||
or the selected one fails, the original text will be returned.
|
||||
|
||||
This function returns a tuple as (title, body)
|
||||
"""
|
||||
|
||||
converters = {
|
||||
(NotifyFormat.MARKDOWN, NotifyFormat.HTML): markdown,
|
||||
(NotifyFormat.MARKDOWN, NotifyFormat.HTML): markdown_to_html,
|
||||
(NotifyFormat.TEXT, NotifyFormat.HTML): text_to_html,
|
||||
(NotifyFormat.HTML, NotifyFormat.TEXT): html_to_text,
|
||||
# For now; use same converter for Markdown support
|
||||
(NotifyFormat.HTML, NotifyFormat.MARKDOWN): html_to_text,
|
||||
}
|
||||
|
||||
if NotifyFormat.MARKDOWN in (from_format, to_format):
|
||||
# Tidy any exising pre-formating configuration
|
||||
title = '' if not title else title.lstrip('\r\n \t\v\b*#-')
|
||||
|
||||
else:
|
||||
title = '' if not title else title
|
||||
|
||||
convert = converters.get((from_format, to_format))
|
||||
return convert(body) if convert is not None else body
|
||||
title, body = convert(title=title, body=body) \
|
||||
if convert is not None else (title, body)
|
||||
|
||||
return (title, body)
|
||||
|
||||
|
||||
def text_to_html(body):
|
||||
def markdown_to_html(body, title=None):
|
||||
"""
|
||||
Handle Markdown conversions
|
||||
"""
|
||||
|
||||
return (
|
||||
# Title
|
||||
'' if not title else markdown(title),
|
||||
|
||||
# Body
|
||||
markdown(body),
|
||||
)
|
||||
|
||||
|
||||
def text_to_html(body, title=None):
|
||||
"""
|
||||
Converts a notification body from plain text to HTML.
|
||||
"""
|
||||
@ -86,11 +112,19 @@ def text_to_html(body):
|
||||
|
||||
# Execute our map against our body in addition to
|
||||
# swapping out new lines and replacing them with <br/>
|
||||
return re.sub(
|
||||
r'\r*\n', '<br/>\n', re_table.sub(lambda x: re_map[x.group()], body))
|
||||
return (
|
||||
# Title; swap whitespace with space
|
||||
'' if not title else re.sub(
|
||||
r'[\r\n]+', ' ', re_table.sub(
|
||||
lambda x: re_map[x.group()], title)),
|
||||
|
||||
# Body Formatting
|
||||
re.sub(
|
||||
r'\r*\n', '<br/>\n', re_table.sub(
|
||||
lambda x: re_map[x.group()], body)))
|
||||
|
||||
|
||||
def html_to_text(body):
|
||||
def html_to_text(body, title=None):
|
||||
"""
|
||||
Converts a notification body from HTML to plain text.
|
||||
"""
|
||||
@ -100,11 +134,20 @@ def html_to_text(body):
|
||||
# Python 2.7 requires an additional parsing to un-escape characters
|
||||
body = parser.unescape(body)
|
||||
|
||||
if title:
|
||||
if six.PY2:
|
||||
# Python 2.7 requires an additional parsing to un-escape characters
|
||||
title = parser.unescape(title)
|
||||
|
||||
parser.feed(title)
|
||||
parser.close()
|
||||
title = parser.converted
|
||||
|
||||
parser.feed(body)
|
||||
parser.close()
|
||||
result = parser.converted
|
||||
body = parser.converted
|
||||
|
||||
return result
|
||||
return ('' if not title else title, body)
|
||||
|
||||
|
||||
class HTMLConverter(HTMLParser, object):
|
||||
|
||||
@ -93,6 +93,9 @@ class NotifyTelegram(NotifyBase):
|
||||
# A URL that takes you to the setup/help of the specific protocol
|
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_telegram'
|
||||
|
||||
# Default Notify Format
|
||||
notify_format = NotifyFormat.HTML
|
||||
|
||||
# Telegram uses the http protocol with JSON requests
|
||||
notify_url = 'https://api.telegram.org/bot'
|
||||
|
||||
@ -543,12 +546,11 @@ class NotifyTelegram(NotifyBase):
|
||||
payload['parse_mode'] = 'MARKDOWN'
|
||||
|
||||
payload['text'] = '{}{}'.format(
|
||||
'{}\r\n'.format(title) if title else '',
|
||||
'# {}\r\n'.format(title) if title else '',
|
||||
body,
|
||||
)
|
||||
|
||||
else: # HTML or TEXT
|
||||
|
||||
else: # TEXT or HTML
|
||||
# Use Telegram's HTML mode
|
||||
payload['parse_mode'] = 'HTML'
|
||||
|
||||
@ -592,6 +594,7 @@ class NotifyTelegram(NotifyBase):
|
||||
mo.string[mo.start():mo.end()][1:5]], title)
|
||||
|
||||
if self.notify_format == NotifyFormat.TEXT:
|
||||
# Further html escaping required...
|
||||
telegram_escape_text_dict = {
|
||||
# We need to escape characters that conflict with html
|
||||
# entity blocks (< and >) when displaying text
|
||||
@ -615,8 +618,9 @@ class NotifyTelegram(NotifyBase):
|
||||
lambda mo: telegram_escape_text_dict[
|
||||
mo.string[mo.start():mo.end()]], title)
|
||||
|
||||
# prepare our payload based on HTML or TEXT
|
||||
payload['text'] = '{}{}'.format(
|
||||
'<b>{}</b>\r\n'.format(title) if title else '',
|
||||
'<h1>{}</h1>'.format(title) if title else '',
|
||||
body,
|
||||
)
|
||||
|
||||
|
||||
@ -250,6 +250,11 @@ def apprise_test(do_notify):
|
||||
assert do_notify(a, title='', body=None) is False
|
||||
assert do_notify(a, title=None, body='') is False
|
||||
|
||||
assert do_notify(a, title=5, body=b'bytes') is False
|
||||
assert do_notify(a, title=b"bytes", body=10) is False
|
||||
assert do_notify(a, title=object(), body=b'bytes') is False
|
||||
assert do_notify(a, title=b"bytes", body=object()) is False
|
||||
|
||||
# As long as one is present, we're good
|
||||
assert do_notify(a, title=None, body='present') is True
|
||||
assert do_notify(a, title='present', body=None) is True
|
||||
|
||||
@ -40,7 +40,7 @@ def test_html_to_text():
|
||||
"""
|
||||
A function to simply html conversion tests
|
||||
"""
|
||||
return convert_between(NotifyFormat.HTML, NotifyFormat.TEXT, body)
|
||||
return convert_between(NotifyFormat.HTML, NotifyFormat.TEXT, body)[1]
|
||||
|
||||
assert to_html("No HTML code here.") == "No HTML code here."
|
||||
|
||||
|
||||
@ -183,10 +183,37 @@ def test_apprise_escaping_py3(mock_post):
|
||||
title=object(), body=False, interpret_escapes=True) is False
|
||||
assert a.notify(
|
||||
title=False, body=object(), interpret_escapes=True) is False
|
||||
|
||||
# We support bytes
|
||||
assert a.notify(
|
||||
title=b'byte title', body=b'byte body',
|
||||
interpret_escapes=True) is True
|
||||
|
||||
# However they're escaped as 'utf-8' by default unless we tell Apprise
|
||||
# otherwise
|
||||
# Now test hebrew types (outside of default utf-8)
|
||||
# כותרת נפלאה translates to 'A wonderful title'
|
||||
# זו הודעה translates to 'This is a notification'
|
||||
title = 'כותרת נפלאה'.encode('ISO-8859-8')
|
||||
body = '[_[זו הודעה](http://localhost)_'.encode('ISO-8859-8')
|
||||
assert a.notify(
|
||||
title=title, body=body,
|
||||
interpret_escapes=True) is False
|
||||
|
||||
# However if we let Apprise know in advance the encoding, it will handle
|
||||
# it for us
|
||||
asset = apprise.AppriseAsset(encoding='ISO-8859-8')
|
||||
a = apprise.Apprise(asset=asset)
|
||||
# Create ourselves a test object to work with
|
||||
a.add('json://localhost')
|
||||
assert a.notify(
|
||||
title=title, body=body,
|
||||
interpret_escapes=True) is True
|
||||
|
||||
# We'll restore our configuration back to how it was now
|
||||
a = apprise.Apprise()
|
||||
a.add('json://localhost')
|
||||
|
||||
# The body is proessed first, so the errors thrown above get tested on
|
||||
# the body only. Now we run similar tests but only make the title
|
||||
# bad and always mark the body good
|
||||
@ -198,8 +225,9 @@ def test_apprise_escaping_py3(mock_post):
|
||||
title=object(), body="valid", interpret_escapes=True) is False
|
||||
assert a.notify(
|
||||
title=False, body="valid", interpret_escapes=True) is True
|
||||
# Bytes are supported
|
||||
assert a.notify(
|
||||
title=b'byte title', body="valid", interpret_escapes=True) is False
|
||||
title=b'byte title', body="valid", interpret_escapes=True) is True
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info.major >= 3, reason="Requires Python 2.x+")
|
||||
@ -311,6 +339,8 @@ def test_apprise_escaping_py2(mock_post):
|
||||
title=None, body=4, interpret_escapes=True) is False
|
||||
assert a.notify(
|
||||
title=4, body=None, interpret_escapes=True) is False
|
||||
assert a.notify(
|
||||
title=4, body="valid body", interpret_escapes=True) is False
|
||||
assert a.notify(
|
||||
title=object(), body=False, interpret_escapes=True) is False
|
||||
assert a.notify(
|
||||
|
||||
@ -25,6 +25,7 @@
|
||||
|
||||
import os
|
||||
import six
|
||||
import sys
|
||||
import pytest
|
||||
import mock
|
||||
import requests
|
||||
@ -34,6 +35,7 @@ from apprise import Apprise
|
||||
from apprise import AppriseAttachment
|
||||
from apprise import AppriseAsset
|
||||
from apprise import NotifyType
|
||||
from apprise import NotifyFormat
|
||||
from apprise import plugins
|
||||
from helpers import AppriseURLTester
|
||||
|
||||
@ -211,20 +213,22 @@ def test_plugin_telegram_urls():
|
||||
|
||||
"""
|
||||
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyTelegram.request_rate_per_sec = 0
|
||||
|
||||
# Run our general tests
|
||||
AppriseURLTester(tests=apprise_url_tests).run_all()
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_plugin_telegram_general(mock_post, mock_get):
|
||||
def test_plugin_telegram_general(mock_post):
|
||||
"""
|
||||
NotifyTelegram() General Tests
|
||||
|
||||
"""
|
||||
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyBase.request_rate_per_sec = 0
|
||||
plugins.NotifyTelegram.request_rate_per_sec = 0
|
||||
|
||||
# Bot Token
|
||||
bot_token = '123456789:abcdefg_hijklmnop'
|
||||
@ -234,11 +238,8 @@ def test_plugin_telegram_general(mock_post, mock_get):
|
||||
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
|
||||
@ -246,20 +247,17 @@ def test_plugin_telegram_general(mock_post, mock_get):
|
||||
plugins.NotifyTelegram(bot_token=None, targets=chat_ids)
|
||||
|
||||
# Invalid JSON while trying to detect bot owner
|
||||
mock_get.return_value.content = '{'
|
||||
mock_post.return_value.content = '}'
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
obj.notify(title='hello', body='world')
|
||||
|
||||
# Invalid JSON while trying to detect bot owner + 400 error
|
||||
mock_get.return_value.status_code = requests.codes.internal_server_error
|
||||
mock_post.return_value.status_code = requests.codes.internal_server_error
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
obj.notify(title='hello', body='world')
|
||||
|
||||
# Return status back to how they were
|
||||
mock_post.return_value.status_code = requests.codes.ok
|
||||
mock_get.return_value.status_code = requests.codes.ok
|
||||
|
||||
# Exception should be thrown about the fact an invalid bot token was
|
||||
# specifed
|
||||
@ -280,9 +278,7 @@ def test_plugin_telegram_general(mock_post, mock_get):
|
||||
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
|
||||
@ -304,7 +300,6 @@ def test_plugin_telegram_general(mock_post, mock_get):
|
||||
response.content = dumps({
|
||||
'description': 'test',
|
||||
})
|
||||
mock_get.return_value = response
|
||||
mock_post.return_value = response
|
||||
|
||||
# No image asset
|
||||
@ -410,10 +405,9 @@ def test_plugin_telegram_general(mock_post, mock_get):
|
||||
assert mock_post.call_count == 1
|
||||
payload = loads(mock_post.call_args_list[0][1]['data'])
|
||||
|
||||
# Our special characters are escaped properly
|
||||
# Test our payload
|
||||
assert payload['text'] == \
|
||||
'<b>special characters</b>\r\n<p>'\
|
||||
'\'"This can\'t\t\r\nfail us"\'</p>'
|
||||
'<h1>special characters</h1><p>\'"This can\'t\t\r\nfail us"\'</p>'
|
||||
|
||||
# Test sending attachments
|
||||
attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'))
|
||||
@ -545,3 +539,415 @@ def test_plugin_telegram_general(mock_post, mock_get):
|
||||
assert isinstance(obj, plugins.NotifyTelegram)
|
||||
assert len(obj.targets) == 1
|
||||
assert '-123456789525' in obj.targets
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info.major <= 2, reason="Requires Python 3.x+")
|
||||
@mock.patch('requests.post')
|
||||
def test_plugin_telegram_formating_py3(mock_post):
|
||||
"""
|
||||
NotifyTelegram() Python v3+ Formatting Tests
|
||||
|
||||
"""
|
||||
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyTelegram.request_rate_per_sec = 0
|
||||
|
||||
# Prepare Mock
|
||||
mock_post.return_value = requests.Request()
|
||||
mock_post.return_value.status_code = requests.codes.ok
|
||||
mock_post.return_value.content = '{}'
|
||||
|
||||
# Simple success response
|
||||
mock_post.return_value.content = dumps({
|
||||
"ok": True,
|
||||
"result": [{
|
||||
"update_id": 645421321,
|
||||
"message": {
|
||||
"message_id": 2,
|
||||
"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
|
||||
|
||||
results = plugins.NotifyTelegram.parse_url(
|
||||
'tgram://123456789:abcdefg_hijklmnop/')
|
||||
|
||||
instance = plugins.NotifyTelegram(**results)
|
||||
assert isinstance(instance, plugins.NotifyTelegram)
|
||||
|
||||
response = instance.send(title='title', body='body')
|
||||
assert response is True
|
||||
# 1 call to look up bot owner, and second for notification
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Now test our HTML Conversion as TEXT)
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/')
|
||||
assert len(aobj) == 1
|
||||
|
||||
title = '🚨 Change detected for <i>Apprise Test Title</i>'
|
||||
body = '<a href="http://localhost"><i>Apprise Body Title</i></a>' \
|
||||
' had <a href="http://127.0.0.1">a change</a>'
|
||||
|
||||
assert aobj.notify(title=title, body=body, body_format=NotifyFormat.TEXT)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a TEXT mode
|
||||
assert payload['text'] == \
|
||||
'<h1>🚨 Change detected for <i>Apprise Test Title</i>' \
|
||||
'</h1><a href="http://localhost"><i>Apprise Body Title' \
|
||||
'</i></a> had <a href="http://127.0.0.1">a change' \
|
||||
'</a>'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Now test our HTML Conversion as TEXT)
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/?format=html')
|
||||
assert len(aobj) == 1
|
||||
|
||||
assert aobj.notify(title=title, body=body, body_format=NotifyFormat.HTML)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a HTML mode
|
||||
assert payload['text'] == \
|
||||
'<h1>🚨 Change detected for <i>Apprise Test Title</i></h1>' \
|
||||
'<a href="http://localhost"><i>Apprise Body Title</i></a> had ' \
|
||||
'<a href="http://127.0.0.1">a change</a>'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Now test our MARKDOWN Handling
|
||||
title = '# 🚨 Change detected for _Apprise Test Title_'
|
||||
body = '_[Apprise Body Title](http://localhost)_' \
|
||||
' had [a change](http://127.0.0.1)'
|
||||
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/?format=markdown')
|
||||
assert len(aobj) == 1
|
||||
|
||||
assert aobj.notify(
|
||||
title=title, body=body, body_format=NotifyFormat.MARKDOWN)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a HTML mode
|
||||
assert payload['text'] == \
|
||||
'# 🚨 Change detected for _Apprise Test Title_\r\n' \
|
||||
'_[Apprise Body Title](http://localhost)_ had ' \
|
||||
'[a change](http://127.0.0.1)'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Upstream to use HTML but input specified as Markdown
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://987654321:abcdefg_hijklmnop/?format=html')
|
||||
assert len(aobj) == 1
|
||||
|
||||
# HTML forced by the command line, but MARKDOWN spacified as
|
||||
# upstream mode
|
||||
assert aobj.notify(
|
||||
title=title, body=body, body_format=NotifyFormat.MARKDOWN)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot987654321:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot987654321:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a HTML mode
|
||||
assert payload['text'] == \
|
||||
'<h1><p>🚨 Change detected for <em>Apprise Test Title</em></p>' \
|
||||
'</h1><p><em><a href="http://localhost">Apprise Body Title</a></em> ' \
|
||||
'had <a href="http://127.0.0.1">a change</a></p>'
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info.major >= 3, reason="Requires Python 2.x+")
|
||||
@mock.patch('requests.post')
|
||||
def test_plugin_telegram_formating_py2(mock_post):
|
||||
"""
|
||||
NotifyTelegram() Python v2 Formatting Tests
|
||||
|
||||
"""
|
||||
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyTelegram.request_rate_per_sec = 0
|
||||
|
||||
# Prepare Mock
|
||||
mock_post.return_value = requests.Request()
|
||||
mock_post.return_value.status_code = requests.codes.ok
|
||||
mock_post.return_value.content = '{}'
|
||||
|
||||
# Simple success response
|
||||
mock_post.return_value.content = dumps({
|
||||
"ok": True,
|
||||
"result": [{
|
||||
"update_id": 645421321,
|
||||
"message": {
|
||||
"message_id": 2,
|
||||
"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
|
||||
|
||||
results = plugins.NotifyTelegram.parse_url(
|
||||
'tgram://123456789:abcdefg_hijklmnop/')
|
||||
|
||||
instance = plugins.NotifyTelegram(**results)
|
||||
assert isinstance(instance, plugins.NotifyTelegram)
|
||||
|
||||
response = instance.send(title='title', body='body')
|
||||
assert response is True
|
||||
# 1 call to look up bot owner, and second for notification
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Now test our HTML Conversion as TEXT)
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/')
|
||||
assert len(aobj) == 1
|
||||
|
||||
title = '🚨 Change detected for <i>Apprise Test Title</i>'
|
||||
body = '<a href="http://localhost"><i>Apprise Body Title</i></a>' \
|
||||
' had <a href="http://127.0.0.1">a change</a>'
|
||||
|
||||
assert aobj.notify(title=title, body=body, body_format=NotifyFormat.TEXT)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a TEXT mode
|
||||
assert payload['text'].encode('utf-8') == \
|
||||
'<h1>\xf0\x9f\x9a\xa8 Change detected for <i>' \
|
||||
'Apprise Test Title</i></h1>' \
|
||||
'<a href="http://localhost"><i>' \
|
||||
'Apprise Body Title</i></a> had <a ' \
|
||||
'href="http://127.0.0.1">a change</a>'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Now test our HTML Conversion as TEXT)
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/?format=html')
|
||||
assert len(aobj) == 1
|
||||
|
||||
assert aobj.notify(title=title, body=body, body_format=NotifyFormat.HTML)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a HTML mode
|
||||
assert payload['text'].encode('utf-8') == \
|
||||
'<h1>\xf0\x9f\x9a\xa8 Change detected for <i>Apprise Test Title</i>' \
|
||||
'</h1><a href="http://localhost"><i>Apprise Body Title</i></a> had ' \
|
||||
'<a href="http://127.0.0.1">a change</a>'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Now test our MARKDOWN Handling
|
||||
title = '# 🚨 Change detected for _Apprise Test Title_'
|
||||
body = '_[Apprise Body Title](http://localhost)_' \
|
||||
' had [a change](http://127.0.0.1)'
|
||||
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/?format=markdown')
|
||||
assert len(aobj) == 1
|
||||
|
||||
assert aobj.notify(
|
||||
title=title, body=body, body_format=NotifyFormat.MARKDOWN)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a HTML mode
|
||||
assert payload['text'].encode('utf-8') == \
|
||||
'# \xf0\x9f\x9a\xa8 Change detected for _Apprise Test Title_\r\n_' \
|
||||
'[Apprise Body Title](http://localhost)_ had ' \
|
||||
'[a change](http://127.0.0.1)'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Upstream to use HTML but input specified as Markdown
|
||||
aobj = Apprise()
|
||||
aobj.add('tgram://987654321:abcdefg_hijklmnop/?format=html')
|
||||
assert len(aobj) == 1
|
||||
|
||||
# HTML forced by the command line, but MARKDOWN spacified as
|
||||
# upstream mode
|
||||
assert aobj.notify(
|
||||
title=title, body=body, body_format=NotifyFormat.MARKDOWN)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot987654321:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot987654321:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a HTML mode
|
||||
assert payload['text'].encode('utf-8') == \
|
||||
'<h1><p>\xf0\x9f\x9a\xa8 Change detected for ' \
|
||||
'<em>Apprise Test Title</em></p></h1><p><em>' \
|
||||
'<a href="http://localhost">Apprise Body Title</a></em>' \
|
||||
' had <a href="http://127.0.0.1">a change</a></p>'
|
||||
|
||||
# Reset our values
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Now test hebrew types (outside of default utf-8)
|
||||
# כותרת נפלאה translates to 'A wonderful title'
|
||||
# זו הודעה translates to 'This is a notification'
|
||||
title = 'כותרת נפלאה' \
|
||||
.decode('utf-8').encode('ISO-8859-8')
|
||||
body = '[_[זו הודעה](http://localhost)_' \
|
||||
.decode('utf-8').encode('ISO-8859-8')
|
||||
|
||||
asset = AppriseAsset(encoding='utf-8')
|
||||
# Now test default entries
|
||||
aobj = Apprise(asset=asset)
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/')
|
||||
assert len(aobj) == 1
|
||||
|
||||
# Our notification will fail because we'll have an encoding error
|
||||
assert not aobj.notify(title=title, body=body)
|
||||
# Nothing was even attempted to be notified
|
||||
assert mock_post.call_count == 0
|
||||
|
||||
# Let's use the expected input
|
||||
asset = AppriseAsset(encoding='ISO-8859-8')
|
||||
|
||||
# Now test default entries
|
||||
aobj = Apprise(asset=asset)
|
||||
|
||||
aobj.add('tgram://123456789:abcdefg_hijklmnop/')
|
||||
assert len(aobj) == 1
|
||||
|
||||
# Our notification will work now
|
||||
assert aobj.notify(title=title, body=body)
|
||||
|
||||
# Test our calls
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/getUpdates'
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.telegram.org/bot123456789:abcdefg_hijklmnop/sendMessage'
|
||||
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
# Test that everything is escaped properly in a HTML mode
|
||||
assert payload['text'].encode('utf-8') == \
|
||||
'<h1>\xd7\x9b\xd7\x95\xd7\xaa\xd7\xa8\xd7\xaa '\
|
||||
'\xd7\xa0\xd7\xa4\xd7\x9c\xd7\x90\xd7\x94</h1>[_[\xd7\x96\xd7\x95 '\
|
||||
'\xd7\x94\xd7\x95\xd7\x93\xd7\xa2\xd7\x94](http://localhost)_'
|
||||
|
||||
Loading…
Reference in New Issue
Block a user