extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-11-22 08:04:02 +01:00
Code Issues Packages Projects Releases Wiki Activity

python v3 support + refactored testing and ci

Browse Source
This commit is contained in:
Chris Caron 2017-12-03 02:00:23 -05:00
parent 06c4654c95
commit 9c4502a18b
32 changed files with 694 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
.coveragerc Normal file
View File

@ -0,0 +1,15 @@
[run]
omit=*/gntp/*,*/tweepy/*,*/pushjet/*
disable_warnings = no-data-collected
branch = True
source =
apprise
[paths]
source =
apprise
.tox/*/lib/python*/site-packages/apprise
.tox/pypy/site-packages/apprise
[report]
show_missing = True

44
.travis.yml
View File

@ -1,15 +1,49 @@
dist: trusty
sudo: false
cache:
directories:
- $HOME/.cache/pip
language: python
python:
- "2.7"
matrix:
include:
- python: "2.7"
env: TOXENV=py27
- python: "3.4"
env: TOXENV=py34
- python: "3.5"
env: TOXENV=py35
- python: "3.6"
env: TOXENV=py36
- python: "pypy2.7-5.8.0"
env: TOXENV=pypy
- python: "pypy3.5-5.8.0"
env: TOXENV=pypy3
install:
- pip install .
- pip install coveralls
- pip install tox
- pip install -r dev-requirements.txt
- pip install -r requirements.txt
before_install:
- pip install codecov
after_success:
- coveralls
- tox -e coverage-report
- codecov
# run tests
script: nosetests --with-coverage --cover-package=apprise
script:
- tox
notifications:
email: false

7
README.md
View File

