mirror of https://github.com/caronc/apprise
Pushover Attachment Support (#190)
Chris Caron
GitHub
parent
8d2a53cedf
commit
370564943e
|
|
@ -32,6 +32,7 @@ from ..common import NotifyType
|
|||
from ..utils import parse_list
|
||||
from ..utils import validate_regex
|
||||
from ..AppriseLocale import gettext_lazy as _
|
||||
from ..attachment.AttachBase import AttachBase
|
||||
|
||||
# Flag used as a placeholder to sending to all devices
|
||||
PUSHOVER_SEND_TO_ALL = 'ALL_DEVICES'
|
||||
|
|
@ -140,6 +141,14 @@ class NotifyPushover(NotifyBase):
|
|||
# Default Pushover sound
|
||||
default_pushover_sound = PushoverSound.PUSHOVER
|
||||
|
||||
# 2.5MB is the maximum supported image filesize as per documentation
|
||||
# here: https://pushover.net/api#attachments (Dec 26th, 2019)
|
||||
attach_max_size_bytes = 2621440
|
||||
|
||||
# The regular expression of the current attachment supported mime types
|
||||
# At this time it is only images
|
||||
attach_supported_mime_type = r'^image/.*'
|
||||
|
||||
# Define object templates
|
||||
templates = (
|
||||
'{schema}://{user_key}@{token}',
|
||||
|
|
@ -281,17 +290,12 @@ class NotifyPushover(NotifyBase):
|
|||
raise TypeError(msg)
|
||||
return
|
||||
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
|
||||
**kwargs):
|
||||
"""
|
||||
Perform Pushover Notification
|
||||
"""
|
||||
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
'Content-Type': 'application/x-www-form-urlencoded'
|
||||
}
|
||||
auth = (self.token, '')
|
||||
|
||||
# error tracking (used for function return)
|
||||
has_error = False
|
||||
|
||||
|
|
@ -314,7 +318,7 @@ class NotifyPushover(NotifyBase):
|
|||
'token': self.token,
|
||||
'user': self.user_key,
|
||||
'priority': str(self.priority),
|
||||
'title': title,
|
||||
'title': title if title else self.app_desc,
|
||||
'message': body,
|
||||
'device': device,
|
||||
'sound': self.sound,
|
||||
|
|
@ -323,60 +327,158 @@ class NotifyPushover(NotifyBase):
|
|||
if self.priority == PushoverPriority.EMERGENCY:
|
||||
payload.update({'retry': self.retry, 'expire': self.expire})
|
||||
|
||||
self.logger.debug('Pushover POST URL: %s (cert_verify=%r)' % (
|
||||
self.notify_url, self.verify_certificate,
|
||||
))
|
||||
self.logger.debug('Pushover Payload: %s' % str(payload))
|
||||
if attach:
|
||||
# Create a copy of our payload
|
||||
_payload = payload.copy()
|
||||
|
||||
# Always call throttle before any remote server i/o is made
|
||||
self.throttle()
|
||||
# Send with attachments
|
||||
for attachment in attach:
|
||||
# Simple send
|
||||
if not self._send(_payload, attachment):
|
||||
# Mark our failure
|
||||
has_error = True
|
||||
# clean exit from our attachment loop
|
||||
break
|
||||
|
||||
try:
|
||||
r = requests.post(
|
||||
self.notify_url,
|
||||
data=payload,
|
||||
headers=headers,
|
||||
auth=auth,
|
||||
verify=self.verify_certificate,
|
||||
)
|
||||
if r.status_code != requests.codes.ok:
|
||||
# We had a problem
|
||||
status_str = \
|
||||
NotifyPushover.http_response_code_lookup(
|
||||
r.status_code, PUSHOVER_HTTP_ERROR_MAP)
|
||||
|
||||
self.logger.warning(
|
||||
'Failed to send Pushover notification to {}: '
|
||||
'{}{}error={}.'.format(
|
||||
device,
|
||||
status_str,
|
||||
', ' if status_str else '',
|
||||
r.status_code))
|
||||
|
||||
self.logger.debug(
|
||||
'Response Details:\r\n{}'.format(r.content))
|
||||
# To handle multiple attachments, clean up our message
|
||||
_payload['title'] = '...'
|
||||
_payload['message'] = attachment.name
|
||||
# No need to alarm for each consecutive attachment uploaded
|
||||
# afterwards
|
||||
_payload['sound'] = PushoverSound.NONE
|
||||
|
||||
else:
|
||||
# Simple send
|
||||
if not self._send(payload):
|
||||
# Mark our failure
|
||||
has_error = True
|
||||
continue
|
||||
|
||||
else:
|
||||
self.logger.info(
|
||||
'Sent Pushover notification to %s.' % device)
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A Connection error occured sending Pushover:%s ' % (
|
||||
device) + 'notification.'
|
||||
)
|
||||
self.logger.debug('Socket Exception: %s' % str(e))
|
||||
|
||||
# Mark our failure
|
||||
has_error = True
|
||||
continue
|
||||
|
||||
return not has_error
|
||||
|
||||
def _send(self, payload, attach=None):
|
||||
"""
|
||||
Wrapper to the requests (post) object
|
||||
"""
|
||||
|
||||
if isinstance(attach, AttachBase):
|
||||
# Perform some simple error checking
|
||||
if not attach:
|
||||
# We could not access the attachment
|
||||
self.logger.warning(
|
||||
'Could not access {}.'.format(
|
||||
attach.url(privacy=True)))
|
||||
return False
|
||||
|
||||
# Perform some basic checks as we want to gracefully skip
|
||||
# over unsupported mime types.
|
||||
if not re.match(
|
||||
self.attach_supported_mime_type,
|
||||
attach.mimetype,
|
||||
re.I):
|
||||
# No problem; we just don't support this attachment
|
||||
# type; gracefully move along
|
||||
self.logger.debug(
|
||||
'Ignored unsupported Pushover attachment ({}): {}'
|
||||
.format(
|
||||
attach.mimetype,
|
||||
attach.url(privacy=True)))
|
||||
|
||||
return True
|
||||
|
||||
# If we get here, we're dealing with a supported image.
|
||||
# Verify that the filesize is okay though.
|
||||
file_size = len(attach)
|
||||
if not (file_size > 0
|
||||
and file_size <= self.attach_max_size_bytes):
|
||||
|
||||
# File size is no good
|
||||
self.logger.warning(
|
||||
'Pushover attachment size ({}B) exceeds limit: {}'
|
||||
.format(file_size, attach.url(privacy=True)))
|
||||
|
||||
return False
|
||||
|
||||
# Default Header
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
}
|
||||
|
||||
# Authentication
|
||||
auth = (self.token, '')
|
||||
|
||||
# Some default values for our request object to which we'll update
|
||||
# depending on what our payload is
|
||||
files = None
|
||||
|
||||
self.logger.debug('Pushover POST URL: %s (cert_verify=%r)' % (
|
||||
self.notify_url, self.verify_certificate,
|
||||
))
|
||||
self.logger.debug('Pushover Payload: %s' % str(payload))
|
||||
|
||||
# Always call throttle before any remote server i/o is made
|
||||
self.throttle()
|
||||
|
||||
try:
|
||||
# Open our attachment path if required:
|
||||
if attach:
|
||||
files = {'attachment': (attach.name, open(attach.path, 'rb'))}
|
||||
|
||||
r = requests.post(
|
||||
self.notify_url,
|
||||
data=payload,
|
||||
headers=headers,
|
||||
files=files,
|
||||
auth=auth,
|
||||
verify=self.verify_certificate,
|
||||
)
|
||||
|
||||
if r.status_code != requests.codes.ok:
|
||||
# We had a problem
|
||||
status_str = \
|
||||
NotifyPushover.http_response_code_lookup(
|
||||
r.status_code, PUSHOVER_HTTP_ERROR_MAP)
|
||||
|
||||
self.logger.warning(
|
||||
'Failed to send Pushover notification to {}: '
|
||||
'{}{}error={}.'.format(
|
||||
payload['device'],
|
||||
status_str,
|
||||
', ' if status_str else '',
|
||||
r.status_code))
|
||||
|
||||
self.logger.debug(
|
||||
'Response Details:\r\n{}'.format(r.content))
|
||||
|
||||
return False
|
||||
|
||||
else:
|
||||
self.logger.info(
|
||||
'Sent Pushover notification to %s.' % payload['device'])
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
'A Connection error occured sending Pushover:%s ' % (
|
||||
payload['device']) + 'notification.'
|
||||
)
|
||||
self.logger.debug('Socket Exception: %s' % str(e))
|
||||
|
||||
return False
|
||||
|
||||
except (OSError, IOError) as e:
|
||||
self.logger.warning(
|
||||
'An I/O error occured while reading {}.'.format(
|
||||
attach.name if attach else 'attachment'))
|
||||
self.logger.debug('I/O Exception: %s' % str(e))
|
||||
return False
|
||||
|
||||
finally:
|
||||
# Close our file (if it's open) stored in the second element
|
||||
# of our files tuple (index 1)
|
||||
if files:
|
||||
files['attachment'][1].close()
|
||||
|
||||
return True
|
||||
|
||||
def url(self, privacy=False, *args, **kwargs):
|
||||
"""
|
||||
Returns the URL built dynamically based on specified arguments.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,107 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# All rights reserved.
|
||||
#
|
||||
# This code is licensed under the MIT License.
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files(the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions :
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in
|
||||
# all copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
# THE SOFTWARE.
|
||||
|
||||
import os
|
||||
import mock
|
||||
import requests
|
||||
from json import dumps
|
||||
from apprise import Apprise
|
||||
from apprise import AppriseAttachment
|
||||
from apprise import plugins
|
||||
|
||||
# Disable logging for a cleaner testing output
|
||||
import logging
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
# Attachment Directory
|
||||
TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var')
|
||||
|
||||
|
||||
@mock.patch('requests.post')
|
||||
def test_pushover_attachments(mock_post, tmpdir):
|
||||
"""
|
||||
API: NotifyPushover() Attachment Checks
|
||||
|
||||
"""
|
||||
# Disable Throttling to speed testing
|
||||
plugins.NotifyBase.request_rate_per_sec = 0
|
||||
|
||||
# Initialize some generic (but valid) tokens
|
||||
user_key = 'u' * 30
|
||||
api_token = 'a' * 30
|
||||
|
||||
# Prepare Mock return object
|
||||
response = mock.Mock()
|
||||
response.content = dumps(
|
||||
{"status": 1, "request": "647d2300-702c-4b38-8b2f-d56326ae460b"})
|
||||
response.status_code = requests.codes.ok
|
||||
mock_post.return_value = response
|
||||
|
||||
# prepare our attachment
|
||||
attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'))
|
||||
|
||||
# Instantiate our object
|
||||
obj = Apprise.instantiate(
|
||||
'pover://{}@{}/'.format(user_key, api_token))
|
||||
assert isinstance(obj, plugins.NotifyPushover)
|
||||
|
||||
# Test our attachment
|
||||
assert obj.notify(body="test", attach=attach) is True
|
||||
|
||||
# Test multiple attachments
|
||||
assert attach.add(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'))
|
||||
assert obj.notify(body="test", attach=attach) is True
|
||||
|
||||
image = tmpdir.mkdir("pover_image").join("test.jpg")
|
||||
image.write('a' * plugins.NotifyPushover.attach_max_size_bytes)
|
||||
|
||||
attach = AppriseAttachment.instantiate(str(image))
|
||||
assert obj.notify(body="test", attach=attach) is True
|
||||
|
||||
# Add 1 more byte to the file (putting it over the limit)
|
||||
image.write('a' * (plugins.NotifyPushover.attach_max_size_bytes + 1))
|
||||
|
||||
attach = AppriseAttachment.instantiate(str(image))
|
||||
assert obj.notify(body="test", attach=attach) is False
|
||||
|
||||
# Test case when file is missing
|
||||
attach = AppriseAttachment.instantiate(
|
||||
'file://{}?cache=False'.format(str(image)))
|
||||
os.unlink(str(image))
|
||||
assert obj.notify(
|
||||
body='body', title='title', attach=attach) is False
|
||||
|
||||
# Test unsuported files:
|
||||
image = tmpdir.mkdir("pover_unsupported").join("test.doc")
|
||||
image.write('a' * 256)
|
||||
attach = AppriseAttachment.instantiate(str(image))
|
||||
|
||||
# Content is silently ignored
|
||||
assert obj.notify(body="test", attach=attach) is True
|
||||
|
||||
# Throw an exception on the second call to requests.post()
|
||||
mock_post.side_effect = OSError()
|
||||
# prepare our attachment
|
||||
attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif'))
|
||||
assert obj.notify(body="test", attach=attach) is False
|
||||
Loading…
Reference in New Issue