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

basic tests passing

pull/1192/head
Chris Caron 2024-08-30 22:26:50 -04:00
parent e5e7ab0784
commit 5e1e4857f1
2 changed files with 183 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

204
apprise/plugins/sendpulse.py
View File

@ -30,7 +30,7 @@
# - https://sendpulse.com/integrations/api/smtp
import requests
from json import dumps
from json import dumps, loads
import base64
from .base import NotifyBase
@ -69,6 +69,9 @@ class NotifySendPulse(NotifyBase):
# The default Email API URL to use
notify_email_url = 'https://api.sendpulse.com/smtp/emails'
# Our OAuth Query
notify_oauth_url = 'https://api.sendpulse.com/oauth/access_token'
# Support attachments
attachment_support = True
@ -167,6 +170,15 @@ class NotifySendPulse(NotifyBase):
"""
super().__init__(**kwargs)
# Api Key is acquired upon a sucessful login
self.access_token = None
result = is_email(from_email)
if not result:
msg = 'Invalid ~From~ email specified: {}'.format(from_email)
self.logger.warning(msg)
raise TypeError(msg)
# Client ID
self.client_id = validate_regex(
client_id, *self.template_tokens['client_id']['regex'])
@ -185,12 +197,6 @@ class NotifySendPulse(NotifyBase):
self.logger.warning(msg)
raise TypeError(msg)
result = is_email(from_email)
if not result:
msg = 'Invalid ~From~ email specified: {}'.format(from_email)
self.logger.warning(msg)
raise TypeError(msg)
# Tracks emails to name lookups (if they exist)
self.__email_map = {}
@ -208,6 +214,8 @@ class NotifySendPulse(NotifyBase):
# Acquire Blind Carbon Copies
self.bcc = set()
# No template
self.template = None
if template:
try:
# Store our template
@ -229,9 +237,9 @@ class NotifySendPulse(NotifyBase):
for recipient in parse_emails(targets):
result = is_email(recipient)
if result:
self.targets.append(
(result['name'] if result['name'] else False,
result['full_email']))
self.targets.append(result['full_email'])
if result['name']:
self.__email_map[result['full_email']] = result['name']
continue
self.logger.warning(
@ -250,7 +258,7 @@ class NotifySendPulse(NotifyBase):
if result:
self.cc.add(result['full_email'])
if result['name']:
self.__email_lookup[result['full_email']] = result['name']
self.__email_map[result['full_email']] = result['name']
continue
self.logger.warning(
@ -265,7 +273,7 @@ class NotifySendPulse(NotifyBase):
if result:
self.bcc.add(result['full_email'])
if result['name']:
self.__email_lookup[result['full_email']] = result['name']
self.__email_map[result['full_email']] = result['name']
continue
self.logger.warning(
@ -300,8 +308,8 @@ class NotifySendPulse(NotifyBase):
# Handle our Carbon Copy Addresses
params['cc'] = ','.join([
formataddr(
(self.__email_lookup[e]
if e in self.__email_lookup else False, e),
(self.__email_map[e]
if e in self.__email_map else False, e),
# Swap comma for it's escaped url code (if detected) since
# we're using that as a delimiter
charset='utf-8').replace(',', '%2C')
@ -311,8 +319,8 @@ class NotifySendPulse(NotifyBase):
# Handle our Blind Carbon Copy Addresses
params['bcc'] = ','.join([
formataddr(
(self.__email_lookup[e]
if e in self.__email_lookup else False, e),
(self.__email_map[e]
if e in self.__email_map else False, e),
# Swap comma for it's escaped url code (if detected) since
# we're using that as a delimiter
charset='utf-8').replace(',', '%2C')
@ -353,10 +361,24 @@ class NotifySendPulse(NotifyBase):
Perform SendPulse Notification
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/json',
'Authorization': 'Bearer {}'.format(self.apikey),
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),
}
# error tracking (used for function return)
@ -451,8 +473,8 @@ class NotifySendPulse(NotifyBase):
to = {
'email': target
}
if target in self.__email_lookup:
to['name'] = self.__email_lookup[target]
if target in self.__email_map:
to['name'] = self.__email_map[target]
# Set our target
payload['email']['to'] = [to]
@ -463,8 +485,8 @@ class NotifySendPulse(NotifyBase):
item = {
'email': email,
}
if email in self.__email_lookup:
item['name'] = self.__email_lookup[email]
if email in self.__email_map:
item['name'] = self.__email_map[email]
payload['email']['cc'].append(item)
@ -474,33 +496,77 @@ class NotifySendPulse(NotifyBase):
item = {
'email': email,
}
if email in self.__email_lookup:
item['name'] = self.__email_lookup[email]
if email in self.__email_map:
item['name'] = self.__email_map[email]
payload['email']['bcc'].append(item)
self.logger.debug('SendPulse POST URL: %s (cert_verify=%r)' % (
self.notify_email_url, self.verify_certificate,
))
self.logger.debug('SendPulse Payload: %s' % str(payload))
# Perform our post
success, response = self._fetch(
self.notify_email_url, payload, target, extended_headers)
if not success:
has_error = True
continue
return not has_error
def _fetch(self, url, payload, target=None, extended_headers={}):
"""
Wrapper to request.post() to manage it's response better and make
the send() function cleaner and easier to maintain.
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)
self.logger.debug('SendPulse POST URL: %s (cert_verify=%r)' % (
self.url, self.verify_certificate,
))
self.logger.debug('SendPulse Payload: %s' % str(payload))
# Prepare our default response
response = {}
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
self.url,
data=dumps(payload),
headers=headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
self.notify_email_url,
data=dumps(payload),
headers=headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code not in (
requests.codes.ok, requests.codes.accepted):
# We had a problem
status_str = \
NotifySendPulse.http_response_code_lookup(
r.status_code)
response = loads(r.content)
except (AttributeError, TypeError, ValueError):
# This gets thrown if we can't parse our JSON Response
# - ValueError = r.content is Unparsable
# - TypeError = r.content is None
# - AttributeError = r is None
self.logger.warning('Invalid response from SendPulse server.')
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
return (False, {})
if r.status_code not in (
requests.codes.ok, requests.codes.accepted):
# We had a problem
status_str = \
NotifySendPulse.http_response_code_lookup(
r.status_code)
if target:
self.logger.warning(
'Failed to send SendPulse notification to {}: '
'{}{}error={}.'.format(
@ -508,29 +574,33 @@ class NotifySendPulse(NotifyBase):
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
# Mark our failure
has_error = True
continue
else:
self.logger.warning(
'SendPulse Authentication Request failed: '
'{}{}error={}.'.format(
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
else:
if target:
self.logger.info(
'Sent SendPulse notification to {}.'.format(target))
else:
self.logger.debug('SendPulse authentication successful')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending SendPulse '
'notification to {}.'.format(target))
self.logger.debug('Socket Exception: %s' % str(e))
return (True, response)
# Mark our failure
has_error = True
continue
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending SendPulse '
'notification to {}.'.format(target))
self.logger.debug('Socket Exception: %s' % str(e))
return not has_error
return (False, response)
@staticmethod
def parse_url(url):
@ -545,6 +615,12 @@ class NotifySendPulse(NotifyBase):
# We're done early as we couldn't load the results
return results
# Define our minimum requirements; defining them now saves us from
# having to if/else all kinds of branches below...
results['from_email'] = None
results['client_id'] = None
results['client_secret'] = None
# Our URL looks like this:
# {schema}://{from_email}:{client_id}/{client_secret}/{targets}
#
@ -564,7 +640,7 @@ class NotifySendPulse(NotifyBase):
results['client_id'] = \
NotifySendPulse.unquote(results['qsd']['id'])
else:
elif results['targets']:
# Store our Client ID
results['client_id'] = results['targets'].pop(0)
@ -573,7 +649,7 @@ class NotifySendPulse(NotifyBase):
results['client_secret'] = \
NotifySendPulse.unquote(results['qsd']['secret'])
else:
elif results['targets']:
# Store our Client Secret
results['client_secret'] = results['targets'].pop(0)

63
test/test_plugin_sendpulse.py
View File

@ -26,6 +26,7 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
from json import dumps
from unittest import mock
import os
@ -42,6 +43,12 @@ from helpers import AppriseURLTester
import logging
logging.disable(logging.CRITICAL)
SENDPULSE_GOOD_RESPONSE = dumps({
"access_token": 'abc123'
})
SENDPULSE_BAD_RESPONSE = '{'
# Attachment Directory
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
@ -54,61 +61,76 @@ apprise_url_tests = (
'instance': None,
}),
('sendpulse://abcd', {
# Just an broken email (no email, client_id or secret)
'instance': None,
# invalid from email
'instance': TypeError,
}),
('sendpulse://abcd@host', {
('sendpulse://abcd@host.com', {
# Just an Email specified, no client_id or client_secret
'instance': None,
'instance': TypeError,
}),
('sendpulse://user@example.com/client_id/client_secret/', {
('sendpulse://user@example.com/client_id/cs1/', {
'instance': NotifySendPulse,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
('sendpulse://user@example.com/client_id/client_secret/'
('sendpulse://user@example.com/client_id/cs1a/', {
'instance': NotifySendPulse,
'requests_response_text': SENDPULSE_BAD_RESPONSE,
# Notify will fail because auth failed
'response': False,
}),
('sendpulse://user@example.com/client_id/cs2/'
'?bcc=l2g@nuxref.com', {
# A good email with Blind Carbon Copy
'instance': NotifySendPulse,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
('sendpulse://user@example.com/client_id/client_secret/'
('sendpulse://user@example.com/client_id/cs3/'
'?cc=l2g@nuxref.com', {
# A good email with Carbon Copy
'instance': NotifySendPulse,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
('sendpulse://user@example.com/client_id/client_secret/'
('sendpulse://user@example.com/client_id/cs4/'
'?to=l2g@nuxref.com', {
# A good email with Carbon Copy
'instance': NotifySendPulse,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
('sendpulse://user@example.com/client_id/client_secret/'
('sendpulse://user@example.com/client_id/cs5/'
'?template=1234', {
# A good email with a template + no substitutions
'instance': NotifySendPulse,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
('sendpulse://user@example.com/client_id/client_secret/'
('sendpulse://user@example.com/client_id/cs6/'
'?template=1234&+sub=value&+sub2=value2', {
# A good email with a template + substitutions
'instance': NotifySendPulse,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
# Our expected url(privacy=True) startswith() response:
'privacy_url': 'sendpulse://a...d:user@example.com/',
'privacy_url': 'sendpulse://user@example.com/c...d/c...6/',
}),
('sendpulse://user@example.com/client_id/client_secret/', {
('sendpulse://user@example.com/client_id/cs7/', {
'instance': NotifySendPulse,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
('sendpulse://user@example.com/client_id/client_secret/', {
('sendpulse://user@example.com/client_id/cs8/', {
'instance': NotifySendPulse,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
('sendpulse://user@example.com/client_id/client_secret/', {
('sendpulse://user@example.com/client_id/cs9/', {
'instance': NotifySendPulse,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
}),
)
@ -139,27 +161,27 @@ def test_plugin_sendpulse_edge_cases(mock_post, mock_get):
with pytest.raises(TypeError):
NotifySendPulse(
client_id=None, client_secret='abcd',
client_id=None, client_secret='abcd123',
from_email='user@example.com')
# invalid from email
with pytest.raises(TypeError):
NotifySendPulse(
client_id='abcd', client_secret='abcd', from_email='!invalid')
client_id='abcd', client_secret='abcd456', from_email='!invalid')
# no email
with pytest.raises(TypeError):
NotifySendPulse(
client_id='abcd', client_secret='abcd', from_email=None)
client_id='abcd', client_secret='abcd789', from_email=None)
# Invalid To email address
NotifySendPulse(
client_id='abcd', client_secret='abcd',
client_id='abcd', client_secret='abcd321',
from_email='user@example.com', targets="!invalid")
# Test invalid bcc/cc entries mixed with good ones
assert isinstance(NotifySendPulse(
client_id='abcd', client_secret='abcd',
client_id='abcd', client_secret='abcd654',
from_email='l2g@example.com',
bcc=('abc@def.com', '!invalid'),
cc=('abc@test.org', '!invalid')), NotifySendPulse)
@ -175,6 +197,7 @@ def test_plugin_sendpulse_attachments(mock_post, mock_get):
request = mock.Mock()
request.status_code = requests.codes.ok
request.content = SENDPULSE_GOOD_RESPONSE
# Prepare Mock
mock_post.return_value = request
@ -182,7 +205,7 @@ def test_plugin_sendpulse_attachments(mock_post, mock_get):
path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif')
attach = AppriseAttachment(path)
obj = Apprise.instantiate('sendpulse://user@example.com/abcd/abcd')
obj = Apprise.instantiate('sendpulse://user@example.com/aaaa/bbbb')
assert isinstance(obj, NotifySendPulse)
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,