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

full test coverage

pull/1225/head
Chris Caron 2024-12-07 16:17:53 -05:00
parent 6b6bc2c65b
commit 79aae7bdb1
2 changed files with 368 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
apprise/plugins/office365.py
View File

@ -37,6 +37,7 @@
# Note: One must set up Application Permissions (not Delegated Permissions)
# - Scopes required: Mail.Send
# - For Large Attachments: Mail.ReadWrite
# - For Email Lookups: User.Read.All
#
import requests
import json
@ -285,9 +286,16 @@ class NotifyOffice365(NotifyBase):
# Our email source; we detect this if the source is an ObjectID
# If it is unknown we set this to None
self.from_email = self.source \
if (self.source and is_email(self.source)) \
else self.store.get('from')
# User is the email associated with the account
self.from_email = self.store.get('from')
result = is_email(self.source)
if result:
self.from_email = result['full_email']
self.from_name = \
result['name'] or self.store.get('name')
else:
self.from_name = self.store.get('name')
return
@ -306,7 +314,7 @@ class NotifyOffice365(NotifyBase):
'There are no Email recipients to notify')
return False
if not self.from_email:
if self.from_email is None:
if not self.authenticate():
# We could not authenticate ourselves; we're done
return False
@ -319,18 +327,27 @@ class NotifyOffice365(NotifyBase):
self.logger.warning(
'Could not acquire From email address; ensure '
'"User.Read.All" Application scope is set!')
return False
# Acquire our from_email
self.from_email = \
response.get("mail") or response.get("userPrincipalName")
if not is_email(self.from_email):
self.logger.warning(
'Could not get From email from the Azure endpoint.')
return False
else: # Acquire our from_email (if possible)
from_email = \
response.get("mail") or response.get("userPrincipalName")
result = is_email(from_email)
if not result:
self.logger.warning(
'Could not get From email from the Azure endpoint.')
# Store our email for future reference
self.store.set('from', self.from_email)
# Prevent re-occuring upstream fetches for info that isn't
# there
self.from_email = False
else:
# Store our email for future reference
self.from_email = result['full_email']
self.store.set('from', result['full_email'])
self.from_name = response.get("displayName")
if self.from_name:
self.store.set('name', self.from_name)
# Setup our Content Type
content_type = \
@ -339,11 +356,6 @@ class NotifyOffice365(NotifyBase):
# Prepare our payload
payload = {
'message': {
'from': {
"emailAddress": {
"address": self.from_email,
}
},
'subject': title,
'body': {
'contentType': content_type,
@ -354,6 +366,19 @@ class NotifyOffice365(NotifyBase):
'saveToSentItems': 'true'
}
if self.from_email:
# Apply from email if it is known
payload.update({
'message': {
'from': {
"emailAddress": {
"address": self.from_email,
"name": self.from_name or self.app_id,
}
},
}
})
# Create a copy of the email list
emails = list(self.targets)
@ -446,7 +471,7 @@ class NotifyOffice365(NotifyBase):
# Prepare our email
payload['message']['toRecipients'] = [{
'emailAddress': {
'Address': to_addr
'address': to_addr
}
}]
if to_name:
@ -461,9 +486,9 @@ class NotifyOffice365(NotifyBase):
# Prepare our CC list
payload['message']['ccRecipients'] = []
for addr in cc:
_payload = {'Address': addr}
_payload = {'address': addr}
if self.names.get(addr):
_payload['Name'] = self.names[addr]
_payload['name'] = self.names[addr]
# Store our address in our payload
payload['message']['ccRecipients']\
@ -787,6 +812,23 @@ class NotifyOffice365(NotifyBase):
# }}
# }
# Error 403; the below is returned if he User.Read.All
# Application scope is not set and a lookup is
# attempted.
# {
# "error": {
# "code": "Authorization_RequestDenied",
# "message":
# "Insufficient privileges to complete the operation.",
# "innerError": {
# "date": "2024-12-06T00:15:57",
# "request-id":
# "48fdb3e7-2f1a-4f45-a5a0-99b8b851278b",
# "client-request-id": "48f-2f1a-4f45-a5a0-99b8"
# }
# }
# }
# Another response type (error 415):
# {
# "error": {

313
test/test_plugin_office365.py
View File

@ -135,6 +135,7 @@ apprise_url_tests = (
'expires_in': 2000,
'access_token': 'abcd1234',
'mail': 'user@example.ca',
"displayName": "John",
},
# Our expected url(privacy=True) startswith() response:
@ -476,6 +477,105 @@ def test_plugin_office365_authentication(mock_get, mock_post):
assert obj.authenticate() is False
@mock.patch('requests.put')
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_plugin_office365_queries(mock_post, mock_get, mock_put):
"""
NotifyOffice365() General Queries
"""
# Initialize some generic (but valid) tokens
source = 'abc-1234-object-id'
tenant = 'ff-gg-hh-ii-jj'
client_id = 'aa-bb-cc-dd-ee'
secret = 'abcd/1234/abcd@ajd@/test'
targets = 'target@example.ca'
# Prepare Mock return object
payload = {
"token_type": "Bearer",
"expires_in": 6000,
"access_token": "abcd1234",
# For 'From:' Lookup (email)
"mail": "user@example.edu",
# For 'From:' Lookup (name)
"displayName": "John",
# For our Draft Email ID:
"id": "draft-id-no",
# For FIle Uploads
"uploadUrl": "https://my.url.path/"
}
okay_response = mock.Mock()
okay_response.content = dumps(payload)
okay_response.status_code = requests.codes.ok
mock_post.return_value = okay_response
mock_put.return_value = okay_response
bad_response = mock.Mock()
bad_response.content = dumps(payload)
bad_response.status_code = requests.codes.forbidden
# Assign our GET a bad response so we fail to look up the user
mock_get.return_value = bad_response
# Instantiate our object
obj = Apprise.instantiate(
'azure://{source}/{tenant}/{client_id}{secret}/{targets}'.format(
client_id=client_id,
tenant=tenant,
source=source,
secret=secret,
targets=targets))
assert isinstance(obj, NotifyOffice365)
# We can still send a notification even if we can't look up the email
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'https://login.microsoftonline.com/{}/oauth2/v2.0/token'.format(tenant)
assert mock_post.call_args_list[1][0][0] == \
'https://graph.microsoft.com/v1.0/users/abc-1234-object-id/sendMail'
mock_post.reset_mock()
# Now test a case where we just couldn't get any email details from the
# payload returned
# Prepare Mock return object
temp_payload = {
"token_type": "Bearer",
"expires_in": 6000,
"access_token": "abcd1234",
# For our Draft Email ID:
"id": "draft-id-no",
# For FIle Uploads
"uploadUrl": "https://my.url.path/"
}
bad_response.content = dumps(temp_payload)
bad_response.status_code = requests.codes.okay
mock_get.return_value = bad_response
obj = Apprise.instantiate(
'azure://{source}/{tenant}/{client_id}{secret}/{targets}'.format(
client_id=client_id,
tenant=tenant,
source=source,
secret=secret,
targets=targets))
assert isinstance(obj, NotifyOffice365)
# We can still send a notification even if we can't look up the email
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO) is True
@mock.patch('requests.put')
@mock.patch('requests.get')
@mock.patch('requests.post')
@ -486,7 +586,7 @@ def test_plugin_office365_attachments(mock_post, mock_get, mock_put):
"""
# Initialize some generic (but valid) tokens
email = 'user@example.net'
source = 'user@example.net'
tenant = 'ff-gg-hh-ii-jj'
client_id = 'aa-bb-cc-dd-ee'
secret = 'abcd/1234/abcd@ajd@/test'
@ -513,10 +613,10 @@ def test_plugin_office365_attachments(mock_post, mock_get, mock_put):
# Instantiate our object
obj = Apprise.instantiate(
'azure://{email}/{tenant}/{client_id}{secret}/{targets}'.format(
'azure://{source}/{tenant}/{client_id}{secret}/{targets}'.format(
client_id=client_id,
tenant=tenant,
email=email,
source=source,
secret=secret,
targets=targets))
@ -535,12 +635,50 @@ def test_plugin_office365_attachments(mock_post, mock_get, mock_put):
assert mock_post.call_args_list[0][1]['headers'] \
.get('Content-Type') == 'application/x-www-form-urlencoded'
assert mock_post.call_args_list[1][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/sendMail'.format(email)
'https://graph.microsoft.com/v1.0/users/{}/sendMail'.format(source)
assert mock_post.call_args_list[1][1]['headers'] \
.get('Content-Type') == 'application/json'
mock_post.reset_mock()
# Test Authentication Failure
obj = Apprise.instantiate(
'azure://{source}/{tenant}/{client_id}{secret}/{targets}'.format(
client_id=client_id,
tenant=tenant,
source='object-id-requiring-lookup',
secret=secret,
targets=targets))
bad_response = mock.Mock()
bad_response.content = dumps(payload)
bad_response.status_code = requests.codes.forbidden
mock_post.return_value = bad_response
assert isinstance(obj, NotifyOffice365)
# Authentication will fail
assert obj.notify(
body='auth-fail', title='title', notify_type=NotifyType.INFO) is False
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://login.microsoftonline.com/ff-gg-hh-ii-jj/oauth2/v2.0/token'
mock_post.reset_mock()
#
# Test invalid attachment
#
# Instantiate our object
obj = Apprise.instantiate(
'azure://{source}/{tenant}/{client_id}{secret}/{targets}'.format(
client_id=client_id,
tenant=tenant,
source=source,
secret=secret,
targets=targets))
assert isinstance(obj, NotifyOffice365)
mock_post.return_value = okay_response
path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg')
assert obj.notify(
body='body', title='title', notify_type=NotifyType.INFO,
@ -556,22 +694,179 @@ def test_plugin_office365_attachments(mock_post, mock_get, mock_put):
assert mock_post.call_count == 0
mock_post.reset_mock()
#
# Test case where we can't authenticate
#
obj = Apprise.instantiate(
'azure://{source}/{tenant}/{client_id}{secret}/{targets}'.format(
client_id=client_id,
tenant=tenant,
source=source,
secret=secret,
targets=targets))
# Force a smaller attachment size forcing us to create an attachment
obj.outlook_attachment_inline_max = 50
# We can't create an attachment now..
assert isinstance(obj, NotifyOffice365)
path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif')
attach = AppriseAttachment(path)
mock_post.return_value = bad_response
assert obj.upload_attachment(attach[0], 'id') is False
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://login.microsoftonline.com/ff-gg-hh-ii-jj/oauth2/v2.0/token'
mock_post.reset_mock()
mock_post.side_effect = (okay_response, bad_response)
mock_post.return_value = None
assert obj.upload_attachment(attach[0], 'id') is False
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'https://login.microsoftonline.com/ff-gg-hh-ii-jj/oauth2/v2.0/token'
assert mock_post.call_args_list[1][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/'.format(source) + \
'message/id/attachments/createUploadSession'
mock_post.reset_mock()
# Return our status
mock_post.side_effect = None
# Prepare Mock return object
payload_no_upload_url = {
"token_type": "Bearer",
"expires_in": 6000,
"access_token": "abcd1234",
# For 'From:' Lookup
"mail": "user@example.edu",
# For our Draft Email ID:
"id": "draft-id-no",
}
tmp_response = mock.Mock()
tmp_response.content = dumps(payload_no_upload_url)
tmp_response.status_code = requests.codes.ok
mock_post.return_value = tmp_response
assert obj.upload_attachment(attach[0], 'id') is False
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/'.format(source) + \
'message/id/attachments/createUploadSession'
mock_post.reset_mock()
# Return our status
mock_post.side_effect = None
mock_post.return_value = okay_response
obj = Apprise.instantiate(
'azure://{source}/{tenant}/{client_id}{secret}/{targets}'.format(
client_id=client_id,
tenant=tenant,
source=source,
secret=secret,
targets=targets))
# Force a smaller attachment size forcing us to create an attachment
obj.outlook_attachment_inline_max = 50
assert isinstance(obj, NotifyOffice365)
# We now have to prepare sepparate session attachments using draft emails
assert obj.notify(
body='body', title='title-test', notify_type=NotifyType.INFO,
attach=attach) is True
# Large Attachments
assert mock_post.call_count == 4
assert mock_post.call_args_list[0][0][0] == \
'https://login.microsoftonline.com/ff-gg-hh-ii-jj/oauth2/v2.0/token'
assert mock_post.call_args_list[1][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/messages'.format(source)
assert mock_post.call_args_list[2][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/'.format(source) + \
'message/draft-id-no/attachments/createUploadSession'
assert mock_post.call_args_list[3][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/sendMail'.format(source)
mock_post.reset_mock()
#
# Handle another case where can't upload the attachment at all
#
path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg')
bad_attach = AppriseAttachment(path)
assert obj.upload_attachment(bad_attach[0], 'id') is False
mock_post.reset_mock()
#
# Handle test case where we can't send the draft email after everything
# has been prepared
#
mock_post.return_value = None
mock_post.side_effect = (okay_response, okay_response, bad_response)
assert obj.notify(
body='body', title='title-test', notify_type=NotifyType.INFO,
attach=attach) is False
assert mock_post.call_count == 3
assert mock_post.call_args_list[0][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/messages'.format(email)
'https://graph.microsoft.com/v1.0/users/{}/messages'.format(source)
assert mock_post.call_args_list[1][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/'.format(email) + \
'https://graph.microsoft.com/v1.0/users/{}/'.format(source) + \
'message/draft-id-no/attachments/createUploadSession'
assert mock_post.call_args_list[2][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/sendMail'.format(email)
'https://graph.microsoft.com/v1.0/users/{}/sendMail'.format(source)
mock_post.reset_mock()
mock_post.side_effect = None
mock_post.return_value = okay_response
#
# Handle test case where we can not upload chunks
#
mock_put.return_value = bad_response
# We now have to prepare sepparate session attachments using draft emails
assert obj.notify(
body='body', title='title-no-chunk', notify_type=NotifyType.INFO,
attach=attach) is False
assert mock_post.call_count == 2
assert mock_post.call_args_list[0][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/messages'.format(source)
assert mock_post.call_args_list[1][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/'.format(source) + \
'message/draft-id-no/attachments/createUploadSession'
mock_put.return_value = okay_response
mock_post.reset_mock()
# Prepare Mock return object
payload_missing_id = {
"token_type": "Bearer",
"expires_in": 6000,
"access_token": "abcd1234",
# For 'From:' Lookup
"mail": "user@example.edu",
# For FIle Uploads
"uploadUrl": "https://my.url.path/"
}
temp_response = mock.Mock()
temp_response.content = dumps(payload_missing_id)
temp_response.status_code = requests.codes.ok
mock_post.return_value = temp_response
# We could not acquire an attachment id, so we'll fail to send our
# notification
assert obj.notify(
body='body', title='title-test', notify_type=NotifyType.INFO,
attach=attach) is False
# Large Attachments
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://graph.microsoft.com/v1.0/users/user@example.net/messages'
mock_post.reset_mock()
# Reset attachment size
@ -583,5 +878,5 @@ def test_plugin_office365_attachments(mock_post, mock_get, mock_put):
# already authenticated
assert mock_post.call_count == 1
assert mock_post.call_args_list[0][0][0] == \
'https://graph.microsoft.com/v1.0/users/{}/sendMail'.format(email)
'https://graph.microsoft.com/v1.0/users/{}/sendMail'.format(source)
mock_post.reset_mock()