mirror of https://github.com/caronc/apprise
Telegram bot owner detection moved from init() to send() (#274)
parent
aa039e74b2
commit
ca695ed398
|
@ -229,23 +229,19 @@ class NotifyTelegram(NotifyBase):
|
|||
# Parse our list
|
||||
self.targets = parse_list(targets)
|
||||
|
||||
# if detect_owner is set to True, we will attempt to determine who
|
||||
# the bot owner is based on the first person who messaged it. This
|
||||
# is not a fool proof way of doing things as over time Telegram removes
|
||||
# the message history for the bot. So what appears (later on) to be
|
||||
# the first message to it, maybe another user who sent it a message
|
||||
# much later. Users who set this flag should update their Apprise
|
||||
# URL later to directly include the user that we should message.
|
||||
self.detect_owner = detect_owner
|
||||
|
||||
if self.user:
|
||||
# Treat this as a channel too
|
||||
self.targets.append(self.user)
|
||||
|
||||
if len(self.targets) == 0 and self.detect_owner:
|
||||
_id = self.detect_bot_owner()
|
||||
if _id:
|
||||
# Store our id
|
||||
self.targets.append(str(_id))
|
||||
|
||||
if len(self.targets) == 0:
|
||||
err = 'No chat_id(s) were specified.'
|
||||
self.logger.warning(err)
|
||||
raise TypeError(err)
|
||||
|
||||
# Track whether or not we want to send an image with our notification
|
||||
# or not.
|
||||
self.include_image = include_image
|
||||
|
@ -474,7 +470,7 @@ class NotifyTelegram(NotifyBase):
|
|||
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)' % (
|
||||
self.logger.info('Detected Telegram user %s (userid=%d)' % (
|
||||
_user, _id))
|
||||
# Return our detected userid
|
||||
return _id
|
||||
|
@ -490,6 +486,19 @@ class NotifyTelegram(NotifyBase):
|
|||
Perform Telegram Notification
|
||||
"""
|
||||
|
||||
if len(self.targets) == 0 and self.detect_owner:
|
||||
_id = self.detect_bot_owner()
|
||||
if _id:
|
||||
# Permanently store our id in our target list for next time
|
||||
self.targets.append(str(_id))
|
||||
self.logger.info(
|
||||
'Update your Telegram Apprise URL to read: '
|
||||
'{}'.format(self.url(privacy=True)))
|
||||
|
||||
if len(self.targets) == 0:
|
||||
self.logger.warning('There were not Telegram chat_ids to notify.')
|
||||
return False
|
||||
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
'Content-Type': 'application/json',
|
||||
|
|
|
@ -73,14 +73,14 @@ def test_notify_telegram_plugin(mock_post, mock_get):
|
|||
# Invalid JSON while trying to detect bot owner
|
||||
mock_get.return_value.content = '{'
|
||||
mock_post.return_value.content = '}'
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
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
|
||||
with pytest.raises(TypeError):
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
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
|
||||
|
@ -218,6 +218,10 @@ def test_notify_telegram_plugin(mock_post, mock_get):
|
|||
attach=path) is False
|
||||
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
# No user detected; this happens after our firsst notification
|
||||
assert len(obj.targets) == 0
|
||||
|
||||
assert obj.notify(title='hello', body='world') is True
|
||||
assert len(obj.targets) == 1
|
||||
assert obj.targets[0] == '532389719'
|
||||
|
||||
|
@ -227,9 +231,12 @@ def test_notify_telegram_plugin(mock_post, mock_get):
|
|||
"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)
|
||||
# No user will be detected now
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
# No user detected; this happens after our firsst notification
|
||||
assert len(obj.targets) == 0
|
||||
assert obj.notify(title='hello', body='world') is False
|
||||
assert len(obj.targets) == 0
|
||||
|
||||
# Detect the bot with a bad response
|
||||
mock_post.return_value.content = dumps({})
|
||||
|
@ -238,22 +245,25 @@ def test_notify_telegram_plugin(mock_post, mock_get):
|
|||
# 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)
|
||||
# internal server error prevents notification from being sent
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
assert len(obj.targets) == 0
|
||||
assert obj.notify(title='hello', body='world') is False
|
||||
assert len(obj.targets) == 0
|
||||
|
||||
# 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)
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
assert len(obj.targets) == 0
|
||||
assert obj.notify(title='hello', body='world') is False
|
||||
assert len(obj.targets) == 0
|
||||
|
||||
# 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)
|
||||
plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
assert len(obj.targets) == 0
|
||||
assert obj.notify(title='hello', body='world') is False
|
||||
assert len(obj.targets) == 0
|
||||
|
||||
# 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
|
||||
|
@ -265,5 +275,7 @@ def test_notify_telegram_plugin(mock_post, mock_get):
|
|||
mock_post.side_effect = requests.HTTPError
|
||||
|
||||
# No chat_ids specified
|
||||
with pytest.raises(TypeError):
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
obj = plugins.NotifyTelegram(bot_token=bot_token, targets=None)
|
||||
assert len(obj.targets) == 0
|
||||
assert obj.notify(title='hello', body='world') is False
|
||||
assert len(obj.targets) == 0
|
||||
|
|
Loading…
Reference in New Issue