mirror of
https://github.com/caronc/apprise.git
synced 2025-01-03 20:49:19 +01:00
Implimented overflow support (upstream, truncate, and split)
This commit is contained in:
Chris Caron
parent
0d56da9ac8
commit
cdd72086ee
@ -363,26 +363,30 @@ class Apprise(object):
|
||||
# Store entry directly
|
||||
conversion_map[server.notify_format] = body
|
||||
|
||||
try:
|
||||
# Send notification
|
||||
if not server.notify(
|
||||
title=title,
|
||||
body=conversion_map[server.notify_format],
|
||||
notify_type=notify_type):
|
||||
# Apply our overflow (if defined)
|
||||
for chunk in server._apply_overflow(
|
||||
body=conversion_map[server.notify_format], title=title):
|
||||
try:
|
||||
# Send notification
|
||||
if not server.notify(
|
||||
title=chunk['title'],
|
||||
body=chunk['body'],
|
||||
notify_type=notify_type):
|
||||
|
||||
# Toggle our return status flag
|
||||
# Toggle our return status flag
|
||||
status = False
|
||||
|
||||
except TypeError:
|
||||
# These our our internally thrown notifications
|
||||
# TODO: Change this to a custom one such as
|
||||
# AppriseNotifyError
|
||||
status = False
|
||||
|
||||
except TypeError:
|
||||
# These our our internally thrown notifications
|
||||
# TODO: Change this to a custom one such as AppriseNotifyError
|
||||
status = False
|
||||
|
||||
except Exception:
|
||||
# A catch all so we don't have to abort early
|
||||
# just because one of our plugins has a bug in it.
|
||||
logging.exception("Notification Exception")
|
||||
status = False
|
||||
except Exception:
|
||||
# A catch all so we don't have to abort early
|
||||
# just because one of our plugins has a bug in it.
|
||||
logging.exception("Notification Exception")
|
||||
status = False
|
||||
|
||||
return status
|
||||
|
||||
|
||||
@ -37,6 +37,8 @@ from .common import NotifyImageSize
|
||||
from .common import NOTIFY_IMAGE_SIZES
|
||||
from .common import NotifyFormat
|
||||
from .common import NOTIFY_FORMATS
|
||||
from .common import OverflowMode
|
||||
from .common import OVERFLOW_MODES
|
||||
from .plugins.NotifyBase import NotifyBase
|
||||
|
||||
from .Apprise import Apprise
|
||||
@ -52,6 +54,6 @@ __all__ = [
|
||||
'Apprise', 'AppriseAsset', 'NotifyBase',
|
||||
|
||||
# Reference
|
||||
'NotifyType', 'NotifyImageSize', 'NotifyFormat', 'NOTIFY_TYPES',
|
||||
'NOTIFY_IMAGE_SIZES', 'NOTIFY_FORMATS',
|
||||
'NotifyType', 'NotifyImageSize', 'NotifyFormat', 'OverflowMode',
|
||||
'NOTIFY_TYPES', 'NOTIFY_IMAGE_SIZES', 'NOTIFY_FORMATS', 'OVERFLOW_MODES',
|
||||
]
|
||||
|
||||
@ -77,3 +77,31 @@ NOTIFY_FORMATS = (
|
||||
NotifyFormat.HTML,
|
||||
NotifyFormat.MARKDOWN,
|
||||
)
|
||||
|
||||
|
||||
class OverflowMode(object):
|
||||
"""
|
||||
A list of pre-defined modes of how to handle the text when it exceeds the
|
||||
defined maximum message size.
|
||||
"""
|
||||
|
||||
# Send the data as is; untouched. Let the upstream server decide how the
|
||||
# content is handled. Some upstream services might gracefully handle this
|
||||
# with expected intentions; others might not.
|
||||
UPSTREAM = 'upstream'
|
||||
|
||||
# Always truncate the text when it exceeds the maximum message size and
|
||||
# send it anyway
|
||||
TRUNCATE = 'truncate'
|
||||
|
||||
# Split the message into multiple smaller messages that fit within the
|
||||
# limits of what is expected. The smaller messages are sent
|
||||
SPLIT = 'split'
|
||||
|
||||
|
||||
# Define our modes so we can verify if we need to
|
||||
OVERFLOW_MODES = (
|
||||
OverflowMode.UPSTREAM,
|
||||
OverflowMode.TRUNCATE,
|
||||
OverflowMode.SPLIT,
|
||||
)
|
||||
|
||||
@ -45,6 +45,8 @@ from ..utils import is_hostname
|
||||
from ..common import NOTIFY_TYPES
|
||||
from ..common import NotifyFormat
|
||||
from ..common import NOTIFY_FORMATS
|
||||
from ..common import OverflowMode
|
||||
from ..common import OVERFLOW_MODES
|
||||
|
||||
from ..AppriseAsset import AppriseAsset
|
||||
|
||||
@ -120,14 +122,26 @@ class NotifyBase(object):
|
||||
image_size = None
|
||||
|
||||
# The maximum allowable characters allowed in the body per message
|
||||
body_maxlen = 32768
|
||||
# We set it to what would virtually be an infinite value really
|
||||
# 2^63 - 1 = 9223372036854775807
|
||||
body_maxlen = 9223372036854775807
|
||||
|
||||
# Defines the maximum allowable characters in the title
|
||||
# Defines the maximum allowable characters in the title; set this to zero
|
||||
# if a title can't be used. Titles that are not used but are defined are
|
||||
# automatically placed into the body
|
||||
title_maxlen = 250
|
||||
|
||||
# Set the maximum line count; if this is set to anything larger then zero
|
||||
# the message (prior to it being sent) will be truncated to this number
|
||||
# of lines. Setting this to zero disables this feature.
|
||||
body_max_line_count = 0
|
||||
|
||||
# Default Notify Format
|
||||
notify_format = NotifyFormat.TEXT
|
||||
|
||||
# Default Overflow Mode
|
||||
overflow_mode = OverflowMode.UPSTREAM
|
||||
|
||||
# Maintain a set of tags to associate with this specific notification
|
||||
tags = set()
|
||||
|
||||
@ -177,6 +191,19 @@ class NotifyBase(object):
|
||||
# Provide override
|
||||
self.notify_format = notify_format
|
||||
|
||||
if 'overflow' in kwargs:
|
||||
# Store the specified format if specified
|
||||
overflow_mode = kwargs.get('overflow', '')
|
||||
if overflow_mode.lower() not in OVERFLOW_MODES:
|
||||
self.logger.error(
|
||||
'Invalid overflow method %s' % overflow_mode,
|
||||
)
|
||||
raise TypeError(
|
||||
'Invalid overflow method %s' % overflow_mode,
|
||||
)
|
||||
# Provide override
|
||||
self.overflow_mode = overflow_mode
|
||||
|
||||
if 'tag' in kwargs:
|
||||
# We want to associate some tags with our notification service.
|
||||
# the code below gets the 'tag' argument if defined, otherwise
|
||||
@ -260,6 +287,78 @@ class NotifyBase(object):
|
||||
color_type=color_type,
|
||||
)
|
||||
|
||||
def _apply_overflow(self, body, title=None):
|
||||
"""
|
||||
Takes the message body and title as input. This function then
|
||||
applies any defined overflow restrictions associated with the
|
||||
notification service and may alter the message if/as required.
|
||||
|
||||
The function will always return a list object in the following
|
||||
structure:
|
||||
[
|
||||
{
|
||||
title: 'the title goes here',
|
||||
body: 'the message body goes here',
|
||||
},
|
||||
{
|
||||
title: 'the title goes here',
|
||||
body: 'the message body goes here',
|
||||
},
|
||||
|
||||
]
|
||||
"""
|
||||
|
||||
response = list()
|
||||
|
||||
if self.title_maxlen <= 0:
|
||||
# Content is appended to body
|
||||
body = '{}\r\n{}'.format(title, body)
|
||||
title = ''
|
||||
|
||||
# Enforce the line count first always
|
||||
if self.body_max_line_count > 0:
|
||||
# Limit results to just the first 2 line otherwise
|
||||
# there is just to much content to display
|
||||
body = re.split('\r*\n', body)
|
||||
body = '\r\n'.join(body[0:self.body_max_line_count])
|
||||
|
||||
if self.overflow_mode == OverflowMode.UPSTREAM:
|
||||
# Nothing to do
|
||||
response.append({
|
||||
'body': body,
|
||||
'title': title,
|
||||
})
|
||||
return response
|
||||
|
||||
elif len(title) > self.title_maxlen:
|
||||
# Truncate our Title
|
||||
title = title[:self.title_maxlen]
|
||||
|
||||
if self.body_maxlen > 0 and len(body) <= self.body_maxlen:
|
||||
response.append({
|
||||
'body': body,
|
||||
'title': title,
|
||||
})
|
||||
return response
|
||||
|
||||
if self.overflow_mode == OverflowMode.TRUNCATE:
|
||||
# Truncate our body and return
|
||||
response.append({
|
||||
'body': body[:self.body_maxlen],
|
||||
'title': title,
|
||||
})
|
||||
# For truncate mode, we're done now
|
||||
return response
|
||||
|
||||
# If we reach here, then we are in SPLIT mode.
|
||||
# For here, we want to split the message as many times as we have to
|
||||
# in order to fit it within the designated limits.
|
||||
response = [{
|
||||
'body': body[i: i + self.body_maxlen],
|
||||
'title': title} for i in range(0, len(body), self.body_maxlen)]
|
||||
|
||||
return response
|
||||
|
||||
def url(self):
|
||||
"""
|
||||
Assembles the URL associated with the notification based on the
|
||||
|
||||
@ -280,7 +280,8 @@ class NotifyBoxcar(NotifyBase):
|
||||
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
return '{schema}://{access}/{secret}/{recipients}/?{args}'.format(
|
||||
|
||||
@ -148,6 +148,14 @@ class NotifyDBus(NotifyBase):
|
||||
# The number of seconds to keep the message present for
|
||||
message_timeout_ms = 13000
|
||||
|
||||
# Limit results to just the first 10 line otherwise there is just to much
|
||||
# content to display
|
||||
body_max_line_count = 10
|
||||
|
||||
# A title can not be used for SMS Messages. Setting this to zero will
|
||||
# cause any title (if defined) to get placed into the message body.
|
||||
title_maxlen = 0
|
||||
|
||||
# This entry is a bit hacky, but it allows us to unit-test this library
|
||||
# in an environment that simply doesn't have the gnome packages
|
||||
# available to us. It also allows us to handle situations where the
|
||||
@ -249,15 +257,6 @@ class NotifyDBus(NotifyBase):
|
||||
"Could not load Gnome notification icon ({}): {}"
|
||||
.format(icon_path, e))
|
||||
|
||||
# Limit results to just the first 10 line otherwise
|
||||
# there is just to much content to display
|
||||
body = re.split('[\r\n]+', body)
|
||||
if title:
|
||||
# Place title on first line if it exists
|
||||
body.insert(0, title)
|
||||
|
||||
body = '\r\n'.join(body[0:10])
|
||||
|
||||
try:
|
||||
dbus_iface.Notify(
|
||||
# Application Identifier
|
||||
|
||||
@ -249,6 +249,7 @@ class NotifyDiscord(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'tts': 'yes' if self.tts else 'no',
|
||||
'avatar': 'yes' if self.avatar else 'no',
|
||||
'footer': 'yes' if self.footer else 'no',
|
||||
|
||||
@ -429,6 +429,7 @@ class NotifyEmail(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'to': self.to_addr,
|
||||
'from': self.from_addr,
|
||||
'name': self.from_name,
|
||||
|
||||
@ -543,6 +543,7 @@ class NotifyEmby(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'modal': 'yes' if self.modal else 'no',
|
||||
}
|
||||
|
||||
|
||||
@ -132,6 +132,7 @@ class NotifyFaast(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
return '{schema}://{authtoken}/?{args}'.format(
|
||||
|
||||
@ -26,8 +26,6 @@
|
||||
from __future__ import absolute_import
|
||||
from __future__ import print_function
|
||||
|
||||
import re
|
||||
|
||||
from .NotifyBase import NotifyBase
|
||||
from ..common import NotifyImageSize
|
||||
|
||||
@ -86,6 +84,14 @@ class NotifyGnome(NotifyBase):
|
||||
# Allows the user to specify the NotifyImageSize object
|
||||
image_size = NotifyImageSize.XY_128
|
||||
|
||||
# Limit results to just the first 10 line otherwise there is just to much
|
||||
# content to display
|
||||
body_max_line_count = 10
|
||||
|
||||
# A title can not be used for SMS Messages. Setting this to zero will
|
||||
# cause any title (if defined) to get placed into the message body.
|
||||
title_maxlen = 0
|
||||
|
||||
# This entry is a bit hacky, but it allows us to unit-test this library
|
||||
# in an environment that simply doesn't have the gnome packages
|
||||
# available to us. It also allows us to handle situations where the
|
||||
@ -119,15 +125,6 @@ class NotifyGnome(NotifyBase):
|
||||
"Gnome Notifications are not supported by this system.")
|
||||
return False
|
||||
|
||||
# Limit results to just the first 10 line otherwise
|
||||
# there is just to much content to display
|
||||
body = re.split('[\r\n]+', body)
|
||||
if title:
|
||||
# Place title on first line if it exists
|
||||
body.insert(0, title)
|
||||
|
||||
body = '\r\n'.join(body[0:10])
|
||||
|
||||
try:
|
||||
# App initialization
|
||||
Notify.init(self.app_id)
|
||||
|
||||
@ -223,6 +223,7 @@ class NotifyGrowl(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'priority':
|
||||
_map[GrowlPriority.NORMAL] if self.priority not in _map
|
||||
else _map[self.priority],
|
||||
|
||||
@ -214,6 +214,7 @@ class NotifyIFTTT(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
return '{schema}://{webhook_id}@{events}/?{args}'.format(
|
||||
|
||||
@ -78,6 +78,7 @@ class NotifyJSON(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
# Determine Authentication
|
||||
|
||||
@ -90,6 +90,10 @@ class NotifyJoin(NotifyBase):
|
||||
# Allows the user to specify the NotifyImageSize object
|
||||
image_size = NotifyImageSize.XY_72
|
||||
|
||||
# Limit results to just the first 2 line otherwise there is just to much
|
||||
# content to display
|
||||
body_max_line_count = 2
|
||||
|
||||
# The maximum allowable characters allowed in the body per message
|
||||
body_maxlen = 1000
|
||||
|
||||
@ -131,17 +135,6 @@ class NotifyJoin(NotifyBase):
|
||||
Perform Join Notification
|
||||
"""
|
||||
|
||||
try:
|
||||
# Limit results to just the first 2 line otherwise
|
||||
# there is just to much content to display
|
||||
body = re.split('[\r\n]+', body)
|
||||
body[0] = body[0].strip('#').strip()
|
||||
body = '\r\n'.join(body[0:2])
|
||||
|
||||
except (AttributeError, TypeError):
|
||||
# body was None or not of a type string
|
||||
body = ''
|
||||
|
||||
headers = {
|
||||
'User-Agent': self.app_id,
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
@ -241,6 +234,7 @@ class NotifyJoin(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
return '{schema}://{apikey}/{devices}/?{args}'.format(
|
||||
|
||||
@ -244,6 +244,7 @@ class NotifyMatrix(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'mode': self.mode,
|
||||
}
|
||||
|
||||
|
||||
@ -187,6 +187,7 @@ class NotifyMatterMost(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
default_port = 443 if self.secure else self.default_port
|
||||
|
||||
@ -206,6 +206,7 @@ class NotifyProwl(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'priority': 'normal' if self.priority not in _map
|
||||
else _map[self.priority]
|
||||
}
|
||||
|
||||
@ -190,6 +190,7 @@ class NotifyPushBullet(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
recipients = '/'.join([self.quote(x) for x in self.recipients])
|
||||
|
||||
@ -265,6 +265,7 @@ class NotifyPushed(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
return '{schema}://{app_key}/{app_secret}/{targets}/?{args}'.format(
|
||||
|
||||
@ -95,6 +95,7 @@ class NotifyPushjet(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
default_port = 443 if self.secure else 80
|
||||
|
||||
@ -253,6 +253,7 @@ class NotifyPushover(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'priority':
|
||||
_map[PushoverPriority.NORMAL] if self.priority not in _map
|
||||
else _map[self.priority],
|
||||
|
||||
@ -148,6 +148,7 @@ class NotifyRocketChat(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
# Determine Authentication
|
||||
|
||||
@ -229,6 +229,7 @@ class NotifyRyver(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'webhook': self.webhook,
|
||||
}
|
||||
|
||||
|
||||
@ -90,6 +90,10 @@ class NotifySNS(NotifyBase):
|
||||
# Source: https://docs.aws.amazon.com/sns/latest/api/API_Publish.html
|
||||
body_maxlen = 140
|
||||
|
||||
# A title can not be used for SMS Messages. Setting this to zero will
|
||||
# cause any title (if defined) to get placed into the message body.
|
||||
title_maxlen = 0
|
||||
|
||||
def __init__(self, access_key_id, secret_access_key, region_name,
|
||||
recipients=None, **kwargs):
|
||||
"""
|
||||
@ -530,6 +534,7 @@ class NotifySNS(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
return '{schema}://{key_id}/{key_secret}/{region}/{targets}/'\
|
||||
|
||||
@ -304,6 +304,7 @@ class NotifySlack(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
# Determine if there is a botname present
|
||||
|
||||
@ -498,6 +498,7 @@ class NotifyTelegram(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
# No need to check the user token because the user automatically gets
|
||||
|
||||
@ -26,7 +26,6 @@
|
||||
from __future__ import absolute_import
|
||||
from __future__ import print_function
|
||||
|
||||
import re
|
||||
from time import sleep
|
||||
|
||||
from .NotifyBase import NotifyBase
|
||||
@ -67,6 +66,10 @@ class NotifyWindows(NotifyBase):
|
||||
# Allows the user to specify the NotifyImageSize object
|
||||
image_size = NotifyImageSize.XY_128
|
||||
|
||||
# Limit results to just the first 2 line otherwise there is just to much
|
||||
# content to display
|
||||
body_max_line_count = 2
|
||||
|
||||
# This entry is a bit hacky, but it allows us to unit-test this library
|
||||
# in an environment that simply doesn't have the windows packages
|
||||
# available to us. It also allows us to handle situations where the
|
||||
@ -110,12 +113,6 @@ class NotifyWindows(NotifyBase):
|
||||
"Windows Notifications are not supported by this system.")
|
||||
return False
|
||||
|
||||
# Limit results to just the first 2 line otherwise
|
||||
# there is just to much content to display
|
||||
body = re.split('[\r\n]+', body)
|
||||
body[0] = body[0].strip('#').strip()
|
||||
body = '\r\n'.join(body[0:2])
|
||||
|
||||
try:
|
||||
# Register destruction callback
|
||||
message_map = {win32con.WM_DESTROY: self._on_destroy, }
|
||||
|
||||
@ -23,7 +23,6 @@
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
# THE SOFTWARE.
|
||||
|
||||
import re
|
||||
import requests
|
||||
from json import dumps
|
||||
|
||||
@ -58,6 +57,10 @@ class NotifyXBMC(NotifyBase):
|
||||
# A URL that takes you to the setup/help of the specific protocol
|
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_kodi'
|
||||
|
||||
# Limit results to just the first 2 line otherwise there is just to much
|
||||
# content to display
|
||||
body_max_line_count = 2
|
||||
|
||||
# XBMC uses the http protocol with JSON requests
|
||||
xbmc_default_port = 8080
|
||||
|
||||
@ -159,12 +162,6 @@ class NotifyXBMC(NotifyBase):
|
||||
Perform XBMC/KODI Notification
|
||||
"""
|
||||
|
||||
# Limit results to just the first 2 line otherwise
|
||||
# there is just to much content to display
|
||||
body = re.split('[\r\n]+', body)
|
||||
body[0] = body[0].strip('#').strip()
|
||||
body = '\r\n'.join(body[0:2])
|
||||
|
||||
if self.protocol == self.xbmc_remote_protocol:
|
||||
# XBMC v2.0
|
||||
(headers, payload) = self._payload_20(
|
||||
@ -237,6 +234,7 @@ class NotifyXBMC(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
# Determine Authentication
|
||||
|
||||
@ -93,6 +93,7 @@ class NotifyXML(NotifyBase):
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
}
|
||||
|
||||
# Determine Authentication
|
||||
|
||||
@ -25,12 +25,18 @@
|
||||
|
||||
from apprise import plugins
|
||||
from apprise import NotifyType
|
||||
from apprise import NotifyBase
|
||||
from apprise import Apprise
|
||||
from apprise import AppriseAsset
|
||||
from apprise.utils import compat_is_basestring
|
||||
from apprise.common import NotifyFormat
|
||||
from apprise.common import OverflowMode
|
||||
|
||||
from json import dumps
|
||||
from random import choice
|
||||
from string import ascii_uppercase as str_alpha
|
||||
from string import digits as str_num
|
||||
|
||||
import requests
|
||||
import mock
|
||||
|
||||
@ -2883,3 +2889,338 @@ def test_notify_telegram_plugin(mock_post, mock_get):
|
||||
except TypeError:
|
||||
# Exception should be thrown about the fact no token was specified
|
||||
assert(True)
|
||||
|
||||
|
||||
def test_notify_overflow_truncate():
|
||||
"""
|
||||
API: Overflow Truncate Functionality Testing
|
||||
|
||||
"""
|
||||
#
|
||||
# A little preparation
|
||||
#
|
||||
|
||||
# Number of characters per line
|
||||
row = 24
|
||||
|
||||
# Some variables we use to control the data we work with
|
||||
body_len = 1024
|
||||
title_len = 1024
|
||||
|
||||
# Create a large body and title with random data
|
||||
body = ''.join(choice(str_alpha + str_num + ' ') for _ in range(body_len))
|
||||
body = '\r\n'.join([body[i: i + row] for i in range(0, len(body), row)])
|
||||
|
||||
# the new lines add a large amount to our body; lets force the content
|
||||
# back to being 1024 characters.
|
||||
body = body[0:1024]
|
||||
|
||||
# Create our title using random data
|
||||
title = ''.join(choice(str_alpha + str_num) for _ in range(title_len))
|
||||
|
||||
#
|
||||
# First Test: Truncated Title
|
||||
#
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Test title max length
|
||||
title_maxlen = 10
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
try:
|
||||
# Load our object
|
||||
obj = TestNotification(overflow='invalid')
|
||||
|
||||
# We should have thrown an exception because our specified overflow
|
||||
# is wrong.
|
||||
assert False
|
||||
|
||||
except TypeError:
|
||||
# Expected to be here
|
||||
assert True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.TRUNCATE)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
assert len(chunks) == 1
|
||||
assert body == chunks[0].get('body')
|
||||
assert title[0:TestNotification.title_maxlen] == chunks[0].get('title')
|
||||
|
||||
#
|
||||
# Next Test: Line Count Control
|
||||
#
|
||||
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Test title max length
|
||||
title_maxlen = 5
|
||||
|
||||
# Maximum number of lines
|
||||
body_max_line_count = 5
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.TRUNCATE)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
assert len(chunks) == 1
|
||||
assert len(chunks[0].get('body').split('\n')) == \
|
||||
TestNotification.body_max_line_count
|
||||
assert title[0:TestNotification.title_maxlen] == chunks[0].get('title')
|
||||
|
||||
#
|
||||
# Next Test: Truncated body
|
||||
#
|
||||
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Test title max length
|
||||
title_maxlen = title_len
|
||||
|
||||
# Enforce a body length of just 10
|
||||
body_maxlen = 10
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.TRUNCATE)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
assert len(chunks) == 1
|
||||
assert body[0:TestNotification.body_maxlen] == chunks[0].get('body')
|
||||
assert title == chunks[0].get('title')
|
||||
|
||||
#
|
||||
# Next Test: Append title to body + Truncated body
|
||||
#
|
||||
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Enforce no title
|
||||
title_maxlen = 0
|
||||
|
||||
# Enforce a body length of just 100
|
||||
body_maxlen = 100
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.TRUNCATE)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
assert len(chunks) == 1
|
||||
|
||||
# The below line should be read carefully... We're actually testing to see
|
||||
# that our title is matched against our body. Behind the scenes, the title
|
||||
# was appended to the body. The body was then truncated to the maxlen.
|
||||
# The thing is, since the title is so large, all of the body was lost
|
||||
# and a good chunk of the title was too. The message sent will just be a
|
||||
# small portion of the title
|
||||
assert len(chunks[0].get('body')) == TestNotification.body_maxlen
|
||||
assert title[0:TestNotification.body_maxlen] == chunks[0].get('body')
|
||||
|
||||
|
||||
def test_notify_overflow_split():
|
||||
"""
|
||||
API: Overflow Split Functionality Testing
|
||||
|
||||
"""
|
||||
|
||||
#
|
||||
# A little preparation
|
||||
#
|
||||
|
||||
# Number of characters per line
|
||||
row = 24
|
||||
|
||||
# Some variables we use to control the data we work with
|
||||
body_len = 1024
|
||||
title_len = 1024
|
||||
|
||||
# Create a large body and title with random data
|
||||
body = ''.join(choice(str_alpha + str_num + ' ') for _ in range(body_len))
|
||||
body = '\r\n'.join([body[i: i + row] for i in range(0, len(body), row)])
|
||||
|
||||
# the new lines add a large amount to our body; lets force the content
|
||||
# back to being 1024 characters.
|
||||
body = body[0:1024]
|
||||
|
||||
# Create our title using random data
|
||||
title = ''.join(choice(str_alpha + str_num) for _ in range(title_len))
|
||||
|
||||
#
|
||||
# First Test: Truncated Title
|
||||
#
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Test title max length
|
||||
title_maxlen = 10
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.SPLIT)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
assert len(chunks) == 1
|
||||
assert body == chunks[0].get('body')
|
||||
assert title[0:TestNotification.title_maxlen] == chunks[0].get('title')
|
||||
|
||||
#
|
||||
# Next Test: Line Count Control
|
||||
#
|
||||
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Test title max length
|
||||
title_maxlen = 5
|
||||
|
||||
# Maximum number of lines
|
||||
body_max_line_count = 5
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.SPLIT)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
assert len(chunks) == 1
|
||||
assert len(chunks[0].get('body').split('\n')) == \
|
||||
TestNotification.body_max_line_count
|
||||
assert title[0:TestNotification.title_maxlen] == chunks[0].get('title')
|
||||
|
||||
#
|
||||
# Next Test: Split body
|
||||
#
|
||||
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Test title max length
|
||||
title_maxlen = title_len
|
||||
|
||||
# Enforce a body length
|
||||
body_maxlen = (body_len / 4)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.SPLIT)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
offset = 0
|
||||
assert len(chunks) == 4
|
||||
for chunk in chunks:
|
||||
# Our title never changes
|
||||
assert title == chunk.get('title')
|
||||
|
||||
# Our body is only broken up; not lost
|
||||
_body = chunk.get('body')
|
||||
assert body[offset: len(_body) + offset] == _body
|
||||
offset += len(_body)
|
||||
|
||||
#
|
||||
# Next Test: Append title to body + split body
|
||||
#
|
||||
|
||||
class TestNotification(NotifyBase):
|
||||
|
||||
# Enforce no title
|
||||
title_maxlen = 0
|
||||
|
||||
# Enforce a body length
|
||||
body_maxlen = (title_len / 4)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(TestNotification, self).__init__(**kwargs)
|
||||
|
||||
def notify(self, *args, **kwargs):
|
||||
# Pretend everything is okay
|
||||
return True
|
||||
|
||||
# Load our object
|
||||
obj = TestNotification(overflow=OverflowMode.SPLIT)
|
||||
assert obj is not None
|
||||
|
||||
# Verify that we break the title to a max length of our title_max
|
||||
# and that the body remains untouched
|
||||
chunks = obj._apply_overflow(body=body, title=title)
|
||||
|
||||
# Our final product is that our title has been appended to our body to
|
||||
# create one great big body. As a result we'll get quite a few lines back
|
||||
# now.
|
||||
offset = 0
|
||||
|
||||
# Our body will look like this in small chunks at the end of the day
|
||||
bulk = title + '\r\n' + body
|
||||
|
||||
# Due to the new line added to the end
|
||||
assert len(chunks) == (
|
||||
(len(bulk) / TestNotification.body_maxlen) +
|
||||
(1 if len(bulk) % TestNotification.body_maxlen else 0))
|
||||
|
||||
for chunk in chunks:
|
||||
# Our title is empty every time
|
||||
assert chunk.get('title') == ''
|
||||
|
||||
_body = chunk.get('body')
|
||||
assert bulk[offset:len(_body)+offset] == _body
|
||||
offset += len(_body)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user