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

Phase No 2 - AWS SNS Support using requests; refs #43

pull/67/head
Chris Caron 2019-02-09 23:36:27 -05:00
parent 90d421cbc8
commit bb7b2dd4ca
5 changed files with 580 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

430
apprise/plugins/NotifySNS.py
View File

@ -25,12 +25,15 @@
import re
# Phase 1; use boto3 for a proof of concept
import boto3
from botocore.exceptions import ClientError
from botocore.exceptions import EndpointConnectionError
import hmac
import requests
from hashlib import sha256
from datetime import datetime
from collections import OrderedDict
from xml.etree import ElementTree
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..utils import compat_is_basestring
# Some Phone Number Detection
@ -58,6 +61,12 @@ LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
IS_REGION = re.compile(
r'^\s*(?P<country>[a-z]{2})-(?P<area>[a-z]+)-(?P<no>[0-9]+)\s*$', re.I)
# Extend HTTP Error Messages
AWS_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
AWS_HTTP_ERROR_MAP.update({
403: 'Unauthorized - Invalid Access/Secret Key Combination.',
})
class NotifySNS(NotifyBase):
"""
@ -77,7 +86,8 @@ class NotifySNS(NotifyBase):
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_sns'
# The maximum length of the body
body_maxlen = 256
# Source: https://docs.aws.amazon.com/sns/latest/api/API_Publish.html
body_maxlen = 140
def __init__(self, access_key_id, secret_access_key, region_name,
recipients=None, **kwargs):
@ -86,22 +96,6 @@ class NotifySNS(NotifyBase):
"""
super(NotifySNS, self).__init__(**kwargs)
# Initialize topic list
self.topics = list()
# Initialize numbers list
self.phone = list()
# Store our AWS API Key
self.access_key_id = access_key_id
# Store our AWS API Secret Access key
self.secret_access_key = secret_access_key
# Acquire our AWS Region Name:
# eg. us-east-1, cn-north-1, us-west-2, ...
self.region_name = region_name
if not access_key_id:
raise TypeError(
'An invalid AWS Access Key ID was specified.'
@ -117,6 +111,35 @@ class NotifySNS(NotifyBase):
'An invalid AWS Region was specified.'
)
# Initialize topic list
self.topics = list()
# Initialize numbers list
self.phone = list()
# Store our AWS API Key
self.aws_access_key_id = access_key_id
# Store our AWS API Secret Access key
self.aws_secret_access_key = secret_access_key
# Acquire our AWS Region Name:
# eg. us-east-1, cn-north-1, us-west-2, ...
self.aws_region_name = region_name
# Set our notify_url based on our region
self.notify_url = 'https://sns.{}.amazonaws.com/'\
.format(self.aws_region_name)
# AWS Service Details
self.aws_service_name = 'sns'
self.aws_canonical_uri = '/'
# AWS Authentication Details
self.aws_auth_version = 'AWS4'
self.aws_auth_algorithm = 'AWS4-HMAC-SHA256'
self.aws_auth_request = 'aws4_request'
if recipients is None:
recipients = []
@ -167,97 +190,336 @@ class NotifySNS(NotifyBase):
wrapper to send_notification since we can alert more then one channel
"""
# Create an SNS client
client = boto3.client(
"sns",
aws_access_key_id=self.access_key_id,
aws_secret_access_key=self.secret_access_key,
region_name=self.region_name,
)
# Initiaize our error tracking
has_error = False
error_count = 0
# Create a copy of our phone #'s and topics to notify against
# Create a copy of our phone #'s to notify against
phone = list(self.phone)
topics = list(self.topics)
while len(phone) > 0:
# Get Phone No
no = phone.pop(0)
try:
if not client.publish(PhoneNumber=no, Message=body):
# Prepare SNS Message Payload
payload = {
'Action': u'Publish',
'Message': body,
'Version': u'2010-03-31',
'PhoneNumber': no,
}
# toggle flag
has_error = True
(result, _) = self._post(payload=payload, to=no)
if not result:
error_count += 1
except ClientError as e:
self.logger.warning("The credentials specified were invalid.")
self.logger.debug('AWS Exception: %s' % str(e))
# We can take an early exit at this point since there
# is no need to potentialy get this error message again
# for the remaining (if any) topic/phone to process
return False
except EndpointConnectionError as e:
self.logger.warning(
"The region specified is invalid.")
self.logger.debug('AWS Exception: %s' % str(e))
# We can take an early exit at this point since there
# is no need to potentialy get this error message again
# for the remaining (if any) topic/phone to process
return False
if len(phone) + len(topics) > 0:
if len(phone) > 0:
# Prevent thrashing requests
self.throttle()
# Send all our defined topic id's
while len(topics):
# Get Topic
topic = topics.pop(0)
# Create the topic if it doesn't exist; nothing breaks if it does
topic = client.create_topic(Name=topic)
# First ensure our topic exists, if it doesn't, it gets created
payload = {
'Action': u'CreateTopic',
'Version': u'2010-03-31',
'Name': topic,
}
(result, response) = self._post(payload=payload, to=topic)
if not result:
error_count += 1
continue
# Get the Amazon Resource Name
topic_arn = topic['TopicArn']
topic_arn = response.get('topic_arn')
if not topic_arn:
# Could not acquire our topic; we're done
error_count += 1
continue
# Publish a message.
try:
if not client.publish(Message=body, TopicArn=topic_arn):
# Build our payload now that we know our topic_arn
payload = {
'Action': u'Publish',
'Version': u'2010-03-31',
'TopicArn': topic_arn,
'Message': body,
}
# toggle flag
has_error = True
except ClientError as e:
self.logger.warning(
"The credentials specified were invalid.")
self.logger.debug('AWS Exception: %s' % str(e))
# We can take an early exit at this point since there
# is no need to potentialy get this error message again
# for the remaining (if any) topics to process
return False
except EndpointConnectionError as e:
self.logger.warning(
"The region specified is invalid.")
self.logger.debug('AWS Exception: %s' % str(e))
# We can take an early exit at this point since there
# is no need to potentialy get this error message again
# for the remaining (if any) topic/phone to process
return False
# Send our payload to AWS
(result, _) = self._post(payload=payload, to=topic)
if not result:
error_count += 1
if len(topics) > 0:
# Prevent thrashing requests
self.throttle()
return not has_error
return error_count == 0
def _post(self, payload, to):
"""
Wrapper to request.post() to manage it's response better and make
the notify() function cleaner and easier to maintain.
This function returns True if the _post was successful and False
if it wasn't.
"""
# Convert our payload from a dict() into a urlencoded string
payload = self.urlencode(payload)
# Prepare our Notification URL
# Prepare our AWS Headers based on our payload
headers = self.aws_prepare_request(payload)
self.logger.debug('AWS POST URL: %s (cert_verify=%r)' % (
self.notify_url, self.verify_certificate,
))
self.logger.debug('AWS Payload: %s' % str(payload))
try:
r = requests.post(
self.notify_url,
data=payload,
headers=headers,
verify=self.verify_certificate,
)
if r.status_code != requests.codes.ok:
# We had a problem
try:
self.logger.warning(
'Failed to send AWS notification to '
'"%s": %s (error=%s).' % (
to,
AWS_HTTP_ERROR_MAP[r.status_code],
r.status_code))
except KeyError:
self.logger.warning(
'Failed to send AWS notification to '
'"%s" (error=%s).' % (to, r.status_code))
self.logger.debug('Response Details: %s' % r.text)
return (False, NotifySNS.aws_response_to_dict(r.text))
else:
self.logger.info(
'Sent AWS notification to "%s".' % (to))
except requests.RequestException as e:
self.logger.warning(
'A Connection error occured sending AWS '
'notification to "%s".' % (to),
)
self.logger.debug('Socket Exception: %s' % str(e))
return (False, NotifySNS.aws_response_to_dict(None))
return (True, NotifySNS.aws_response_to_dict(r.text))
def aws_prepare_request(self, payload, reference=None):
"""
Takes the intended payload and returns the headers for it.
The payload is presumed to have been already urlencoded()
"""
# Define our AWS header
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',
# Populated below
'Content-Length': 0,
'Authorization': None,
'X-Amz-Date': None,
}
# Get a reference time (used for header construction)
reference = datetime.utcnow()
# Provide Content-Length
headers['Content-Length'] = str(len(payload))
# Amazon Date Format
amzdate = reference.strftime('%Y%m%dT%H%M%SZ')
headers['X-Amz-Date'] = amzdate
# Credential Scope
scope = '{date}/{region}/{service}/{request}'.format(
date=reference.strftime('%Y%m%d'),
region=self.aws_region_name,
service=self.aws_service_name,
request=self.aws_auth_request,
)
# Similar to headers; but a subset. keys must be lowercase
signed_headers = OrderedDict([
('content-type', headers['Content-Type']),
('host', '{service}.{region}.amazonaws.com'.format(
service=self.aws_service_name,
region=self.aws_region_name)),
('x-amz-date', headers['X-Amz-Date']),
])
#
# Build Canonical Request Object
#
canonical_request = '\n'.join([
# Method
u'POST',
# URL
self.aws_canonical_uri,
# Query String (none set for POST)
'',
# Header Content (must include \n at end!)
# All entries except characters in amazon date must be
# lowercase
'\n'.join(['%s:%s' % (k, v)
for k, v in signed_headers.items()]) + '\n',
# Header Entries (in same order identified above)
';'.join(signed_headers.keys()),
# Payload
sha256(payload.encode('utf-8')).hexdigest(),
])
# Prepare Unsigned Signature
to_sign = '\n'.join([
self.aws_auth_algorithm,
amzdate,
scope,
sha256(canonical_request.encode('utf-8')).hexdigest(),
])
# Our Authorization header
headers['Authorization'] = ', '.join([
'{algorithm} Credential={key}/{scope}'.format(
algorithm=self.aws_auth_algorithm,
key=self.aws_access_key_id,
scope=scope,
),
'SignedHeaders={signed_headers}'.format(
signed_headers=';'.join(signed_headers.keys()),
),
'Signature={signature}'.format(
signature=self.aws_auth_signature(to_sign, reference)
),
])
return headers
def aws_auth_signature(self, to_sign, reference):
"""
Generates a AWS v4 signature based on provided payload
which should be in the form of a string.
"""
def _sign(key, msg, to_hex=False):
"""
Perform AWS Signing
"""
if to_hex:
return hmac.new(key, msg.encode('utf-8'), sha256).hexdigest()
return hmac.new(key, msg.encode('utf-8'), sha256).digest()
_date = _sign((
self.aws_auth_version +
self.aws_secret_access_key).encode('utf-8'),
reference.strftime('%Y%m%d'))
_region = _sign(_date, self.aws_region_name)
_service = _sign(_region, self.aws_service_name)
_signed = _sign(_service, self.aws_auth_request)
return _sign(_signed, to_sign, to_hex=True)
@staticmethod
def aws_response_to_dict(aws_response):
"""
Takes an AWS Response object as input and returns it as a dictionary
but not befor extracting out what is useful to us first.
eg:
IN:
<CreateTopicResponse
xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<CreateTopicResult>
<TopicArn>arn:aws:sns:us-east-1:000000000000:abcd</TopicArn>
</CreateTopicResult>
<ResponseMetadata>
<RequestId>604bef0f-369c-50c5-a7a4-bbd474c83d6a</RequestId>
</ResponseMetadata>
</CreateTopicResponse>
OUT:
{
type: 'CreateTopicResponse',
request_id: '604bef0f-369c-50c5-a7a4-bbd474c83d6a',
topic_arn: 'arn:aws:sns:us-east-1:000000000000:abcd',
}
"""
# Define ourselves a set of directives we want to keep if found and
# then identify the value we want to map them to in our response
# object
aws_keep_map = {
'RequestId': 'request_id',
'TopicArn': 'topic_arn',
'MessageId': 'message_id',
# Error Message Handling
'Type': 'error_type',
'Code': 'error_code',
'Message': 'error_message',
}
# A default response object that we'll manipulate as we pull more data
# from our AWS Response object
response = {
'type': None,
'request_id': None,
}
try:
# we build our tree, but not before first eliminating any
# reference to namespacing (if present) as it makes parsing
# the tree so much easier.
root = ElementTree.fromstring(
re.sub(' xmlns="[^"]+"', '', aws_response, count=1))
# Store our response tag object name
response['type'] = str(root.tag)
def _xml_iter(root, response):
if len(root) > 0:
for child in root:
# use recursion to parse everything
_xml_iter(child, response)
elif root.tag in aws_keep_map.keys():
response[aws_keep_map[root.tag]] = (root.text).strip()
# Recursivly iterate over our AWS Response to extract the
# fields we're interested in in efforts to populate our response
# object.
_xml_iter(root, response)
except (ElementTree.ParseError, TypeError):
# bad data just causes us to generate a bad response
pass
return response
@staticmethod
def parse_url(url):

1
apprise/plugins/__init__.py
View File

@ -47,6 +47,7 @@ from .NotifyPushjet.NotifyPushjet import NotifyPushjet
from .NotifyPushover import NotifyPushover
from .NotifyRocketChat import NotifyRocketChat
from .NotifySlack import NotifySlack
from .NotifySNS import NotifySNS
from .NotifyTelegram import NotifyTelegram
from .NotifyTwitter.NotifyTwitter import NotifyTwitter
from .NotifyXBMC import NotifyXBMC

1
requirements.txt
View File

@ -6,4 +6,3 @@ urllib3
six
click >= 5.0
markdown
boto3

44
test/test_rest_plugins.py
View File

@ -1122,6 +1122,48 @@ TEST_URLS = (
'test_requests_exceptions': True,
}),
##################################
# NotifySNS (AWS)
##################################
('sns://', {
'instance': None,
}),
('sns://:@/', {
'instance': None,
}),
('sns://T1JJ3T3L2', {
# Just Token 1 provided
'instance': TypeError,
}),
('sns://T1JJ3TD4JD/TIiajkdnlazk7FQ/', {
# Missing a region
'instance': TypeError,
}),
('sns://T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcevi7FQ/us-west-2/12223334444', {
# we have a valid URL here
'instance': plugins.NotifySNS,
}),
('sns://T1JJ3TD4JD/TIiajkdnlazk7FQ/us-west-2/12223334444/12223334445', {
# Multi SNS Suppport
'instance': plugins.NotifySNS,
}),
('sns://T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcOXrIdevi7FQ/us-east-1', {
# Missing a topic and/or phone No
'instance': plugins.NotifySNS,
}),
('sns://T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcevi7FQ/us-west-2/12223334444', {
'instance': plugins.NotifySNS,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('sns://T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkcevi7FQ/us-west-2/15556667777', {
'instance': plugins.NotifySNS,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
##################################
# NotifyTelegram
##################################
@ -1511,7 +1553,7 @@ def test_rest_plugins(mock_post, mock_get):
assert isinstance(e, response)
#
# Stage 1: without title defined
# Stage 2: without title defined
#
try:
if test_requests_exceptions is False:

278
test/test_sns_plugin.py
View File

@ -24,9 +24,7 @@
# THE SOFTWARE.
import mock
from botocore.exceptions import ClientError
from botocore.exceptions import EndpointConnectionError
import requests
from apprise import plugins
from apprise import Apprise
@ -35,91 +33,6 @@ TEST_ACCESS_KEY_SECRET = 'bu1dHSdO22pfaaVy/wmNsdljF4C07D3bndi9PQJ9'
TEST_REGION = 'us-east-2'
@mock.patch('boto3.client')
def test_object_notifications(mock_client):
"""
API: NotifySNS Plugin() notifications
"""
# Create our object
a = Apprise()
assert(a.add('sns://oh/yeah/us-west-2/12223334444') is True)
# Multi Number Support
assert(a.add('sns://oh/yeah/us-west-2/12223334444/12223334445') is True)
# Set a successful notification
client = mock.Mock()
client.publish.return_value = True
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is True)
# Set an unsuccessful notification
client = mock.Mock()
client.publish.return_value = False
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is False)
client = mock.Mock()
client.publish.return_value = True
client.publish.side_effect = \
ClientError({'ResponseMetadata': {'RetryAttempts': 1}}, '')
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is False)
client = mock.Mock()
client.publish.return_value = True
client.publish.side_effect = EndpointConnectionError(endpoint_url='')
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is False)
# Create a new object
a = Apprise()
assert(a.add('sns://oh/yeah/us-east-2/ATopic') is True)
# Multi-Topic
assert(a.add('sns://oh/yeah/us-east-2/ATopic/AnotherTopic') is True)
# Set a successful notification
client = mock.Mock()
client.publish.return_value = True
client.create_topic.return_value = {'TopicArn': 'goodtopic'}
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is True)
# Set an unsuccessful notification
client = mock.Mock()
client.publish.return_value = False
client.create_topic.return_value = {'TopicArn': 'goodtopic'}
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is False)
client = mock.Mock()
client.publish.return_value = True
client.publish.side_effect = \
ClientError({'ResponseMetadata': {'RetryAttempts': 1}}, '')
client.create_topic.return_value = {'TopicArn': 'goodtopic'}
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is False)
client = mock.Mock()
client.publish.return_value = True
client.publish.side_effect = EndpointConnectionError(endpoint_url='')
client.create_topic.return_value = {'TopicArn': 'goodtopic'}
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is False)
# Create a new object
a = Apprise()
# Combiniation handling
assert(a.add('sns://oh/yeah/us-west-2/12223334444/ATopicToo') is True)
client = mock.Mock()
client.publish.return_value = True
client.create_topic.return_value = {'TopicArn': 'goodtopic'}
mock_client.return_value = client
assert(a.notify(title='', body='apprise notification') is True)
def test_object_initialization():
"""
API: NotifySNS Plugin() initialization
@ -305,3 +218,192 @@ def test_object_parsing():
assert(a.add('sns://oh/yeah/us-west-2/12223334444') is True)
assert(len(a) == 3)
def test_aws_response_handling():
"""
API: NotifySNS Plugin() AWS Response Handling
"""
# Not a string
response = plugins.NotifySNS.aws_response_to_dict(None)
assert(response['type'] is None)
assert(response['request_id'] is None)
# Invalid XML
response = plugins.NotifySNS.aws_response_to_dict(
'<Bad Response xmlns="http://sns.amazonaws.com/doc/2010-03-31/">')
assert(response['type'] is None)
assert(response['request_id'] is None)
# Single Element in XML
response = plugins.NotifySNS.aws_response_to_dict(
'<SingleElement></SingleElement>')
assert(response['type'] == 'SingleElement')
assert(response['request_id'] is None)
# Empty String
response = plugins.NotifySNS.aws_response_to_dict('')
assert(response['type'] is None)
assert(response['request_id'] is None)
response = plugins.NotifySNS.aws_response_to_dict(
"""
<PublishResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<PublishResult>
<MessageId>5e16935a-d1fb-5a31-a716-c7805e5c1d2e</MessageId>
</PublishResult>
<ResponseMetadata>
<RequestId>dc258024-d0e6-56bb-af1b-d4fe5f4181a4</RequestId>
</ResponseMetadata>
</PublishResponse>
""")
assert(response['type'] == 'PublishResponse')
assert(response['request_id'] == 'dc258024-d0e6-56bb-af1b-d4fe5f4181a4')
assert(response['message_id'] == '5e16935a-d1fb-5a31-a716-c7805e5c1d2e')
response = plugins.NotifySNS.aws_response_to_dict(
"""
<CreateTopicResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<CreateTopicResult>
<TopicArn>arn:aws:sns:us-east-1:000000000000:abcd</TopicArn>
</CreateTopicResult>
<ResponseMetadata>
<RequestId>604bef0f-369c-50c5-a7a4-bbd474c83d6a</RequestId>
</ResponseMetadata>
</CreateTopicResponse>
""")
assert(response['type'] == 'CreateTopicResponse')
assert(response['request_id'] == '604bef0f-369c-50c5-a7a4-bbd474c83d6a')
assert(response['topic_arn'] == 'arn:aws:sns:us-east-1:000000000000:abcd')
response = plugins.NotifySNS.aws_response_to_dict(
"""
<ErrorResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<Error>
<Type>Sender</Type>
<Code>InvalidParameter</Code>
<Message>Invalid parameter: TopicArn or TargetArn Reason:
no value for required parameter</Message>
</Error>
<RequestId>b5614883-babe-56ca-93b2-1c592ba6191e</RequestId>
</ErrorResponse>
""")
assert(response['type'] == 'ErrorResponse')
assert(response['request_id'] == 'b5614883-babe-56ca-93b2-1c592ba6191e')
assert(response['error_type'] == 'Sender')
assert(response['error_code'] == 'InvalidParameter')
assert(response['error_message'].startswith('Invalid parameter:'))
assert(response['error_message'].endswith('required parameter'))
@mock.patch('requests.post')
def test_aws_topic_handling(mock_post):
"""
API: NotifySNS Plugin() AWS Topic Handling
"""
arn_response = \
"""
<CreateTopicResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<CreateTopicResult>
<TopicArn>arn:aws:sns:us-east-1:000000000000:abcd</TopicArn>
</CreateTopicResult>
<ResponseMetadata>
<RequestId>604bef0f-369c-50c5-a7a4-bbd474c83d6a</RequestId>
</ResponseMetadata>
</CreateTopicResponse>
"""
def post(url, data, **kwargs):
"""
Since Publishing a token requires 2 posts, we need to return our
response depending on what step we're on
"""
# A request
robj = mock.Mock()
robj.text = ''
robj.status_code = requests.codes.ok
if data.find('=CreateTopic') >= 0:
# Topic Post Failure
robj.status_code = requests.codes.bad_request
return robj
# Assign ourselves a new function
mock_post.side_effect = post
# Disable Throttling to speed testing
plugins.NotifyBase.NotifyBase.throttle_attempt = 0
# Create our object
a = Apprise()
a.add([
# Single Topic
'sns://T1JJ3T3L2/A1BRTD4JD/TIiajkdnl/us-west-2/TopicA',
# Multi-Topic
'sns://T1JJ3T3L2/A1BRTD4JD/TIiajkdnl/us-east-1/TopicA/TopicB/'
# Topic-Mix
'sns://T1JJ3T3L2/A1BRTD4JD/TIiajkdnlazkce/us-west-2/' \
'12223334444/TopicA'])
# CreateTopic fails
assert(a.notify(title='', body='test') is False)
def post(url, data, **kwargs):
"""
Since Publishing a token requires 2 posts, we need to return our
response depending on what step we're on
"""
# A request
robj = mock.Mock()
robj.text = ''
robj.status_code = requests.codes.ok
if data.find('=CreateTopic') >= 0:
robj.text = arn_response
# Manipulate Topic Publishing only (not phone)
elif data.find('=Publish') >= 0 and data.find('TopicArn=') >= 0:
# Topic Post Failure
robj.status_code = requests.codes.bad_request
return robj
# Assign ourselves a new function
mock_post.side_effect = post
# Publish fails
assert(a.notify(title='', body='test') is False)
# Disable our side effect
mock_post.side_effect = None
# Handle case where TopicArn is missing:
robj = mock.Mock()
robj.text = "<CreateTopicResponse></CreateTopicResponse>"
robj.status_code = requests.codes.ok
# Assign ourselves a new function
mock_post.return_value = robj
assert(a.notify(title='', body='test') is False)
# Handle case where we fails get a bad response
robj = mock.Mock()
robj.text = ''
robj.status_code = requests.codes.bad_request
mock_post.return_value = robj
assert(a.notify(title='', body='test') is False)
# Handle case where we get a valid response and TopicARN
robj = mock.Mock()
robj.text = arn_response
robj.status_code = requests.codes.ok
mock_post.return_value = robj
# We would have failed to make Post
assert(a.notify(title='', body='test') is True)