mirror of https://github.com/caronc/apprise
attachment code hardening + 100% test coverage
parent
717910e969
commit
002cebfaa3
|
@ -314,10 +314,15 @@ class Apprise(object):
|
||||||
# Tracks conversions
|
# Tracks conversions
|
||||||
conversion_map = dict()
|
conversion_map = dict()
|
||||||
|
|
||||||
# Prepare attachments
|
# Prepare attachments if required
|
||||||
if attach is not None and not isinstance(attach, AppriseAttachment):
|
if attach is not None and not isinstance(attach, AppriseAttachment):
|
||||||
|
try:
|
||||||
attach = AppriseAttachment(attach, asset=self.asset)
|
attach = AppriseAttachment(attach, asset=self.asset)
|
||||||
|
|
||||||
|
except TypeError:
|
||||||
|
# bad attachments
|
||||||
|
return False
|
||||||
|
|
||||||
# Iterate over our loaded plugins
|
# Iterate over our loaded plugins
|
||||||
for server in self.find(tag):
|
for server in self.find(tag):
|
||||||
if status is None:
|
if status is None:
|
||||||
|
|
|
@ -26,7 +26,6 @@
|
||||||
import six
|
import six
|
||||||
|
|
||||||
from . import attachment
|
from . import attachment
|
||||||
from . import AttachBase
|
|
||||||
from . import URLBase
|
from . import URLBase
|
||||||
from .AppriseAsset import AppriseAsset
|
from .AppriseAsset import AppriseAsset
|
||||||
from .logger import logger
|
from .logger import logger
|
||||||
|
@ -58,7 +57,9 @@ class AppriseAttachment(object):
|
||||||
# Now parse any paths specified
|
# Now parse any paths specified
|
||||||
if paths is not None:
|
if paths is not None:
|
||||||
# Store our path(s)
|
# Store our path(s)
|
||||||
self.add(paths)
|
if not self.add(paths):
|
||||||
|
# Parse Source domain based on from_addr
|
||||||
|
raise TypeError("One or more attachments could not be added.")
|
||||||
|
|
||||||
def add(self, attachments, asset=None, db=None):
|
def add(self, attachments, asset=None, db=None):
|
||||||
"""
|
"""
|
||||||
|
@ -72,7 +73,7 @@ class AppriseAttachment(object):
|
||||||
# prepare default asset
|
# prepare default asset
|
||||||
asset = self.asset
|
asset = self.asset
|
||||||
|
|
||||||
if isinstance(attachments, AttachBase):
|
if isinstance(attachments, attachment.AttachBase):
|
||||||
# Go ahead and just add our attachments into our list
|
# Go ahead and just add our attachments into our list
|
||||||
self.attachments.append(attachments)
|
self.attachments.append(attachments)
|
||||||
return True
|
return True
|
||||||
|
@ -90,7 +91,7 @@ class AppriseAttachment(object):
|
||||||
# Iterate over our attachments
|
# Iterate over our attachments
|
||||||
for _attachment in attachments:
|
for _attachment in attachments:
|
||||||
|
|
||||||
if isinstance(_attachment, AttachBase):
|
if isinstance(_attachment, attachment.AttachBase):
|
||||||
# Go ahead and just add our attachment into our list
|
# Go ahead and just add our attachment into our list
|
||||||
self.attachments.append(_attachment)
|
self.attachments.append(_attachment)
|
||||||
continue
|
continue
|
||||||
|
@ -107,7 +108,7 @@ class AppriseAttachment(object):
|
||||||
# Instantiate ourselves an object, this function throws or
|
# Instantiate ourselves an object, this function throws or
|
||||||
# returns None if it fails
|
# returns None if it fails
|
||||||
instance = AppriseAttachment.instantiate(_attachment, asset=asset)
|
instance = AppriseAttachment.instantiate(_attachment, asset=asset)
|
||||||
if not isinstance(instance, AttachBase):
|
if not isinstance(instance, attachment.AttachBase):
|
||||||
return_status = False
|
return_status = False
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
|
|
@ -131,10 +131,15 @@ class AttachHTTP(AttachBase):
|
||||||
# Handle Errors
|
# Handle Errors
|
||||||
r.raise_for_status()
|
r.raise_for_status()
|
||||||
|
|
||||||
# Store our response if within reason
|
# Get our file-size (if known)
|
||||||
if self.max_file_size > 0 \
|
try:
|
||||||
and r.headers.get(
|
file_size = int(r.headers.get('Content-Length', '0'))
|
||||||
'Content-Length', 0) > self.max_file_size:
|
except (TypeError, ValueError):
|
||||||
|
# Handle edge case where Content-Length is a bad value
|
||||||
|
file_size = 0
|
||||||
|
|
||||||
|
# Perform a little Q/A on file limitations and restrictions
|
||||||
|
if self.max_file_size > 0 and file_size > self.max_file_size:
|
||||||
|
|
||||||
# The content retrieved is to large
|
# The content retrieved is to large
|
||||||
self.logger.error(
|
self.logger.error(
|
||||||
|
|
|
@ -33,6 +33,7 @@ from ..common import NOTIFY_FORMATS
|
||||||
from ..common import OverflowMode
|
from ..common import OverflowMode
|
||||||
from ..common import OVERFLOW_MODES
|
from ..common import OVERFLOW_MODES
|
||||||
from ..AppriseLocale import gettext_lazy as _
|
from ..AppriseLocale import gettext_lazy as _
|
||||||
|
from ..AppriseAttachment import AppriseAttachment
|
||||||
|
|
||||||
|
|
||||||
class NotifyBase(URLBase):
|
class NotifyBase(URLBase):
|
||||||
|
@ -247,6 +248,15 @@ class NotifyBase(URLBase):
|
||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
# Prepare attachments if required
|
||||||
|
if attach is not None and not isinstance(attach, AppriseAttachment):
|
||||||
|
try:
|
||||||
|
attach = AppriseAttachment(attach, asset=self.asset)
|
||||||
|
|
||||||
|
except TypeError:
|
||||||
|
# bad attachments
|
||||||
|
return False
|
||||||
|
|
||||||
# Handle situations where the title is None
|
# Handle situations where the title is None
|
||||||
title = '' if not title else title
|
title = '' if not title else title
|
||||||
|
|
||||||
|
|
|
@ -27,8 +27,9 @@ import re
|
||||||
import six
|
import six
|
||||||
import smtplib
|
import smtplib
|
||||||
from email.mime.text import MIMEText
|
from email.mime.text import MIMEText
|
||||||
from email.MIMEMultipart import MIMEMultipart
|
|
||||||
from email.mime.application import MIMEApplication
|
from email.mime.application import MIMEApplication
|
||||||
|
from email.mime.multipart import MIMEMultipart
|
||||||
|
|
||||||
from socket import error as SocketError
|
from socket import error as SocketError
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
|
@ -33,9 +33,11 @@ import mock
|
||||||
from os import chmod
|
from os import chmod
|
||||||
from os import getuid
|
from os import getuid
|
||||||
from os.path import dirname
|
from os.path import dirname
|
||||||
|
from os.path import join
|
||||||
|
|
||||||
from apprise import Apprise
|
from apprise import Apprise
|
||||||
from apprise import AppriseAsset
|
from apprise import AppriseAsset
|
||||||
|
from apprise import AppriseAttachment
|
||||||
from apprise import NotifyBase
|
from apprise import NotifyBase
|
||||||
from apprise import NotifyType
|
from apprise import NotifyType
|
||||||
from apprise import NotifyFormat
|
from apprise import NotifyFormat
|
||||||
|
@ -54,6 +56,9 @@ import inspect
|
||||||
import logging
|
import logging
|
||||||
logging.disable(logging.CRITICAL)
|
logging.disable(logging.CRITICAL)
|
||||||
|
|
||||||
|
# Attachment Directory
|
||||||
|
TEST_VAR_DIR = join(dirname(__file__), 'var')
|
||||||
|
|
||||||
|
|
||||||
def test_apprise():
|
def test_apprise():
|
||||||
"""
|
"""
|
||||||
|
@ -166,7 +171,7 @@ def test_apprise():
|
||||||
# Support URL
|
# Support URL
|
||||||
return ''
|
return ''
|
||||||
|
|
||||||
def notify(self, **kwargs):
|
def send(self, **kwargs):
|
||||||
# Pretend everything is okay
|
# Pretend everything is okay
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
@ -200,6 +205,39 @@ def test_apprise():
|
||||||
assert a.notify(title='present', body=None) is True
|
assert a.notify(title='present', body=None) is True
|
||||||
assert a.notify(title="present", body="present") is True
|
assert a.notify(title="present", body="present") is True
|
||||||
|
|
||||||
|
# Send Attachment with success
|
||||||
|
attach = join(TEST_VAR_DIR, 'apprise-test.gif')
|
||||||
|
assert a.notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach=attach) is True
|
||||||
|
|
||||||
|
# Send the attachment as an AppriseAttachment object
|
||||||
|
assert a.notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach=AppriseAttachment(attach)) is True
|
||||||
|
|
||||||
|
# test a invalid attachment
|
||||||
|
assert a.notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach='invalid://') is False
|
||||||
|
|
||||||
|
# Repeat the same tests above...
|
||||||
|
# however do it by directly accessing the object; this grants the similar
|
||||||
|
# results:
|
||||||
|
assert a[0].notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach=attach) is True
|
||||||
|
|
||||||
|
# Send the attachment as an AppriseAttachment object
|
||||||
|
assert a[0].notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach=AppriseAttachment(attach)) is True
|
||||||
|
|
||||||
|
# test a invalid attachment
|
||||||
|
assert a[0].notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach='invalid://') is False
|
||||||
|
|
||||||
# Clear our server listings again
|
# Clear our server listings again
|
||||||
a.clear()
|
a.clear()
|
||||||
|
|
||||||
|
|
|
@ -140,6 +140,11 @@ def test_apprise_attachment():
|
||||||
assert len(aa) == 0
|
assert len(aa) == 0
|
||||||
assert not aa
|
assert not aa
|
||||||
|
|
||||||
|
# if instantiating attachments from the class, it will throw a TypeError
|
||||||
|
# if attachments couldn't be loaded
|
||||||
|
with pytest.raises(TypeError):
|
||||||
|
AppriseAttachment('garbage://')
|
||||||
|
|
||||||
# Garbage in produces garbage out
|
# Garbage in produces garbage out
|
||||||
assert aa.add(None) is False
|
assert aa.add(None) is False
|
||||||
assert aa.add(object()) is False
|
assert aa.add(object()) is False
|
||||||
|
|
|
@ -238,6 +238,7 @@ def test_attach_http(mock_get):
|
||||||
|
|
||||||
# Set our limit to be the length of our image; everything should work
|
# Set our limit to be the length of our image; everything should work
|
||||||
# without a problem
|
# without a problem
|
||||||
|
max_file_size = AttachHTTP.max_file_size
|
||||||
AttachHTTP.max_file_size = getsize(path)
|
AttachHTTP.max_file_size = getsize(path)
|
||||||
# Set ourselves a Content-Disposition (providing a filename)
|
# Set ourselves a Content-Disposition (providing a filename)
|
||||||
dummy_response.headers['Content-Disposition'] = \
|
dummy_response.headers['Content-Disposition'] = \
|
||||||
|
@ -298,6 +299,25 @@ def test_attach_http(mock_get):
|
||||||
assert attachment
|
assert attachment
|
||||||
assert len(attachment) == getsize(path)
|
assert len(attachment) == getsize(path)
|
||||||
|
|
||||||
|
# Set our header up with an invalid Content-Length; we can still process
|
||||||
|
# this data. It just means we track it lower when reading back content
|
||||||
|
dummy_response.headers = {
|
||||||
|
'Content-Length': 'invalid'
|
||||||
|
}
|
||||||
|
results = AttachHTTP.parse_url('http://localhost/invalid-length.gif')
|
||||||
|
assert isinstance(results, dict)
|
||||||
|
attachment = AttachHTTP(**results)
|
||||||
|
assert isinstance(attachment.url(), six.string_types) is True
|
||||||
|
# No mime-type and/or filename over-ride was specified, so therefore it
|
||||||
|
# won't show up in the generated URL
|
||||||
|
assert re.search(r'[?&]mime=', attachment.url()) is None
|
||||||
|
assert re.search(r'[?&]name=', attachment.url()) is None
|
||||||
|
assert attachment.mimetype == 'image/gif'
|
||||||
|
# Because we could determine our mime type, we could build an extension
|
||||||
|
# for our unknown filename
|
||||||
|
assert attachment.name == 'invalid-length.gif'
|
||||||
|
assert attachment
|
||||||
|
|
||||||
# Give ourselves nothing to work with
|
# Give ourselves nothing to work with
|
||||||
dummy_response.headers = {}
|
dummy_response.headers = {}
|
||||||
results = AttachHTTP.parse_url('http://user@localhost')
|
results = AttachHTTP.parse_url('http://user@localhost')
|
||||||
|
@ -328,3 +348,6 @@ def test_attach_http(mock_get):
|
||||||
|
|
||||||
mock_get.side_effect = _exception
|
mock_get.side_effect = _exception
|
||||||
assert not aa
|
assert not aa
|
||||||
|
|
||||||
|
# Restore value
|
||||||
|
AttachHTTP.max_file_size = max_file_size
|
||||||
|
|
|
@ -23,6 +23,7 @@
|
||||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||||
# THE SOFTWARE.
|
# THE SOFTWARE.
|
||||||
|
|
||||||
|
import os
|
||||||
import re
|
import re
|
||||||
import six
|
import six
|
||||||
import mock
|
import mock
|
||||||
|
@ -31,12 +32,17 @@ import smtplib
|
||||||
from apprise import plugins
|
from apprise import plugins
|
||||||
from apprise import NotifyType
|
from apprise import NotifyType
|
||||||
from apprise import Apprise
|
from apprise import Apprise
|
||||||
|
from apprise import AttachBase
|
||||||
|
from apprise import AppriseAttachment
|
||||||
from apprise.plugins import NotifyEmailBase
|
from apprise.plugins import NotifyEmailBase
|
||||||
|
|
||||||
# Disable logging for a cleaner testing output
|
# Disable logging for a cleaner testing output
|
||||||
import logging
|
import logging
|
||||||
logging.disable(logging.CRITICAL)
|
logging.disable(logging.CRITICAL)
|
||||||
|
|
||||||
|
# Attachment Directory
|
||||||
|
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
|
||||||
|
|
||||||
|
|
||||||
TEST_URLS = (
|
TEST_URLS = (
|
||||||
##################################
|
##################################
|
||||||
|
@ -462,6 +468,42 @@ def test_smtplib_send_okay(mock_smtplib):
|
||||||
assert obj.notify(
|
assert obj.notify(
|
||||||
body='body', title='test', notify_type=NotifyType.INFO) is True
|
body='body', title='test', notify_type=NotifyType.INFO) is True
|
||||||
|
|
||||||
|
# Create an apprise object to work with as well
|
||||||
|
a = Apprise()
|
||||||
|
assert a.add('mailto://user:pass@gmail.com?format=text')
|
||||||
|
|
||||||
|
# Send Attachment with success
|
||||||
|
attach = os.path.join(TEST_VAR_DIR, 'apprise-test.gif')
|
||||||
|
assert obj.notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach=attach) is True
|
||||||
|
|
||||||
|
# same results happen from our Apprise object
|
||||||
|
assert a.notify(body='body', title='test', attach=attach) is True
|
||||||
|
|
||||||
|
# test using an Apprise Attachment object
|
||||||
|
assert obj.notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach=AppriseAttachment(attach)) is True
|
||||||
|
|
||||||
|
# same results happen from our Apprise object
|
||||||
|
assert a.notify(
|
||||||
|
body='body', title='test', attach=AppriseAttachment(attach)) is True
|
||||||
|
|
||||||
|
max_file_size = AttachBase.max_file_size
|
||||||
|
# Now do a case where the file can't be sent
|
||||||
|
|
||||||
|
AttachBase.max_file_size = 1
|
||||||
|
assert obj.notify(
|
||||||
|
body='body', title='test', notify_type=NotifyType.INFO,
|
||||||
|
attach=attach) is False
|
||||||
|
|
||||||
|
# same results happen from our Apprise object
|
||||||
|
assert a.notify(body='body', title='test', attach=attach) is False
|
||||||
|
|
||||||
|
# Restore value
|
||||||
|
AttachBase.max_file_size = max_file_size
|
||||||
|
|
||||||
|
|
||||||
def test_email_url_escaping():
|
def test_email_url_escaping():
|
||||||
"""
|
"""
|
||||||
|
|
Loading…
Reference in New Issue