extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-11-24 17:14:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fix Telegram Thread/Topic handling (#1177)

Browse Source 
* Fix Telegram Thread/Topic handling

* fixed test coverage
This commit is contained in:
Chris Caron 2024-07-28 12:23:59 -04:00 committed by GitHub
parent 65f9caed87
commit 47540523e0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 128 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
apprise/plugins/telegram.py
View File

@ -361,6 +361,9 @@ class NotifyTelegram(NotifyBase):
'name': _('Topic Thread ID'),
'type': 'int',
},
'thread': {
'alias_of': 'topic',
},
'mdv': {
'name': _('Markdown Version'),
'type': 'choice:string',
@ -460,14 +463,13 @@ class NotifyTelegram(NotifyBase):
self.detect_owner = False
continue
try:
if results.group('topic'):
topic = int(
results.group('topic')
if results.group('topic') else self.topic)
except TypeError:
# No worries
topic = None
else:
# Default (if one set)
topic = self.topic
if results.group('name') is not None:
# Name
@ -730,7 +732,7 @@ class NotifyTelegram(NotifyBase):
_id = self.detect_bot_owner()
if _id:
# Permanently store our id in our target list for next time
self.targets.append((str(_id), None))
self.targets.append((str(_id), self.topic))
self.logger.info(
'Update your Telegram Apprise URL to read: '
'{}'.format(self.url(privacy=True)))
@ -1043,6 +1045,9 @@ class NotifyTelegram(NotifyBase):
if 'topic' in results['qsd'] and len(results['qsd']['topic']):
results['topic'] = results['qsd']['topic']
elif 'thread' in results['qsd'] and len(results['qsd']['thread']):
results['topic'] = results['qsd']['thread']
# Silent (Sends the message Silently); users will receive
# notification with no sound.
results['silent'] = \

121
test/test_plugin_telegram.py
View File

@ -98,6 +98,10 @@ apprise_url_tests = (
('tgram://bottest@123456789:abcdefg_hijklmnop/id1/?topic=12345', {
'instance': NotifyTelegram,
}),
# Thread is just an alias of topic
('tgram://bottest@123456789:abcdefg_hijklmnop/id1/?thread=12345', {
'instance': NotifyTelegram,
}),
# Threads must be numeric
('tgram://bottest@123456789:abcdefg_hijklmnop/id1/?topic=invalid', {
'instance': TypeError,
@ -106,6 +110,11 @@ apprise_url_tests = (
('tgram://bottest@123456789:abcdefg_hijklmnop/id1/?content=invalid', {
'instance': TypeError,
}),
('tgram://bottest@123456789:abcdefg_hijklmnop/id1:invalid/?thread=12345', {
'instance': NotifyTelegram,
# Notify will fail (bad target)
'response': False,
}),
# Testing image
('tgram://123456789:abcdefg_hijklmnop/lead2gold/?image=Yes', {
'instance': NotifyTelegram,
@ -907,12 +916,9 @@ def test_plugin_telegram_html_formatting(mock_post):
NotifyTelegram() HTML Formatting
"""
# on't send anything other than <b>, <i>, <a>,<code> and <pre>
# 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({
@ -942,7 +948,6 @@ def test_plugin_telegram_html_formatting(mock_post):
}},
],
})
mock_post.return_value.status_code = requests.codes.ok
aobj = Apprise()
aobj.add('tgram://123456789:abcdefg_hijklmnop/')
@ -1005,3 +1010,111 @@ def test_plugin_telegram_html_formatting(mock_post):
'<b>Heading 3</b>\r\n<b>Heading 4</b>\r\n<b>Heading 5</b>\r\n' \
'<b>Heading 6</b>\r\nA set of text\r\n' \
'Another line after the set of text\r\nMore text\r\nlabel'
@mock.patch('requests.post')
def test_plugin_telegram_threads(mock_post):
"""
NotifyTelegram() Threads/Topics
"""
# Prepare Mock
mock_post.return_value = requests.Request()
mock_post.return_value.status_code = requests.codes.ok
# 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",
}],
}},
],
})
aobj = Apprise()
aobj.add('tgram://123456789:abcdefg_hijklmnop/?thread=1234')
assert len(aobj) == 1
assert isinstance(aobj[0], NotifyTelegram)
body = 'my message'
assert aobj.notify(body=body)
# 1 call to look up bot owner, and second for notification
assert mock_post.call_count == 2
payload = loads(mock_post.call_args_list[1][1]['data'])
assert 'message_thread_id' in payload
assert payload['message_thread_id'] == 1234
mock_post.reset_mock()
aobj = Apprise()
aobj.add('tgram://123456789:abcdefg_hijklmnop/?topic=1234')
assert len(aobj) == 1
assert isinstance(aobj[0], NotifyTelegram)
body = 'my message'
assert aobj.notify(body=body)
# 1 call to look up bot owner, and second for notification
assert mock_post.call_count == 2
payload = loads(mock_post.call_args_list[1][1]['data'])
assert 'message_thread_id' in payload
assert payload['message_thread_id'] == 1234
mock_post.reset_mock()
aobj = Apprise()
aobj.add('tgram://123456789:abcdefg_hijklmnop/9876:1234/9876:1111')
assert len(aobj) == 1
assert isinstance(aobj[0], NotifyTelegram)
body = 'my message'
assert aobj.notify(body=body)
# 1 call to look up bot owner, and second for notification
assert mock_post.call_count == 2
payload = loads(mock_post.call_args_list[0][1]['data'])
assert 'message_thread_id' in payload
assert payload['message_thread_id'] == 1111
payload = loads(mock_post.call_args_list[1][1]['data'])
assert 'message_thread_id' in payload
assert payload['message_thread_id'] == 1234
mock_post.reset_mock()