github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity

persistent storage added

pull/1192/head
Chris Caron 2024-08-31 20:46:12 -04:00
parent 70b6803005
commit c1fb2b4072
2 changed files with 196 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

107
apprise/plugins/sendpulse.py
View File

@ -38,6 +38,7 @@ from .base import NotifyBase
from .. import exception
from ..common import NotifyFormat
from ..common import NotifyType
from ..common import PersistentStoreMode
from ..utils import is_email
from ..utils import parse_emails
from ..conversion import convert_between
@ -79,6 +80,18 @@ class NotifySendPulse(NotifyBase):
# 60/300 = 0.2
request_rate_per_sec = 0.2
# Our default is to no not use persistent storage beyond in-memory
# reference
storage_mode = PersistentStoreMode.AUTO
# Token expiry if not detected in seconds (below is 1 hr)
token_expiry = 3600
# The number of seconds to grace for early token renewal
# Below states that 10 seconds bfore our token expiry, we'll
# attempt to renew it
token_expiry_edge = 10
# Support attachments
attachment_support = True
@ -174,9 +187,6 @@ class NotifySendPulse(NotifyBase):
"""
super().__init__(**kwargs)
# Api Key is acquired upon a sucessful login
self.access_token = None
# For tracking our email -> name lookups
self.names = {}
@ -395,31 +405,58 @@ class NotifySendPulse(NotifyBase):
"""
return len(self.targets)
def login(self):
"""
Authenticates with the server to get a access_token
"""
self.store.clear('access_token')
payload = {
'grant_type': 'client_credentials',
'client_id': self.client_id,
'client_secret': self.client_secret,
}
success, response = self._fetch(self.notify_oauth_url, payload)
if not success:
return False
access_token = response.get('access_token')
# If we get here, we're authenticated
try:
expires = \
int(response.get('expires_in')) - self.token_expiry_edge
if expires <= self.token_expiry_edge:
self.logger.error(
'SendPulse token expiry limit returned was invalid')
return False
elif expires > self.token_expiry:
self.logger.warning(
'SendPulse token expiry limit fixed to: {}s'
.format(self.token_expiry))
expires = self.token_expiry - self.token_expiry_edge
except (AttributeError, TypeError, ValueError):
# expires_in was not an integer
self.logger.warning(
'SendPulse token expiry limit presumed to be: {}s'.format(
self.token_expiry))
expires = self.token_expiry - self.token_expiry_edge
self.store.set('access_token', access_token, expires=expires)
return access_token
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
**kwargs):
"""
Perform SendPulse Notification
"""
if not self.access_token:
# Attempt to acquire acquire a login
_payload = {
'grant_type': 'client_credentials',
'client_id': self.client_id,
'client_secret': self.client_secret,
}
success, response = self._fetch(self.notify_oauth_url, _payload)
if not success:
return False
# If we get here, we're authenticated
self.access_token = response.get('access_token')
# Store our bearer
extended_headers = {
'Authorization': 'Bearer {}'.format(self.access_token),
}
access_token = self.store.get('access_token') or self.login()
if not access_token:
return False
# error tracking (used for function return)
has_error = False
@ -543,14 +580,14 @@ class NotifySendPulse(NotifyBase):
# Perform our post
success, response = self._fetch(
self.notify_email_url, payload, target, extended_headers)
self.notify_email_url, payload, target, retry=1)
if not success:
has_error = True
continue
return not has_error
def _fetch(self, url, payload, target=None, extended_headers={}):
def _fetch(self, url, payload, target=None, retry=0):
"""
Wrapper to request.post() to manage it's response better and make
the send() function cleaner and easier to maintain.
@ -558,14 +595,14 @@ class NotifySendPulse(NotifyBase):
This function returns True if the _post was successful and False
if it wasn't.
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/json',
}
if extended_headers:
headers.update(extended_headers)
access_token = self.store.get('access_token')
if access_token:
headers.update({'Authorization': f'Bearer {access_token}'})
self.logger.debug('SendPulse POST URL: %s (cert_verify=%r)' % (
url, self.verify_certificate,
@ -599,12 +636,20 @@ class NotifySendPulse(NotifyBase):
'Response Details:\r\n{}'.format(r.content))
return (False, {})
if r.status_code not in (
# Reference status code
status_code = r.status_code
if status_code == requests.codes.unauthorized:
# Key likely expired, we'll reset it and try one more time
if retry and self.login():
return self._fetch(url, payload, target, retry=retry - 1)
if status_code not in (
requests.codes.ok, requests.codes.accepted):
# We had a problem
status_str = \
NotifySendPulse.http_response_code_lookup(
r.status_code)
status_code)
if target:
self.logger.warning(
@ -613,14 +658,14 @@ class NotifySendPulse(NotifyBase):
target,
status_str,
', ' if status_str else '',
r.status_code))
status_code))
else:
self.logger.warning(
'SendPulse Authentication Request failed: '
'{}{}error={}.'.format(
status_str,
', ' if status_str else '',
r.status_code))
status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))

121
test/test_plugin_sendpulse.py
View File

@ -44,7 +44,8 @@ import logging
logging.disable(logging.CRITICAL)
SENDPULSE_GOOD_RESPONSE = dumps({
"access_token": 'abc123'
"access_token": 'abc123',
"expires_in": 3600,
})
SENDPULSE_BAD_RESPONSE = '{'
@ -353,6 +354,124 @@ def test_plugin_sendpulse_edge_cases(mock_post):
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is False
# Test re-authentication
mock_post.reset_mock()
request = mock.Mock()
obj = Apprise.instantiate('sendpulse://usr2@example.com/ci/cs/?from=Retry')
class sendpulse():
def __init__(self):
# 200 login okay
# 401 on retrival
# recursive re-attempt to login returns 200
# fetch after works
self._side_effect = iter([
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.ok,
])
@property
def status_code(self):
return next(self._side_effect)
@property
def content(self):
return SENDPULSE_GOOD_RESPONSE
mock_post.return_value = sendpulse()
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
assert mock_post.call_count == 4
# Authentication
assert mock_post.call_args_list[0][0][0] == \
'https://api.sendpulse.com/oauth/access_token'
# 401 received
assert mock_post.call_args_list[1][0][0] == \
'https://api.sendpulse.com/smtp/emails'
# Re-authenticate
assert mock_post.call_args_list[2][0][0] == \
'https://api.sendpulse.com/oauth/access_token'
# Try again
assert mock_post.call_args_list[3][0][0] == \
'https://api.sendpulse.com/smtp/emails'
# Test re-authentication (no recursive loops)
mock_post.reset_mock()
request = mock.Mock()
obj = Apprise.instantiate('sendpulse://usr2@example.com/ci/cs/?from=Retry')
class sendpulse():
def __init__(self):
# oauth always returns okay but notify returns 401
# recursive re-attempt only once
self._side_effect = iter([
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.unauthorized,
requests.codes.ok, requests.codes.unauthorized,
])
@property
def status_code(self):
return next(self._side_effect)
@property
def content(self):
return SENDPULSE_GOOD_RESPONSE
mock_post.return_value = sendpulse()
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is False
assert mock_post.call_count == 4
# Authentication
assert mock_post.call_args_list[0][0][0] == \
'https://api.sendpulse.com/oauth/access_token'
# 401 received
assert mock_post.call_args_list[1][0][0] == \
'https://api.sendpulse.com/smtp/emails'
# Re-authenticate
assert mock_post.call_args_list[2][0][0] == \
'https://api.sendpulse.com/oauth/access_token'
# Last failed attempt
assert mock_post.call_args_list[3][0][0] == \
'https://api.sendpulse.com/smtp/emails'
mock_post.side_effect = None
request = mock.Mock()
request.status_code = requests.codes.ok
request.content = SENDPULSE_GOOD_RESPONSE
mock_post.return_value = request
for expires_in in (None, -1, 'garbage', 3600, 300000):
request.content = dumps({
"access_token": 'abc123',
"expires_in": expires_in,
})
# Instantiate our object
obj = Apprise.instantiate('sendpulse://user@example.com/ci/cs/')
# Test variations of responses
obj.notify(
body='body', title='title', notify_type=NotifyType.INFO)
# expires_in is missing
request.content = dumps({
"access_token": 'abc123',
})
# Instantiate our object
obj = Apprise.instantiate('sendpulse://user@example.com/ci/cs/')
obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
def test_plugin_sendpulse_fail_cases():
"""