@ -1,3 +1,5 @@
![Apprise Logo](http://repo.nuxref.com/pub/img/logo-apprise.png)
<hr/>
**ap·prise** / *verb*<br/>
@ -6,7 +8,8 @@ To inform or tell (someone). To make one aware of something.
*Apprise* allows you to take advantage of *just about* every notification service available to us today. Send a notification to almost all of the most popular services out there today (such as Telegram, Slack, Twitter, etc). The ones that don't exist can be adapted and supported too!
[![Build Status](https://travis-ci.org/caronc/apprise.svg?branch=master)](https://travis-ci.org/caronc/apprise)[![Coverage Status](https://coveralls.io/repos/caronc/apprise/badge.svg?branch=master)](https://coveralls.io/r/caronc/apprise?branch=master)
[![Build Status](https://travis-ci.org/caronc/apprise.svg?branch=master)](https://travis-ci.org/caronc/apprise)
[![CodeCov Status](https://codecov.io/github/caronc/apprise/branch/master/graph/badge.svg)](https://codecov.io/github/caronc/apprise)
[![Paypal](http://repo.nuxref.com/pub/img/paypaldonate.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=MHANV39UZNQ5E)
[![Patreon](http://repo.nuxref.com/pub/img/patreondonate.svg)](https://www.patreon.com/lead2gold)
@ -76,7 +79,7 @@ To send a notification from within your python application, just do the followin
import apprise
# create an Apprise instance
apobj = Apprise()
apobj = apprise.Apprise()
# Add all of the notification services by their server url.
# A sample email notification

13
apprise/Apprise.py
View File

@ -23,8 +23,8 @@ import re
import logging
from .common import NotifyType
from .common import NOTIFY_TYPES
from .utils import parse_list
from .utils import compat_is_basestring
from .AppriseAsset import AppriseAsset
@ -54,7 +54,7 @@ def __load_matrix():
# Load protocol(s) if defined
proto = getattr(plugin, 'protocol', None)
if isinstance(proto, basestring):
if compat_is_basestring(proto):
if proto not in SCHEMA_MAP:
SCHEMA_MAP[proto] = plugin
@ -66,7 +66,7 @@ def __load_matrix():
# Load secure protocol(s) if defined
protos = getattr(plugin, 'secure_protocol', None)
if isinstance(protos, basestring):
if compat_is_basestring(protos):
if protos not in SCHEMA_MAP:
SCHEMA_MAP[protos] = plugin
@ -191,9 +191,9 @@ class Apprise(object):
"""
self.servers[:] = []
def notify(self, title, body, notify_type=NotifyType.SUCCESS, **kwargs):
def notify(self, title, body, notify_type=NotifyType.INFO):
"""
This should be over-rided by the class that inherits this one.
Send a notification to all of the plugins previously loaded
"""
# Initialize our return result
@ -206,7 +206,8 @@ class Apprise(object):
for server in self.servers:
try:
# Send notification
if not server.notify(title=title, body=body):
if not server.notify(
title=title, body=body, notify_type=notify_type):
# Toggle our return status flag
status = False

2
apprise/AppriseAsset.py
View File

@ -142,6 +142,6 @@ class AppriseAsset(object):
except (OSError, IOError):
# We can't access the file
pass
return None
return None

6
apprise/__init__.py
View File

@ -33,13 +33,7 @@ from .AppriseAsset import AppriseAsset
# Set default logging handler to avoid "No handler found" warnings.
import logging
try: # Python 2.7+
from logging import NullHandler
except ImportError:
class NullHandler(logging.Handler):
def emit(self, record):
pass
logging.getLogger(__name__).addHandler(NullHandler())
__all__ = [

113
apprise/plugins/NotifyBase.py
View File

@ -17,13 +17,19 @@
# GNU Lesser General Public License for more details.
import re
import markdown
import logging
from time import sleep
try:
# Python 2.7
from urllib import unquote as _unquote
from urllib import quote as _quote
from urllib import urlencode as _urlencode
# For conversion
from chardet import detect as chardet_detect
except ImportError:
# Python 3.x
from urllib.parse import unquote as _unquote
from urllib.parse import quote as _quote
from urllib.parse import urlencode as _urlencode
from ..utils import parse_url
from ..utils import parse_bool
@ -159,7 +165,7 @@ class NotifyBase(object):
self.include_image = include_image
self.secure = secure
if throttle:
if isinstance(throttle, (float, int)):
# Custom throttle override
self.throttle_attempt = throttle
@ -171,6 +177,7 @@ class NotifyBase(object):
if self.port:
try:
self.port = int(self.port)
except (TypeError, ValueError):
self.port = None
@ -238,7 +245,8 @@ class NotifyBase(object):
image_size=self.image_size,
)
def escape_html(self, html, convert_new_lines=False):
@staticmethod
def escape_html(html, convert_new_lines=False):
"""
Takes html text as input and escapes it so that it won't
conflict with any xml/html wrapping characters.
@ -248,76 +256,53 @@ class NotifyBase(object):
replace(u' ', u'&nbsp;')
if convert_new_lines:
return escaped.replace(u'\n', u'<br />')
return escaped.replace(u'\n', u'&lt;br/&gt;')
return escaped
def to_utf8(self, content):
@staticmethod
def unquote(content, encoding='utf-8', errors='replace'):
"""
Attempts to convert non-utf8 content to... (you guessed it) utf8
common unquote function
"""
if not content:
return ''
if isinstance(content, unicode):
return content.encode('utf-8')
result = chardet_detect(content)
encoding = result['encoding']
try:
content = content.decode(
encoding,
errors='replace',
)
return content.encode('utf-8')
except UnicodeError:
raise ValueError(
'%s contains invalid characters' % (
content))
except KeyError:
raise ValueError(
'%s encoding could not be detected ' % (
content))
# Python v3.x
return _unquote(content, encoding=encoding, errors=errors)
except TypeError:
# Python v2.7
return _unquote(content)
@staticmethod
def quote(content, safe='/', encoding=None, errors=None):
"""
common quote function
"""
try:
content = content.decode(
encoding,
'replace',
)
return content.encode('utf-8')
# Python v3.x
return _quote(content, safe=safe, encoding=encoding, errors=errors)
except UnicodeError:
raise ValueError(
'%s contains invalid characters' % (
content))
except TypeError:
# Python v2.7
return _quote(content, safe=safe)
except KeyError:
raise ValueError(
'%s encoding could not be detected ' % (
content))
return ''
def to_html(self, body):
@staticmethod
def urlencode(query, doseq=False, safe='', encoding=None, errors=None):
"""
Returns the specified title in an html format and factors
in a titles defined max length
common urlencode function
"""
html = markdown.markdown(body)
try:
# Python v3.x
return _urlencode(
query, doseq=doseq, safe=safe, encoding=encoding,
errors=errors)
# TODO:
# This function should return multiple messages if we exceed
# the maximum number of characters. the second message should
# The new message should factor in the title and add ' cont...'
# to the end of it. It should also include the added characters
# put in place by the html characters. So there is a little bit
# of math and manipulation that needs to go on here.
# we always return a list
return [html, ]
except TypeError:
# Python v2.7
return _urlencode(query, oseq=doseq)
@staticmethod
def split_path(path, unquote=True):
@ -326,7 +311,8 @@ class NotifyBase(object):
"""
if unquote:
return PATHSPLIT_LIST_DELIM.split(_unquote(path).lstrip('/'))
return PATHSPLIT_LIST_DELIM.split(
NotifyBase.unquote(path).lstrip('/'))
return PATHSPLIT_LIST_DELIM.split(path.lstrip('/'))
@staticmethod
@ -360,7 +346,8 @@ class NotifyBase(object):
if 'qsd' in results:
if 'verify' in results['qsd']:
parse_bool(results['qsd'].get('verify', True))
results['verify'] = parse_bool(
results['qsd'].get('verify', True))
# Password overrides
if 'pass' in results['qsd']:

9
apprise/plugins/NotifyBoxcar.py
View File

@ -17,13 +17,14 @@
# GNU Lesser General Public License for more details.
from json import dumps
from urllib import unquote
import requests
import re
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..utils import compat_is_basestring
# Used to validate Tags, Aliases and Devices
IS_TAG = re.compile(r'^[A-Za-z0-9]{1,63}$')
IS_ALIAS = re.compile(r'^[@]?[A-Za-z0-9]+$')
@ -70,12 +71,12 @@ class NotifyBoxcar(NotifyBase):
if recipients is None:
recipients = []
elif isinstance(recipients, basestring):
elif compat_is_basestring(recipients):
recipients = filter(bool, TAGS_LIST_DELIM.split(
recipients,
))
elif not isinstance(recipients, (tuple, list)):
elif not isinstance(recipients, (set, tuple, list)):
recipients = []
# Validate recipients and drop bad ones:
@ -189,7 +190,7 @@ class NotifyBoxcar(NotifyBase):
# Acquire our recipients and include them in the response
try:
recipients = unquote(results['fullpath'])
recipients = NotifyBase.unquote(results['fullpath'])
except (AttributeError, KeyError):
# no recipients detected

22
apprise/plugins/NotifyEmail.py
View File

@ -23,10 +23,9 @@ from smtplib import SMTP
from smtplib import SMTPException
from socket import error as SocketError
from urllib import unquote as unquote
from email.mime.text import MIMEText
from ..utils import compat_is_basestring
from .NotifyBase import NotifyBase
from .NotifyBase import NotifyFormat
@ -166,13 +165,13 @@ class NotifyEmail(NotifyBase):
# Keep trying to be clever and make it equal to the to address
self.from_addr = self.to_addr
if not isinstance(self.to_addr, basestring):
if not compat_is_basestring(self.to_addr):
raise TypeError('No valid ~To~ email address specified.')
if not NotifyBase.is_email(self.to_addr):
raise TypeError('Invalid ~To~ email format: %s' % self.to_addr)
if not isinstance(self.from_addr, basestring):
if not compat_is_basestring(self.from_addr):
raise TypeError('No valid ~From~ email address specified.')
match = NotifyBase.is_email(self.from_addr)
@ -294,7 +293,7 @@ class NotifyEmail(NotifyBase):
self.to_addr,
))
except (SocketError, SMTPException), e:
except (SocketError, SMTPException) as e:
self.logger.warning(
'A Connection error occured sending Email '
'notification to %s.' % self.smtp_host)
@ -336,7 +335,7 @@ class NotifyEmail(NotifyBase):
if 'format' in results['qsd'] and len(results['qsd']['format']):
# Extract email format (Text/Html)
try:
format = unquote(results['qsd']['format']).lower()
format = NotifyBase.unquote(results['qsd']['format']).lower()
if len(format) > 0 and format[0] == 't':
results['notify_format'] = NotifyFormat.TEXT
@ -364,7 +363,7 @@ class NotifyEmail(NotifyBase):
if not NotifyBase.is_email(to_addr):
NotifyBase.logger.error(
'%s does not contain a recipient email.' %
unquote(results['url'].lstrip('/')),
NotifyBase.unquote(results['url'].lstrip('/')),
)
return None
@ -384,7 +383,7 @@ class NotifyEmail(NotifyBase):
if not NotifyBase.is_email(from_addr):
NotifyBase.logger.error(
'%s does not contain a from address.' %
unquote(results['url'].lstrip('/')),
NotifyBase.unquote(results['url'].lstrip('/')),
)
return None
@ -394,7 +393,7 @@ class NotifyEmail(NotifyBase):
try:
if 'name' in results['qsd'] and len(results['qsd']['name']):
# Extract from name to associate with from address
results['name'] = unquote(results['qsd']['name'])
results['name'] = NotifyBase.unquote(results['qsd']['name'])
except AttributeError:
pass
@ -402,7 +401,8 @@ class NotifyEmail(NotifyBase):
try:
if 'timeout' in results['qsd'] and len(results['qsd']['timeout']):
# Extract the timeout to associate with smtp server
results['timeout'] = unquote(results['qsd']['timeout'])
results['timeout'] = NotifyBase.unquote(
results['qsd']['timeout'])
except AttributeError:
pass
@ -411,7 +411,7 @@ class NotifyEmail(NotifyBase):
try:
# Extract from password to associate with smtp server
if 'smtp' in results['qsd'] and len(results['qsd']['smtp']):
smtp_host = unquote(results['qsd']['smtp'])
smtp_host = NotifyBase.unquote(results['qsd']['smtp'])
except AttributeError:
pass

4
apprise/plugins/NotifyGrowl/NotifyGrowl.py
View File

@ -17,7 +17,6 @@
# GNU Lesser General Public License for more details.
import re
from urllib import unquote
from .gntp.notifier import GrowlNotifier
from .gntp.errors import NetworkError as GrowlNetworkError
@ -212,7 +211,8 @@ class NotifyGrowl(NotifyBase):
# Allow the user to specify the version of the protocol to use.
try:
version = int(
unquote(results['qsd']['version']).strip().split('.')[0])
NotifyBase.unquote(
results['qsd']['version']).strip().split('.')[0])
except (AttributeError, IndexError, TypeError, ValueError):
NotifyBase.logger.warning(

3
apprise/plugins/NotifyJSON.py
View File

@ -22,6 +22,7 @@ from json import dumps
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..utils import compat_is_basestring
# Image Support (128x128)
JSON_IMAGE_XY = NotifyImageSize.XY_128
@ -53,7 +54,7 @@ class NotifyJSON(NotifyBase):
self.schema = 'http'
self.fullpath = kwargs.get('fullpath')
if not isinstance(self.fullpath, basestring):
if not compat_is_basestring(self.fullpath):
self.fullpath = '/'
return

13
apprise/plugins/NotifyJoin.py
View File

@ -28,19 +28,20 @@
import re
import requests
from urllib import urlencode
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..utils import compat_is_basestring
# Token required as part of the API request
VALIDATE_APIKEY = re.compile(r'[A-Za-z0-9]{32}')
# Extend HTTP Error Messages
JOIN_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
JOIN_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
JOIN_HTTP_ERROR_MAP.update({
401: 'Unauthorized - Invalid Token.',
}.items())
})
# Used to break path apart into list of devices
DEVICE_LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
@ -90,12 +91,12 @@ class NotifyJoin(NotifyBase):
# The token associated with the account
self.apikey = apikey.strip()
if isinstance(devices, basestring):
if compat_is_basestring(devices):
self.devices = filter(bool, DEVICE_LIST_DELIM.split(
devices,
))
elif isinstance(devices, (tuple, list)):
elif isinstance(devices, (set, tuple, list)):
self.devices = devices
else:
@ -158,7 +159,7 @@ class NotifyJoin(NotifyBase):
payload = {}
# Prepare the URL
url = '%s?%s' % (self.notify_url, urlencode(url_args))
url = '%s?%s' % (self.notify_url, NotifyBase.urlencode(url_args))
self.logger.debug('Join POST URL: %s (cert_verify=%r)' % (
url, self.verify_certificate,

6
apprise/plugins/NotifyMatterMost.py
View File

@ -19,7 +19,6 @@
import re
import requests
from json import dumps
from urllib import unquote as unquote
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
@ -181,8 +180,7 @@ class NotifyMatterMost(NotifyBase):
# Apply our settings now
try:
authtoken = filter(
bool, NotifyBase.split_path(results['fullpath']))[0]
authtoken = NotifyBase.split_path(results['fullpath'])[0]
except (AttributeError, IndexError):
# Force some bad values that will get caught
@ -193,7 +191,7 @@ class NotifyMatterMost(NotifyBase):
if 'channel' in results['qsd'] and len(results['qsd']['channel']):
# Allow the user to specify the channel to post to
try:
channel = unquote(results['qsd']['channel']).strip()
channel = NotifyBase.unquote(results['qsd']['channel']).strip()
except (AttributeError, TypeError, ValueError):
NotifyBase.logger.warning(

8
apprise/plugins/NotifyMyAndroid.py
View File

@ -18,18 +18,18 @@
import re
import requests
from urllib import unquote
from .NotifyBase import NotifyBase
from .NotifyBase import NotifyFormat
from .NotifyBase import HTTP_ERROR_MAP
# Extend HTTP Error Messages
NMA_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
NMA_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
NMA_HTTP_ERROR_MAP.update({
400: 'Data is wrong format, invalid length or null.',
401: 'API Key provided is invalid',
402: 'Maximum number of API calls per hour reached.',
}.items())
})
# Used to validate Authorization Token
VALIDATE_APIKEY = re.compile(r'[A-Za-z0-9]{48}')
@ -185,7 +185,7 @@ class NotifyMyAndroid(NotifyBase):
if 'format' in results['qsd'] and len(results['qsd']['format']):
# Extract email format (Text/Html)
try:
format = unquote(results['qsd']['format']).lower()
format = NotifyBase.unquote(results['qsd']['format']).lower()
if len(format) > 0 and format[0] == 't':
results['notify_format'] = NotifyFormat.TEXT

5
apprise/plugins/NotifyProwl.py
View File

@ -47,10 +47,11 @@ PROWL_PRIORITIES = (
)
# Extend HTTP Error Messages
PROWL_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
PROWL_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
HTTP_ERROR_MAP.update({
406: 'IP address has exceeded API limit',
409: 'Request not aproved.',
}.items())
})
class NotifyProwl(NotifyBase):

12
apprise/plugins/NotifyPushBullet.py
View File

@ -19,12 +19,13 @@
import re
import requests
from json import dumps
from urllib import unquote
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from .NotifyBase import IS_EMAIL_RE
from ..utils import compat_is_basestring
# Flag used as a placeholder to sending to all devices
PUSHBULLET_SEND_TO_ALL = 'ALL_DEVICES'
@ -33,9 +34,10 @@ PUSHBULLET_SEND_TO_ALL = 'ALL_DEVICES'
RECIPIENTS_LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
# Extend HTTP Error Messages
PUSHBULLET_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
PUSHBULLET_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
PUSHBULLET_HTTP_ERROR_MAP.update({
401: 'Unauthorized - Invalid Token.',
}.items())
})
class NotifyPushBullet(NotifyBase):
@ -60,7 +62,7 @@ class NotifyPushBullet(NotifyBase):
title_maxlen=250, body_maxlen=32768, **kwargs)
self.accesstoken = accesstoken
if isinstance(recipients, basestring):
if compat_is_basestring(recipients):
self.recipients = filter(bool, RECIPIENTS_LIST_DELIM.split(
recipients,
))
@ -178,7 +180,7 @@ class NotifyPushBullet(NotifyBase):
# Apply our settings now
try:
recipients = unquote(results['fullpath'])
recipients = NotifyBase.unquote(results['fullpath'])
except AttributeError:
recipients = ''

5
apprise/plugins/NotifyPushalot.py
View File

@ -28,10 +28,11 @@ from ..common import NotifyImageSize
PUSHALOT_IMAGE_XY = NotifyImageSize.XY_72
# Extend HTTP Error Messages
PUSHALOT_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
PUSHALOT_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
PUSHALOT_HTTP_ERROR_MAP.update({
406: 'Message throttle limit hit.',
410: 'AuthorizedToken is no longer valid.',
}.items())
})
# Used to validate Authorization Token
VALIDATE_AUTHTOKEN = re.compile(r'[A-Za-z0-9]{32}')

16
apprise/plugins/NotifyPushover.py
View File

@ -18,8 +18,8 @@
import re
import requests
from urllib import unquote
from ..utils import compat_is_basestring
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
@ -57,9 +57,10 @@ PUSHOVER_PRIORITIES = (
DEVICE_LIST_DELIM = re.compile(r'[ \t\r\n,\\/]+')
# Extend HTTP Error Messages
PUSHOVER_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
PUSHOVER_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
PUSHOVER_HTTP_ERROR_MAP.update({
401: 'Unauthorized - Invalid Token.',
}.items())
})
class NotifyPushover(NotifyBase):
@ -95,12 +96,14 @@ class NotifyPushover(NotifyBase):
# The token associated with the account
self.token = token.strip()
if isinstance(devices, basestring):
if compat_is_basestring(devices):
self.devices = filter(bool, DEVICE_LIST_DELIM.split(
devices,
))
elif isinstance(devices, (tuple, list)):
elif isinstance(devices, (set, tuple, list)):
self.devices = devices
else:
self.devices = list()
@ -110,6 +113,7 @@ class NotifyPushover(NotifyBase):
# The Priority of the message
if priority not in PUSHOVER_PRIORITIES:
self.priority = PushoverPriority.NORMAL
else:
self.priority = priority
@ -230,7 +234,7 @@ class NotifyPushover(NotifyBase):
# Apply our settings now
try:
devices = unquote(results['fullpath'])
devices = NotifyBase.unquote(results['fullpath'])
except AttributeError:
devices = ''

13
apprise/plugins/NotifyRocketChat.py
View File

@ -19,19 +19,20 @@
import re
import requests
from json import loads
from urllib import unquote
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..utils import compat_is_basestring
IS_CHANNEL = re.compile(r'^#(?P<name>[A-Za-z0-9]+)$')
IS_ROOM_ID = re.compile(r'^(?P<name>[A-Za-z0-9]+)$')
# Extend HTTP Error Messages
RC_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
RC_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
RC_HTTP_ERROR_MAP.update({
400: 'Channel/RoomId is wrong format, or missing from server.',
401: 'Authentication tokens provided is invalid or missing.',
}.items())
})
# Used to break apart list of potential tags by their delimiter
# into a usable list.
@ -79,12 +80,12 @@ class NotifyRocketChat(NotifyBase):
if recipients is None:
recipients = []
elif isinstance(recipients, basestring):
elif compat_is_basestring(recipients):
recipients = filter(bool, LIST_DELIM.split(
recipients,
))
elif not isinstance(recipients, (tuple, list)):
elif not isinstance(recipients, (set, tuple, list)):
recipients = []
# Validate recipients and drop bad ones:
@ -316,7 +317,7 @@ class NotifyRocketChat(NotifyBase):
# Apply our settings now
try:
results['recipients'] = unquote(results['fullpath'])
results['recipients'] = NotifyBase.unquote(results['fullpath'])
except AttributeError:
return None

10
apprise/plugins/NotifySlack.py
View File

@ -36,6 +36,7 @@ from time import time
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..utils import compat_is_basestring
# Token required as part of the API request
# /AAAAAAAAA/........./........................
@ -53,9 +54,10 @@ VALIDATE_TOKEN_C = re.compile(r'[A-Za-z0-9]{24}')
SLACK_DEFAULT_USER = 'apprise'
# Extend HTTP Error Messages
SLACK_HTTP_ERROR_MAP = dict(HTTP_ERROR_MAP.items() + {
SLACK_HTTP_ERROR_MAP = HTTP_ERROR_MAP.copy()
SLACK_HTTP_ERROR_MAP.update({
401: 'Unauthorized - Invalid Token.',
}.items())
})
# Used to break path apart into list of devices
CHANNEL_LIST_DELIM = re.compile(r'[ \t\r\n,#\\/]+')
@ -124,11 +126,11 @@ class NotifySlack(NotifyBase):
'No user was specified; using %s.' % SLACK_DEFAULT_USER)
self.user = SLACK_DEFAULT_USER
if isinstance(channels, basestring):
if compat_is_basestring(channels):
self.channels = filter(bool, CHANNEL_LIST_DELIM.split(
channels,
))
elif isinstance(channels, (tuple, list)):
elif isinstance(channels, (set, tuple, list)):
self.channels = channels
else:
self.channels = list()

11
apprise/plugins/NotifyTelegram.py
View File

@ -50,6 +50,8 @@ from .NotifyBase import NotifyBase
from .NotifyBase import NotifyFormat
from .NotifyBase import HTTP_ERROR_MAP
from ..utils import compat_is_basestring
# Token required as part of the API request
# allow the word 'bot' infront
VALIDATE_BOT_TOKEN = re.compile(
@ -108,11 +110,12 @@ class NotifyTelegram(NotifyBase):
# Store our API Key
self.bot_token = result.group('key')
if isinstance(chat_ids, basestring):
if compat_is_basestring(chat_ids):
self.chat_ids = filter(bool, CHAT_ID_LIST_DELIM.split(
chat_ids,
))
elif isinstance(chat_ids, (tuple, list)):
elif isinstance(chat_ids, (set, tuple, list)):
self.chat_ids = list(chat_ids)
else:
@ -246,8 +249,8 @@ class NotifyTelegram(NotifyBase):
# payload['parse_mode'] = 'Markdown'
payload['parse_mode'] = 'HTML'
payload['text'] = '<b>%s</b>\r\n%s' % (
self.escape_html(title),
self.escape_html(body),
NotifyBase.escape_html(title),
NotifyBase.escape_html(body),
)
# Create a copy of the chat_ids list

13
apprise/plugins/NotifyToasty.py
View File

@ -18,12 +18,11 @@
import re
import requests
from urllib import quote
from urllib import unquote
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..utils import compat_is_basestring
# Image Support (128x128)
TOASTY_IMAGE_XY = NotifyImageSize.XY_128
@ -52,7 +51,7 @@ class NotifyToasty(NotifyBase):
title_maxlen=250, body_maxlen=32768, image_size=TOASTY_IMAGE_XY,
**kwargs)
if isinstance(devices, basestring):
if compat_is_basestring(devices):
self.devices = filter(bool, DEVICES_LIST_DELIM.split(
devices,
))
@ -86,9 +85,9 @@ class NotifyToasty(NotifyBase):
# prepare JSON Object
payload = {
'sender': quote(self.user),
'title': quote(title),
'text': quote(body),
'sender': NotifyBase.quote(self.user),
'title': NotifyBase.quote(title),
'text': NotifyBase.quote(body),
}
if self.include_image:
@ -163,7 +162,7 @@ class NotifyToasty(NotifyBase):
# Apply our settings now
try:
devices = unquote(results['fullpath'])
devices = NotifyBase.unquote(results['fullpath'])
except AttributeError:
devices = ''

10
apprise/plugins/NotifyXML.py
View File

@ -18,11 +18,11 @@
import re
import requests
from urllib import quote
from .NotifyBase import NotifyBase
from .NotifyBase import HTTP_ERROR_MAP
from ..common import NotifyImageSize
from ..utils import compat_is_basestring
# Image Support (128x128)
XML_IMAGE_XY = NotifyImageSize.XY_128
@ -69,7 +69,7 @@ class NotifyXML(NotifyBase):
self.schema = 'http'
self.fullpath = kwargs.get('fullpath')
if not isinstance(self.fullpath, basestring):
if not compat_is_basestring(self.fullpath):
self.fullpath = '/'
return
@ -86,9 +86,9 @@ class NotifyXML(NotifyBase):
}
re_map = {
'{MESSAGE_TYPE}': quote(notify_type),
'{SUBJECT}': quote(title),
'{MESSAGE}': quote(body),
'{MESSAGE_TYPE}': NotifyBase.quote(notify_type),
'{SUBJECT}': NotifyBase.quote(title),
'{MESSAGE}': NotifyBase.quote(body),
}
# Iterate over above list and store content accordingly

114
apprise/utils.py
View File

@ -20,18 +20,30 @@ import re
from os.path import expanduser
try:
# Python 2.7
from urllib import unquote
from urllib import quote
from urlparse import urlparse
from urlparse import parse_qsl
from urllib import quote
from urllib import unquote
except ImportError:
# Python 3.x
from urllib.parse import unquote
from urllib.parse import quote
from urllib.parse import urlparse
from urllib.parse import parse_qsl
import logging
logger = logging.getLogger(__name__)
# URL Indexing Table for returns via parse_url()
VALID_URL_RE = re.compile(r'^[\s]*([^:\s]+):[/\\]*([^?]+)(\?(.+))?[\s]*$')
VALID_HOST_RE = re.compile(r'^[\s]*([^:/\s]+)')
VALID_QUERY_RE = re.compile(r'^(.*[/\\])([^/\\]*)$')
VALID_URL_RE = re.compile(
r'^[\s]*(?P<schema>[^:\s]+):[/\\]*(?P<path>[^?]+)'
r'(\?(?P<kwargs>.+))?[\s]*$',
)
VALID_HOST_RE = re.compile(r'^[\s]*(?P<path>[^?\s]+)(\?(?P<kwargs>.+))?')
VALID_QUERY_RE = re.compile(r'^(?P<path>.*[/\\])(?P<query>[^/\\]*)$')
# delimiters used to separate values when content is passed in by string.
# This is useful when turning a string into a list
@ -43,7 +55,7 @@ ESCAPED_WIN_PATH_SEPARATOR = re.escape('\\')
ESCAPED_NUX_PATH_SEPARATOR = re.escape('/')
TIDY_WIN_PATH_RE = re.compile(
'(^[%s]{2}|[^%s\s][%s]|[\s][%s]{2}])([%s]+)' % (
r'(^[%s]{2}|[^%s\s][%s]|[\s][%s]{2}])([%s]+)' % (
ESCAPED_WIN_PATH_SEPARATOR,
ESCAPED_WIN_PATH_SEPARATOR,
ESCAPED_WIN_PATH_SEPARATOR,
@ -52,27 +64,41 @@ TIDY_WIN_PATH_RE = re.compile(
),
)
TIDY_WIN_TRIM_RE = re.compile(
'^(.+[^:][^%s])[\s%s]*$' % (
r'^(.+[^:][^%s])[\s%s]*$' % (
ESCAPED_WIN_PATH_SEPARATOR,
ESCAPED_WIN_PATH_SEPARATOR,
),
)
TIDY_NUX_PATH_RE = re.compile(
'([%s])([%s]+)' % (
r'([%s])([%s]+)' % (
ESCAPED_NUX_PATH_SEPARATOR,
ESCAPED_NUX_PATH_SEPARATOR,
),
)
TIDY_NUX_TRIM_RE = re.compile(
'([^%s])[\s%s]+$' % (
r'([^%s])[\s%s]+$' % (
ESCAPED_NUX_PATH_SEPARATOR,
ESCAPED_NUX_PATH_SEPARATOR,
),
)
def compat_is_basestring(content):
"""
Python 3 support for checking if content is unicode and/or
of a string type
"""
try:
# Python v2.x
return isinstance(content, basestring)
except NameError:
# Python v3.x
return isinstance(content, str)
def tidy_path(path):
"""take a filename and or directory and attempts to tidy it up by removing
trailing slashes and correcting any formatting issues.
@ -116,7 +142,7 @@ def parse_url(url, default_schema='http'):
content could not be extracted.
"""
if not isinstance(url, basestring):
if not compat_is_basestring(url):
# Simple error checking
return None
@ -150,32 +176,36 @@ def parse_url(url, default_schema='http'):
match = VALID_URL_RE.search(url)
if match:
# Extract basic results
result['schema'] = match.group(1).lower().strip()
host = match.group(2).strip()
result['schema'] = match.group('schema').lower().strip()
host = match.group('path').strip()
try:
qsdata = match.group(4).strip()
qsdata = match.group('kwargs').strip()
except AttributeError:
# No qsdata
pass
else:
match = VALID_HOST_RE.search(url)
if not match:
return None
result['schema'] = default_schema
host = match.group(1).strip()
if not result['schema']:
result['schema'] = default_schema
if not host:
# Invalid Hostname
return None
host = match.group('path').strip()
try:
qsdata = match.group('kwargs').strip()
except AttributeError:
# No qsdata
pass
# Now do a proper extraction of data
parsed = urlparse('http://%s' % host)
# Parse results
result['host'] = parsed[1].strip()
if not result['host']:
# Nothing more we can do without a hostname
return None
result['fullpath'] = quote(unquote(tidy_path(parsed[2].strip())))
try:
# Handle trailing slashes removed by tidy_path
@ -201,14 +231,13 @@ def parse_url(url, default_schema='http'):
if not result['fullpath']:
# Default
result['fullpath'] = None
else:
# Using full path, extract query from path
match = VALID_QUERY_RE.search(result['fullpath'])
if match:
result['path'] = match.group(1)
result['query'] = match.group(2)
if not result['path']:
result['path'] = None
result['path'] = match.group('path')
result['query'] = match.group('query')
if not result['query']:
result['query'] = None
try:
@ -242,18 +271,22 @@ def parse_url(url, default_schema='http'):
if result['port']:
try:
result['port'] = int(result['port'])
except (ValueError, TypeError):
# Invalid Port Specified
return None
if result['port'] == 0:
result['port'] = None
# Re-assemble cleaned up version of the url
result['url'] = '%s://' % result['schema']
if isinstance(result['user'], basestring):
if compat_is_basestring(result['user']):
result['url'] += result['user']
if isinstance(result['password'], basestring):
if compat_is_basestring(result['password']):
result['url'] += ':%s@' % result['password']
else:
result['url'] += '@'
result['url'] += result['host']
@ -277,7 +310,7 @@ def parse_bool(arg, default=False):
If the content could not be parsed, then the default is returned.
"""
if isinstance(arg, basestring):
if compat_is_basestring(arg):
# no = no - False
# of = short for off - False
# 0 = int for False
@ -330,23 +363,20 @@ def parse_list(*args):
result = []
for arg in args:
if isinstance(arg, basestring):
if compat_is_basestring(arg):
result += re.split(STRING_DELIMITERS, arg)
elif isinstance(arg, (list, tuple)):
for _arg in arg:
if isinstance(arg, basestring):
result += re.split(STRING_DELIMITERS, arg)
# A list inside a list? - use recursion
elif isinstance(_arg, (list, tuple)):
result += parse_list(_arg)
else:
# Convert whatever it is to a string and work with it
result += parse_list(str(_arg))
elif isinstance(arg, (set, list, tuple)):
result += parse_list(*arg)
else:
# Convert whatever it is to a string and work with it
result += parse_list(str(arg))
# apply as well as make the list unique by converting it
# to a set() first. filter() eliminates any empty entries
return filter(bool, list(set(result)))
#
# filter() eliminates any empty entries
#
# Since Python v3 returns a filter (iterator) where-as Python v2 returned
# a list, we need to change it into a list object to remain compatible with
# both distribution types.
return sorted([x for x in filter(bool, list(set(result)))])

5
dev-requirements.txt Normal file
View File

@ -0,0 +1,5 @@
pytest
coverage
pytest-cov
pycodestyle
tox

2
requirements.txt
View File

@ -1,5 +1,3 @@
chardet
markdown
decorator
requests
requests-oauthlib

26
setup.cfg
View File

@ -1,14 +1,22 @@
[egg_info]
tag_build =
tag_date = 0
tag_svn_revision = 0
[bdist_wheel]
universal = 1
[metadata]
# ensure LICENSE is included in wheel metadata
license_file = LICENSE
[pycodestyle]
# We exclude packages we don't maintain
exclude = gntp,tweepy,pushjet
exclude = .eggs,.tox,gntp,tweepy,pushjet
ignore = E722
statistics = True
statistics = true
[coverage:run]
source=apprise
omit=*/gntp/*,*/tweepy/*,*/pushjet/*
[aliases]
test=pytest
[tool:pytest]
addopts = --verbose -ra
python_files = test/test_*.py
filterwarnings =
once::Warning
strict = true

13
setup.py
View File

@ -45,9 +45,11 @@ setup(
author='Chris Caron',
author_email='lead2gold@gmail.com',
packages=find_packages(),
include_package_data=True,
package_data={
'apprise': ['assets'],
'apprise': [
'assets/NotifyXML-1.0.xsd',
'assets/themes/default/*.png',
],
},
scripts=['cli/notify.py', ],
install_requires=open('requirements.txt').readlines(),
@ -58,10 +60,11 @@ setup(
'Natural Language :: English',
'Programming Language :: Python',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3',
'License :: OSI Approved :: GNU General Public License v3 (GPLv3)',
),
entry_points={'console_scripts': console_scripts},
python_requires='>=2.7, <3',
test_suite='nose.collector',
tests_require=['nose', 'coverage', 'pycodestyle'],
python_requires='>=2.7',
setup_requires=['pytest-runner', ],
tests_require=['pytest', 'coverage', 'pytest-cov', 'pycodestyle', 'tox'],
)

91
test/test_api.py
View File

@ -1,9 +1,25 @@
"""API properties.
"""
# -*- coding: utf-8 -*-
#
# Apprise and AppriseAsset Unit Tests
#
# Copyright (C) 2017 Chris Caron <lead2gold@gmail.com>
#
# This file is part of apprise.
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
from __future__ import print_function
from __future__ import unicode_literals
from os import chmod
from os.path import dirname
from apprise import Apprise
from apprise import AppriseAsset
from apprise.Apprise import SCHEMA_MAP
@ -150,7 +166,7 @@ def test_apprise():
assert(a.notify(title="present", body="present") is False)
def test_apprise_asset():
def test_apprise_asset(tmpdir):
"""
API: AppriseAsset() object
@ -175,6 +191,10 @@ def test_apprise_asset():
NotifyImageSize.XY_256,
must_exist=False) == '/dark/info-256x256.png')
# This path doesn't exist so image_raw will fail (since we just
# randompyl picked it for testing)
assert(a.image_raw(NotifyType.INFO, NotifyImageSize.XY_256) is None)
assert(a.image_path(
NotifyType.INFO,
NotifyImageSize.XY_256,
@ -190,3 +210,66 @@ def test_apprise_asset():
must_exist=True) is not None)
assert(a.image_raw(NotifyType.INFO, NotifyImageSize.XY_256) is not None)
# Create a temporary directory
sub = tmpdir.mkdir("great.theme")
# Write a file
sub.join("{0}-{1}.png".format(
NotifyType.INFO,
NotifyImageSize.XY_256,
)).write("the content doesn't matter for testing.")
# Create an asset that will reference our file we just created
a = AppriseAsset(
theme='great.theme',
image_path_mask='%s/{THEME}/{TYPE}-{XY}.png' % dirname(sub.strpath),
)
# We'll be able to read file we just created
assert(a.image_raw(NotifyType.INFO, NotifyImageSize.XY_256) is not None)
# We can retrieve the filename at this point even with must_exist set
# to True
assert(a.image_path(
NotifyType.INFO,
NotifyImageSize.XY_256,
must_exist=True) is not None)
# If we make the file un-readable however, we won't be able to read it
# This test is just showing that we won't throw an exception
chmod(dirname(sub.strpath), 0o000)
assert(a.image_raw(NotifyType.INFO, NotifyImageSize.XY_256) is None)
# Our path doesn't exist anymore using this logic
assert(a.image_path(
NotifyType.INFO,
NotifyImageSize.XY_256,
must_exist=True) is None)
# Return our permission so we don't have any problems with our cleanup
chmod(dirname(sub.strpath), 0o700)
# Our content is retrivable again
assert(a.image_raw(NotifyType.INFO, NotifyImageSize.XY_256) is not None)
# our file path is accessible again too
assert(a.image_path(
NotifyType.INFO,
NotifyImageSize.XY_256,
must_exist=True) is not None)
# We do the same test, but set the permission on the file
chmod(a.image_path(NotifyType.INFO, NotifyImageSize.XY_256), 0o000)
# our path will still exist in this case
assert(a.image_path(
NotifyType.INFO,
NotifyImageSize.XY_256,
must_exist=True) is not None)
# but we will not be able to open it
assert(a.image_raw(NotifyType.INFO, NotifyImageSize.XY_256) is None)
# Restore our permissions
chmod(a.image_path(NotifyType.INFO, NotifyImageSize.XY_256), 0o640)

158
test/test_notifybase.py Normal file
View File

@ -0,0 +1,158 @@
# -*- coding: utf-8 -*-
#
# NotifyBase Unit Tests
#
# Copyright (C) 2017 Chris Caron <lead2gold@gmail.com>
#
# This file is part of apprise.
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
from apprise.plugins.NotifyBase import NotifyBase
from apprise import NotifyType
from apprise import NotifyImageSize
from timeit import default_timer
def test_notify_base():
"""
API: NotifyBase() object
"""
# invalid types throw exceptions
try:
nb = NotifyBase(notify_format='invalid')
# We should never reach here as an exception should be thrown
assert(False)
except TypeError:
assert(True)
try:
nb = NotifyBase(image_size='invalid')
# We should never reach here as an exception should be thrown
assert(False)
except TypeError:
assert(True)
# Bad port information
nb = NotifyBase(port='invalid')
assert nb.port is None
nb = NotifyBase(port=10)
assert nb.port == 10
# Throttle overrides..
nb = NotifyBase(throttle=0)
start_time = default_timer()
nb.throttle()
elapsed = default_timer() - start_time
# Should be a very fast response time since we set it to zero but we'll
# check for less then 500 to be fair as some testing systems may be slower
# then other
assert elapsed < 0.5
start_time = default_timer()
nb.throttle(1.0)
elapsed = default_timer() - start_time
# Should be a very fast response time since we set it to zero but we'll
# check for less then 500 to be fair as some testing systems may be slower
# then other
assert elapsed < 1.5
# our NotifyBase wasn't initialized with an ImageSize so this will fail
assert nb.image_url(notify_type=NotifyType.INFO) is None
assert nb.image_path(notify_type=NotifyType.INFO) is None
assert nb.image_raw(notify_type=NotifyType.INFO) is None
# Create an object with an ImageSize loaded into it
nb = NotifyBase(image_size=NotifyImageSize.XY_256)
# We'll get an object thi time around
assert nb.image_url(notify_type=NotifyType.INFO) is not None
assert nb.image_path(notify_type=NotifyType.INFO) is not None
assert nb.image_raw(notify_type=NotifyType.INFO) is not None
# But we will not get a response with an invalid notification type
assert nb.image_url(notify_type='invalid') is None
assert nb.image_path(notify_type='invalid') is None
assert nb.image_raw(notify_type='invalid') is None
# Static function testing
assert NotifyBase.escape_html("<content>'\t \n</content>") == \
'&lt;content&gt;&apos;&emsp;&nbsp;\n&lt;/content&gt;'
assert NotifyBase.escape_html(
"<content>'\t \n</content>", convert_new_lines=True) == \
'&lt;content&gt;&apos;&emsp;&nbsp;&lt;br/&gt;&lt;/content&gt;'
assert NotifyBase.split_path(
'/path/?name=Dr%20Disrespect', unquote=False) == \
['path', '?name=Dr%20Disrespect']
assert NotifyBase.split_path(
'/path/?name=Dr%20Disrespect', unquote=True) == \
['path', '?name=Dr', 'Disrespect']
assert NotifyBase.is_email('test@gmail.com') is True
assert NotifyBase.is_email('invalid.com') is False
def test_notify_base_urls():
"""
API: NotifyBase() URLs
"""
# Test verify switch whih is used as part of the SSL Verification
# by default all SSL sites are verified unless this flag is set to
# something like 'No', 'False', 'Disabled', etc. Boolean values are
# pretty forgiving.
results = NotifyBase.parse_url('https://localhost:8080/?verify=No')
assert 'verify' in results
assert results['verify'] is False
results = NotifyBase.parse_url('https://localhost:8080/?verify=Yes')
assert 'verify' in results
assert results['verify'] is True
# The default is to verify
results = NotifyBase.parse_url('https://localhost:8080')
assert 'verify' in results
assert results['verify'] is True
# Password Handling
# pass keyword over-rides default password
results = NotifyBase.parse_url('https://user:pass@localhost')
assert 'password' in results
assert results['password'] == "pass"
# pass keyword over-rides default password
results = NotifyBase.parse_url(
'https://user:pass@localhost?pass=newpassword')
assert 'password' in results
assert results['password'] == "newpassword"
# User Handling
# user keyword over-rides default password
results = NotifyBase.parse_url('https://user:pass@localhost')
assert 'user' in results
assert results['user'] == "user"
# user keyword over-rides default password
results = NotifyBase.parse_url(
'https://user:pass@localhost?user=newuser')
assert 'user' in results
assert results['user'] == "newuser"

95
test/test_utils.py
View File

@ -1,10 +1,31 @@
"""API properties.
"""
# -*- coding: utf-8 -*-
#
# Unit Tests for common shared utility functions
#
# Copyright (C) 2017 Chris Caron <lead2gold@gmail.com>
#
# This file is part of apprise.
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
from __future__ import print_function
from __future__ import unicode_literals
try:
# Python 2.7
from urllib import unquote
except ImportError:
# Python 3.x
from urllib.parse import unquote
from apprise import utils
@ -153,6 +174,54 @@ def test_parse_url():
)
assert(result is None)
# just hostnames
result = utils.parse_url(
'nuxref.com'
)
assert(result['schema'] == 'http')
assert(result['host'] == 'nuxref.com')
assert(result['port'] is None)
assert(result['user'] is None)
assert(result['password'] is None)
assert(result['fullpath'] is None)
assert(result['path'] is None)
assert(result['query'] is None)
assert(result['url'] == 'http://nuxref.com')
assert(result['qsd'] == {})
# just host and path
result = utils.parse_url(
'invalid/host'
)
assert(result['schema'] == 'http')
assert(result['host'] == 'invalid')
assert(result['port'] is None)
assert(result['user'] is None)
assert(result['password'] is None)
assert(result['fullpath'] == '/host')
assert(result['path'] == '/')
assert(result['query'] == 'host')
assert(result['url'] == 'http://invalid/host')
assert(result['qsd'] == {})
# just all out invalid
assert(utils.parse_url('?') is None)
assert(utils.parse_url('/') is None)
# A default port of zero is still considered valid, but
# is removed in the response.
result = utils.parse_url('http://nuxref.com:0')
assert(result['schema'] == 'http')
assert(result['host'] == 'nuxref.com')
assert(result['port'] is None)
assert(result['user'] is None)
assert(result['password'] is None)
assert(result['fullpath'] is None)
assert(result['path'] is None)
assert(result['query'] is None)
assert(result['url'] == 'http://nuxref.com')
assert(result['qsd'] == {})
def test_parse_bool():
"utils: parse_bool() testing """
@ -202,21 +271,25 @@ def test_parse_list():
results = utils.parse_list(
'.mkv,.avi,.divx,.xvid,.mov,.wmv,.mp4,.mpg,.mpeg,.vob,.iso')
assert(results == [
assert(results == sorted([
'.divx', '.iso', '.mkv', '.mov', '.mpg', '.avi', '.mpeg', '.vob',
'.xvid', '.wmv', '.mp4',
])
]))
class StrangeObject(object):
def __str__(self):
return '.avi'
# Now 2 lists with lots of duplicates and other delimiters
results = utils.parse_list(
'.mkv,.avi,.divx,.xvid,.mov,.wmv,.mp4,.mpg .mpeg,.vob,,; ;',
'.mkv,.avi,.divx,.xvid,.mov .wmv,.mp4;.mpg,.mpeg,'
'.vob,.iso')
('.mkv,.avi,.divx,.xvid,.mov ', ' .wmv,.mp4;.mpg,.mpeg,'),
'.vob,.iso', ['.vob', ['.vob', '.mkv', StrangeObject(), ], ],
StrangeObject())
assert(results == [
assert(results == sorted([
'.divx', '.iso', '.mkv', '.mov', '.mpg', '.avi', '.mpeg', '.vob',
'.xvid', '.wmv', '.mp4',
])
]))
# Now a list with extras we want to add as strings
# empty entries are removed
@ -224,7 +297,7 @@ def test_parse_list():
'.divx', '.iso', '.mkv', '.mov', '', ' ', '.avi', '.mpeg', '.vob',
'.xvid', '.mp4'], '.mov,.wmv,.mp4,.mpg')
assert(results == [
assert(results == sorted([
'.divx', '.wmv', '.iso', '.mkv', '.mov', '.mpg', '.avi', '.vob',
'.xvid', '.mpeg', '.mp4',
])
]))

52
tox.ini Normal file
View File

@ -0,0 +1,52 @@
[tox]
envlist = py27,py34,py35,py36,pypy,pypy3,coverage-report
[testenv]
# Prevent random setuptools/pip breakages like
# https://github.com/pypa/setuptools/issues/1042 from breaking our builds.
setenv =
VIRTUALENV_NO_DOWNLOAD=1
deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = python -m pytest {posargs}
[testenv:py27]
deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:py34]
deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:py35]
deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:py36]
deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:pypy]
deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:coverage-report]
deps = coverage
skip_install = true
commands=
coverage combine
coverage report