mirror of https://github.com/caronc/apprise
improved test cases
Chris Caron
parent
5e1e4857f1
commit
70b6803005
|
|
@ -29,15 +29,15 @@
|
|||
# Simple API Reference:
|
||||
# - https://sendpulse.com/integrations/api/smtp
|
||||
|
||||
import re
|
||||
import requests
|
||||
from json import dumps, loads
|
||||
import base64
|
||||
from json import dumps, loads
|
||||
|
||||
from .base import NotifyBase
|
||||
from .. import exception
|
||||
from ..common import NotifyFormat
|
||||
from ..common import NotifyType
|
||||
from ..utils import parse_list
|
||||
from ..utils import is_email
|
||||
from ..utils import parse_emails
|
||||
from ..conversion import convert_between
|
||||
|
|
@ -87,14 +87,18 @@ class NotifySendPulse(NotifyBase):
|
|||
|
||||
# Define object templates
|
||||
templates = (
|
||||
'{schema}://{from_email}/{client_id}/{client_secret}/',
|
||||
'{schema}://{from_email}/{client_id}/{client_secret}/{targets}',
|
||||
'{schema}://{user}@{host}/{client_secret}/',
|
||||
'{schema}://{user}@{host}/{client_id}/{client_secret}/{targets}',
|
||||
)
|
||||
|
||||
# Define our template arguments
|
||||
template_tokens = dict(NotifyBase.template_tokens, **{
|
||||
'from_email': {
|
||||
'name': _('Source Email'),
|
||||
'user': {
|
||||
'name': _('User Name'),
|
||||
'type': 'string',
|
||||
},
|
||||
'host': {
|
||||
'name': _('Domain'),
|
||||
'type': 'string',
|
||||
'required': True,
|
||||
},
|
||||
|
|
@ -131,7 +135,7 @@ class NotifySendPulse(NotifyBase):
|
|||
'from': {
|
||||
'name': _('From Email'),
|
||||
'type': 'string',
|
||||
'map_to': 'from_email',
|
||||
'map_to': 'from_addr',
|
||||
},
|
||||
'cc': {
|
||||
'name': _('Carbon Copy'),
|
||||
|
|
@ -162,9 +166,9 @@ class NotifySendPulse(NotifyBase):
|
|||
},
|
||||
}
|
||||
|
||||
def __init__(self, from_email, client_id, client_secret, targets=None,
|
||||
cc=None, bcc=None, template=None,
|
||||
template_data=None, **kwargs):
|
||||
def __init__(self, client_id, client_secret, from_addr=None, targets=None,
|
||||
cc=None, bcc=None, template=None, template_data=None,
|
||||
**kwargs):
|
||||
"""
|
||||
Initialize Notify SendPulse Object
|
||||
"""
|
||||
|
|
@ -173,12 +177,52 @@ class NotifySendPulse(NotifyBase):
|
|||
# Api Key is acquired upon a sucessful login
|
||||
self.access_token = None
|
||||
|
||||
result = is_email(from_email)
|
||||
# For tracking our email -> name lookups
|
||||
self.names = {}
|
||||
|
||||
# Temporary from_addr to work with for parsing
|
||||
_from_addr = [self.app_id, '']
|
||||
|
||||
if self.user:
|
||||
if self.host:
|
||||
# Prepare the bases of our email
|
||||
_from_addr = [_from_addr[0], '{}@{}'.format(
|
||||
re.split(r'[\s@]+', self.user)[0],
|
||||
self.host,
|
||||
)]
|
||||
|
||||
else:
|
||||
result = is_email(self.user)
|
||||
if result:
|
||||
# Prepare the bases of our email and include domain
|
||||
self.host = result['domain']
|
||||
_from_addr = [
|
||||
result['name'] if result['name']
|
||||
else _from_addr[0], self.user]
|
||||
|
||||
if isinstance(from_addr, str):
|
||||
result = is_email(from_addr)
|
||||
if result:
|
||||
_from_addr = (
|
||||
result['name'] if result['name'] else _from_addr[0],
|
||||
result['full_email'])
|
||||
else:
|
||||
# Only update the string but use the already detected info
|
||||
_from_addr[0] = from_addr
|
||||
|
||||
result = is_email(_from_addr[1])
|
||||
if not result:
|
||||
msg = 'Invalid ~From~ email specified: {}'.format(from_email)
|
||||
# Parse Source domain based on from_addr
|
||||
msg = 'Invalid ~From~ email specified: {}'.format(
|
||||
'{} <{}>'.format(_from_addr[0], _from_addr[1])
|
||||
if _from_addr[0] else '{}'.format(_from_addr[1]))
|
||||
self.logger.warning(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
# Store our lookup
|
||||
self.from_addr = _from_addr[1]
|
||||
self.names[_from_addr[1]] = _from_addr[0]
|
||||
|
||||
# Client ID
|
||||
self.client_id = validate_regex(
|
||||
client_id, *self.template_tokens['client_id']['regex'])
|
||||
|
|
@ -197,14 +241,6 @@ class NotifySendPulse(NotifyBase):
|
|||
self.logger.warning(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
# Tracks emails to name lookups (if they exist)
|
||||
self.__email_map = {}
|
||||
|
||||
# Store from email address
|
||||
self.from_email = result['full_email']
|
||||
self.__email_map[self.from_email] = result['name'] \
|
||||
if result['name'] else self.app_id
|
||||
|
||||
# Acquire Targets (To Emails)
|
||||
self.targets = list()
|
||||
|
||||
|
|
@ -239,7 +275,7 @@ class NotifySendPulse(NotifyBase):
|
|||
if result:
|
||||
self.targets.append(result['full_email'])
|
||||
if result['name']:
|
||||
self.__email_map[result['full_email']] = result['name']
|
||||
self.names[result['full_email']] = result['name']
|
||||
continue
|
||||
|
||||
self.logger.warning(
|
||||
|
|
@ -249,16 +285,16 @@ class NotifySendPulse(NotifyBase):
|
|||
|
||||
else:
|
||||
# If our target email list is empty we want to add ourselves to it
|
||||
self.targets.append(self.from_email)
|
||||
self.targets.append(self.from_addr)
|
||||
|
||||
# Validate recipients (cc:) and drop bad ones:
|
||||
for recipient in parse_list(cc):
|
||||
for recipient in parse_emails(cc):
|
||||
|
||||
result = is_email(recipient)
|
||||
if result:
|
||||
self.cc.add(result['full_email'])
|
||||
if result['name']:
|
||||
self.__email_map[result['full_email']] = result['name']
|
||||
self.names[result['full_email']] = result['name']
|
||||
continue
|
||||
|
||||
self.logger.warning(
|
||||
|
|
@ -267,13 +303,13 @@ class NotifySendPulse(NotifyBase):
|
|||
)
|
||||
|
||||
# Validate recipients (bcc:) and drop bad ones:
|
||||
for recipient in parse_list(bcc):
|
||||
for recipient in parse_emails(bcc):
|
||||
|
||||
result = is_email(recipient)
|
||||
if result:
|
||||
self.bcc.add(result['full_email'])
|
||||
if result['name']:
|
||||
self.__email_map[result['full_email']] = result['name']
|
||||
self.names[result['full_email']] = result['name']
|
||||
continue
|
||||
|
||||
self.logger.warning(
|
||||
|
|
@ -283,7 +319,7 @@ class NotifySendPulse(NotifyBase):
|
|||
|
||||
if len(self.targets) == 0:
|
||||
# Notify ourselves
|
||||
self.targets.append(self.from_email)
|
||||
self.targets.append(self.from_addr)
|
||||
|
||||
return
|
||||
|
||||
|
|
@ -308,8 +344,8 @@ class NotifySendPulse(NotifyBase):
|
|||
# Handle our Carbon Copy Addresses
|
||||
params['cc'] = ','.join([
|
||||
formataddr(
|
||||
(self.__email_map[e]
|
||||
if e in self.__email_map else False, e),
|
||||
(self.names[e]
|
||||
if e in self.names 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')
|
||||
|
|
@ -319,8 +355,8 @@ class NotifySendPulse(NotifyBase):
|
|||
# Handle our Blind Carbon Copy Addresses
|
||||
params['bcc'] = ','.join([
|
||||
formataddr(
|
||||
(self.__email_map[e]
|
||||
if e in self.__email_map else False, e),
|
||||
(self.names[e]
|
||||
if e in self.names 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')
|
||||
|
|
@ -330,6 +366,10 @@ class NotifySendPulse(NotifyBase):
|
|||
# Handle our Template ID if if was specified
|
||||
params['template'] = self.template
|
||||
|
||||
# handle from=
|
||||
if self.names[self.from_addr] != self.app_id:
|
||||
params['from'] = self.names[self.from_addr]
|
||||
|
||||
# Append our template_data into our parameter list
|
||||
params.update(
|
||||
{'+{}'.format(k): v for k, v in self.template_data.items()})
|
||||
|
|
@ -337,11 +377,11 @@ class NotifySendPulse(NotifyBase):
|
|||
# a simple boolean check as to whether we display our target emails
|
||||
# or not
|
||||
has_targets = \
|
||||
not (len(self.targets) == 1 and self.targets[0] == self.from_email)
|
||||
not (len(self.targets) == 1 and self.targets[0] == self.from_addr)
|
||||
|
||||
return '{schema}://{source}/{cid}/{secret}/{targets}?{params}'.format(
|
||||
schema=self.secure_protocol,
|
||||
source=self.from_email,
|
||||
source=self.from_addr,
|
||||
cid=self.pprint(self.client_id, privacy, safe=''),
|
||||
secret=self.pprint(self.client_secret, privacy, safe=''),
|
||||
targets='' if not has_targets else '/'.join(
|
||||
|
|
@ -388,8 +428,8 @@ class NotifySendPulse(NotifyBase):
|
|||
_payload = {
|
||||
'email': {
|
||||
'from': {
|
||||
'name': self.from_email[0],
|
||||
'email': self.from_email[1],
|
||||
'name': self.names[self.from_addr],
|
||||
'email': self.from_addr,
|
||||
},
|
||||
# To is populated further on
|
||||
'to': [],
|
||||
|
|
@ -473,8 +513,8 @@ class NotifySendPulse(NotifyBase):
|
|||
to = {
|
||||
'email': target
|
||||
}
|
||||
if target in self.__email_map:
|
||||
to['name'] = self.__email_map[target]
|
||||
if target in self.names:
|
||||
to['name'] = self.names[target]
|
||||
|
||||
# Set our target
|
||||
payload['email']['to'] = [to]
|
||||
|
|
@ -485,8 +525,8 @@ class NotifySendPulse(NotifyBase):
|
|||
item = {
|
||||
'email': email,
|
||||
}
|
||||
if email in self.__email_map:
|
||||
item['name'] = self.__email_map[email]
|
||||
if email in self.names:
|
||||
item['name'] = self.names[email]
|
||||
|
||||
payload['email']['cc'].append(item)
|
||||
|
||||
|
|
@ -496,8 +536,8 @@ class NotifySendPulse(NotifyBase):
|
|||
item = {
|
||||
'email': email,
|
||||
}
|
||||
if email in self.__email_map:
|
||||
item['name'] = self.__email_map[email]
|
||||
if email in self.names:
|
||||
item['name'] = self.names[email]
|
||||
|
||||
payload['email']['bcc'].append(item)
|
||||
|
||||
|
|
@ -528,7 +568,7 @@ class NotifySendPulse(NotifyBase):
|
|||
headers.update(extended_headers)
|
||||
|
||||
self.logger.debug('SendPulse POST URL: %s (cert_verify=%r)' % (
|
||||
self.url, self.verify_certificate,
|
||||
url, self.verify_certificate,
|
||||
))
|
||||
self.logger.debug('SendPulse Payload: %s' % str(payload))
|
||||
|
||||
|
|
@ -539,7 +579,7 @@ class NotifySendPulse(NotifyBase):
|
|||
self.throttle()
|
||||
try:
|
||||
r = requests.post(
|
||||
self.url,
|
||||
url,
|
||||
data=dumps(payload),
|
||||
headers=headers,
|
||||
verify=self.verify_certificate,
|
||||
|
|
@ -610,19 +650,22 @@ class NotifySendPulse(NotifyBase):
|
|||
|
||||
"""
|
||||
|
||||
results = NotifyBase.parse_url(url)
|
||||
results = NotifyBase.parse_url(url, verify_host=False)
|
||||
if not results:
|
||||
# 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['from_addr'] = None
|
||||
results['client_id'] = None
|
||||
results['client_secret'] = None
|
||||
|
||||
# Prepare our targets
|
||||
results['targets'] = []
|
||||
|
||||
# Our URL looks like this:
|
||||
# {schema}://{from_email}:{client_id}/{client_secret}/{targets}
|
||||
# {schema}://{from_addr}:{client_id}/{client_secret}/{targets}
|
||||
#
|
||||
# which actually equates to:
|
||||
# {schema}://{user}@{host}/{client_id}/{client_secret}
|
||||
|
|
@ -630,10 +673,27 @@ class NotifySendPulse(NotifyBase):
|
|||
# ^ ^
|
||||
# | |
|
||||
# -from addr-
|
||||
if 'from' in results['qsd']:
|
||||
results['from_addr'] = \
|
||||
NotifySendPulse.unquote(results['qsd']['from'].rstrip())
|
||||
|
||||
if is_email(results['from_addr']):
|
||||
# Our hostname is free'd up to be interpreted as part of the
|
||||
# targets
|
||||
results['targets'].append(
|
||||
NotifySendPulse.unquote(results['host']))
|
||||
results['host'] = ''
|
||||
|
||||
if 'user' in results['qsd'] and \
|
||||
is_email(NotifySendPulse.unquote(results['user'])):
|
||||
# Our hostname is free'd up to be interpreted as part of the
|
||||
# targets
|
||||
results['targets'].append(NotifySendPulse.unquote(results['host']))
|
||||
results['host'] = ''
|
||||
|
||||
# Get our potential email targets
|
||||
# First 2 elements are the client_id and client_secret
|
||||
results['targets'] = NotifySendPulse.split_path(results['fullpath'])
|
||||
results['targets'] += NotifySendPulse.split_path(results['fullpath'])
|
||||
# check for our client id
|
||||
if 'id' in results['qsd'] and len(results['qsd']['id']):
|
||||
# Store our Client ID
|
||||
|
|
@ -653,41 +713,20 @@ class NotifySendPulse(NotifyBase):
|
|||
# Store our Client Secret
|
||||
results['client_secret'] = results['targets'].pop(0)
|
||||
|
||||
if 'from' in results['qsd'] and len(results['qsd']['from']):
|
||||
results['from_email'] = \
|
||||
NotifySendPulse.unquote(results['qsd']['from_email'])
|
||||
|
||||
# This means any user@host is the To Address if defined
|
||||
if results.get('user') and results.get('host'):
|
||||
results['targets'] += '{}@{}'.format(
|
||||
NotifySendPulse.unquote(
|
||||
results['password']
|
||||
if results['password'] else results['user']),
|
||||
NotifySendPulse.unquote(results['host']),
|
||||
)
|
||||
|
||||
elif results.get('user') and results.get('host'):
|
||||
results['from_email'] = '{}@{}'.format(
|
||||
NotifySendPulse.unquote(
|
||||
results['password']
|
||||
if results['password'] else results['user']),
|
||||
NotifySendPulse.unquote(results['host']),
|
||||
)
|
||||
|
||||
# The 'to' makes it easier to use yaml configuration
|
||||
if 'to' in results['qsd'] and len(results['qsd']['to']):
|
||||
results['targets'] += \
|
||||
NotifySendPulse.parse_list(results['qsd']['to'])
|
||||
NotifySendPulse.unquote(results['qsd']['to'])
|
||||
|
||||
# Handle Carbon Copy Addresses
|
||||
if 'cc' in results['qsd'] and len(results['qsd']['cc']):
|
||||
results['cc'] = \
|
||||
NotifySendPulse.parse_list(results['qsd']['cc'])
|
||||
NotifySendPulse.unquote(results['qsd']['cc'])
|
||||
|
||||
# Handle Blind Carbon Copy Addresses
|
||||
if 'bcc' in results['qsd'] and len(results['qsd']['bcc']):
|
||||
results['bcc'] = \
|
||||
NotifySendPulse.parse_list(results['qsd']['bcc'])
|
||||
NotifySendPulse.unquote(results['qsd']['bcc'])
|
||||
|
||||
# Handle Blind Carbon Copy Addresses
|
||||
if 'template' in results['qsd'] and len(results['qsd']['template']):
|
||||
|
|
|
|||
|
|
@ -26,7 +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 json import dumps, loads
|
||||
from unittest import mock
|
||||
|
||||
import os
|
||||
|
|
@ -55,10 +55,10 @@ TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
|
|||
# Our Testing URLs
|
||||
apprise_url_tests = (
|
||||
('sendpulse://', {
|
||||
'instance': None,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('sendpulse://:@/', {
|
||||
'instance': None,
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('sendpulse://abcd', {
|
||||
# invalid from email
|
||||
|
|
@ -68,10 +68,60 @@ apprise_url_tests = (
|
|||
# Just an Email specified, no client_id or client_secret
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs/?template=invalid', {
|
||||
# Invalid template
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs1/?template=123', {
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs1/', {
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs1/?format=text', {
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs1/?format=html', {
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://chris@example.com/client_id/cs1/?from=Chris', {
|
||||
# Set name only
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://?id=ci&secret=cs&user=chris@example.com', {
|
||||
# Set login through user= only
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://?id=ci&secret=cs&user=chris', {
|
||||
# Set login through user= only - invaild email
|
||||
'instance': TypeError,
|
||||
}),
|
||||
('sendpulse://example.com/client_id/cs1/?user=chris', {
|
||||
# Set user as a name only
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://client_id/cs1/?user=chris@example.ca', {
|
||||
# Set user as email
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://client_id/cs1/?from=Chris<chris@example.com>', {
|
||||
# set full email with name
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://?from=Chris<chris@example.com>&id=ci&secret=cs', {
|
||||
# leverage all get params from URL
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs1a/', {
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_BAD_RESPONSE,
|
||||
|
|
@ -82,31 +132,90 @@ apprise_url_tests = (
|
|||
'?bcc=l2g@nuxref.com', {
|
||||
# A good email with Blind Carbon Copy
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs2/'
|
||||
'?bcc=invalid', {
|
||||
# A good email with Blind Carbon Copy
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('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,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs4/'
|
||||
'?to=l2g@nuxref.com', {
|
||||
('sendpulse://user@example.com/client_id/cs3/'
|
||||
'?cc=invalid', {
|
||||
# A good email with Carbon Copy
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs4/'
|
||||
'?to=invalid', {
|
||||
# an invalid to email
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/chris@example.com', {
|
||||
# An email with a designated to email
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?to=Chris<chris@example.com>', {
|
||||
# An email with a full name in in To field
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?template=1234', {
|
||||
# A good email with a template + no substitutions
|
||||
'chris@example.com/chris2@example.com/Test<test@test.com>', {
|
||||
# Several emails to notify
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?cc=Chris<chris@example.com>', {
|
||||
# An email with a full name in cc
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?cc=chris@example.com', {
|
||||
# An email with a full name in cc
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?bcc=Chris<chris@example.com>', {
|
||||
# An email with a full name in bcc
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?bcc=chris@example.com', {
|
||||
# An email with a full name in bcc
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?to=Chris<chris@example.com>', {
|
||||
# An email with a full name in bcc
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('sendpulse://user@example.com/client_id/cs5/'
|
||||
'?to=chris@example.com', {
|
||||
# An email with a full name in bcc
|
||||
'instance': NotifySendPulse,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
}),
|
||||
('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,
|
||||
'requests_response_text': SENDPULSE_GOOD_RESPONSE,
|
||||
|
||||
# Our expected url(privacy=True) startswith() response:
|
||||
'privacy_url': 'sendpulse://user@example.com/c...d/c...6/',
|
||||
|
|
@ -145,11 +254,109 @@ def test_plugin_sendpulse_urls():
|
|||
AppriseURLTester(tests=apprise_url_tests).run_all()
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_plugin_sendpulse_edge_cases(mock_post, mock_get):
|
||||
def test_plugin_sendpulse_edge_cases(mock_post):
|
||||
"""
|
||||
NotifySendPulse() Edge Cases
|
||||
"""
|
||||
request = mock.Mock()
|
||||
request.status_code = requests.codes.ok
|
||||
request.content = SENDPULSE_GOOD_RESPONSE
|
||||
|
||||
# Prepare Mock
|
||||
mock_post.return_value = request
|
||||
|
||||
obj = Apprise.instantiate(
|
||||
'sendpulse://user@example.com/ci/cs/Test<test@example.com>')
|
||||
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is True
|
||||
|
||||
# Test our call count
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
# Authentication
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.sendpulse.com/oauth/access_token'
|
||||
|
||||
payload = loads(mock_post.call_args_list[0][1]['data'])
|
||||
assert payload == {
|
||||
'grant_type': 'client_credentials',
|
||||
'client_id': 'ci',
|
||||
'client_secret': 'cs',
|
||||
}
|
||||
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.sendpulse.com/smtp/emails'
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
assert payload == {
|
||||
'email': {
|
||||
'from': {
|
||||
'email': 'user@example.com', 'name': 'Apprise'
|
||||
},
|
||||
'to': [{'email': 'test@example.com', 'name': 'Test'}],
|
||||
'subject': 'title', 'text': 'body', 'html': 'Ym9keQ=='}}
|
||||
|
||||
mock_post.reset_mock()
|
||||
|
||||
obj = Apprise.instantiate('sendpulse://user@example.com/ci/cs/?from=John')
|
||||
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is True
|
||||
|
||||
# Test our call count
|
||||
assert mock_post.call_count == 2
|
||||
|
||||
# Authentication
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.sendpulse.com/oauth/access_token'
|
||||
|
||||
payload = loads(mock_post.call_args_list[0][1]['data'])
|
||||
assert payload == {
|
||||
'grant_type': 'client_credentials',
|
||||
'client_id': 'ci',
|
||||
'client_secret': 'cs',
|
||||
}
|
||||
|
||||
assert mock_post.call_args_list[1][0][0] == \
|
||||
'https://api.sendpulse.com/smtp/emails'
|
||||
payload = loads(mock_post.call_args_list[1][1]['data'])
|
||||
|
||||
assert payload == {
|
||||
'email': {
|
||||
'from': {
|
||||
'email': 'user@example.com', 'name': 'John'
|
||||
},
|
||||
'to': [{'email': 'user@example.com', 'name': 'John'}],
|
||||
'subject': 'title', 'text': 'body', 'html': 'Ym9keQ=='}}
|
||||
|
||||
mock_post.reset_mock()
|
||||
|
||||
# Second call no longer needs to authenticate
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is True
|
||||
|
||||
assert mock_post.call_count == 1
|
||||
|
||||
assert mock_post.call_args_list[0][0][0] == \
|
||||
'https://api.sendpulse.com/smtp/emails'
|
||||
|
||||
# force an exception
|
||||
mock_post.side_effect = requests.RequestException
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
|
||||
# Set an invalid return code
|
||||
mock_post.side_effect = None
|
||||
request.status_code = 403
|
||||
assert obj.notify(
|
||||
body='body', title='title', notify_type=NotifyType.INFO) is False
|
||||
|
||||
|
||||
def test_plugin_sendpulse_fail_cases():
|
||||
"""
|
||||
NotifySendPulse() Fail Cases
|
||||
|
||||
"""
|
||||
|
||||
|
|
@ -157,39 +364,38 @@ def test_plugin_sendpulse_edge_cases(mock_post, mock_get):
|
|||
with pytest.raises(TypeError):
|
||||
NotifySendPulse(
|
||||
client_id='abcd', client_secret=None,
|
||||
from_email='user@example.com')
|
||||
from_addr='user@example.com')
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
NotifySendPulse(
|
||||
client_id=None, client_secret='abcd123',
|
||||
from_email='user@example.com')
|
||||
from_addr='user@example.com')
|
||||
|
||||
# invalid from email
|
||||
with pytest.raises(TypeError):
|
||||
NotifySendPulse(
|
||||
client_id='abcd', client_secret='abcd456', from_email='!invalid')
|
||||
client_id='abcd', client_secret='abcd456', from_addr='!invalid')
|
||||
|
||||
# no email
|
||||
with pytest.raises(TypeError):
|
||||
NotifySendPulse(
|
||||
client_id='abcd', client_secret='abcd789', from_email=None)
|
||||
client_id='abcd', client_secret='abcd789', from_addr=None)
|
||||
|
||||
# Invalid To email address
|
||||
NotifySendPulse(
|
||||
client_id='abcd', client_secret='abcd321',
|
||||
from_email='user@example.com', targets="!invalid")
|
||||
from_addr='user@example.com', targets="!invalid")
|
||||
|
||||
# Test invalid bcc/cc entries mixed with good ones
|
||||
assert isinstance(NotifySendPulse(
|
||||
client_id='abcd', client_secret='abcd654',
|
||||
from_email='l2g@example.com',
|
||||
from_addr='l2g@example.com',
|
||||
bcc=('abc@def.com', '!invalid'),
|
||||
cc=('abc@test.org', '!invalid')), NotifySendPulse)
|
||||
|
||||
|
||||
@mock.patch('requests.get')
|
||||
@mock.patch('requests.post')
|
||||
def test_plugin_sendpulse_attachments(mock_post, mock_get):
|
||||
def test_plugin_sendpulse_attachments(mock_post):
|
||||
"""
|
||||
NotifySendPulse() Attachments
|
||||
|
||||
|
|
@ -201,7 +407,6 @@ def test_plugin_sendpulse_attachments(mock_post, mock_get):
|
|||
|
||||
# Prepare Mock
|
||||
mock_post.return_value = request
|
||||
mock_get.return_value = request
|
||||
|
||||
path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif')
|
||||
attach = AppriseAttachment(path)
|
||||
|
|
@ -212,7 +417,6 @@ def test_plugin_sendpulse_attachments(mock_post, mock_get):
|
|||
attach=attach) is True
|
||||
|
||||
mock_post.reset_mock()
|
||||
mock_get.reset_mock()
|
||||
|
||||
# Try again in a use case where we can't access the file
|
||||
with mock.patch("os.path.isfile", return_value=False):
|
||||
|
|
|
|||
Loading…
Reference in New Issue