mirror of
https://github.com/caronc/apprise.git
synced 2024-11-08 09:14:53 +01:00
Config file support added for http & file (yaml); refs #55
This commit is contained in:
Chris Caron
parent
0ab86c2115
commit
1f9daeaa2b
@ -23,9 +23,11 @@
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
# THE SOFTWARE.
|
||||
|
||||
import os
|
||||
import re
|
||||
import six
|
||||
import logging
|
||||
import yaml
|
||||
|
||||
from .. import plugins
|
||||
from ..AppriseAsset import AppriseAsset
|
||||
@ -184,6 +186,18 @@ class ConfigBase(URLBase):
|
||||
|
||||
Optionally associate an asset with the notification.
|
||||
|
||||
The file syntax is:
|
||||
|
||||
#
|
||||
# pound/hashtag allow for line comments
|
||||
#
|
||||
# One or more tags can be idenified using comma's (,) to separate
|
||||
# them.
|
||||
<Tag(s)>=<URL>
|
||||
|
||||
# Or you can use this format (no tags associated)
|
||||
<URL>
|
||||
|
||||
"""
|
||||
# For logging, track the line number
|
||||
line = 0
|
||||
@ -196,8 +210,14 @@ class ConfigBase(URLBase):
|
||||
r'(\s*(?P<tags>[^=]+)=|=)?\s*'
|
||||
r'(?P<url>[a-z0-9]{2,9}://.*))?$', re.I)
|
||||
|
||||
# split our content up to read line by line
|
||||
content = re.split(r'\r*\n', content)
|
||||
try:
|
||||
# split our content up to read line by line
|
||||
content = re.split(r'\r*\n', content)
|
||||
|
||||
except TypeError:
|
||||
# content was not expected string type
|
||||
logger.error('Invalid apprise text data specified')
|
||||
return list()
|
||||
|
||||
for entry in content:
|
||||
# Increment our line count
|
||||
@ -244,7 +264,7 @@ class ConfigBase(URLBase):
|
||||
# containing all of the information parsed from our URL
|
||||
results = plugins.SCHEMA_MAP[schema].parse_url(_url)
|
||||
|
||||
if not results:
|
||||
if results is None:
|
||||
# Failed to parse the server URL
|
||||
logger.warning(
|
||||
'Unparseable URL {} on line {}.'.format(url, line))
|
||||
@ -288,7 +308,234 @@ class ConfigBase(URLBase):
|
||||
"""
|
||||
response = list()
|
||||
|
||||
# TODO
|
||||
try:
|
||||
# Load our data
|
||||
result = yaml.load(content)
|
||||
|
||||
except (AttributeError, yaml.error.MarkedYAMLError) as e:
|
||||
# Invalid content
|
||||
logger.error('Invalid apprise yaml data specified.')
|
||||
logger.debug('YAML Exception:{}{}'.format(os.linesep, e))
|
||||
return list()
|
||||
|
||||
if not isinstance(result, dict):
|
||||
# Invalid content
|
||||
logger.error('Invalid apprise yaml structure specified')
|
||||
return list()
|
||||
|
||||
# YAML Version
|
||||
version = result.get('version', 1)
|
||||
if version != 1:
|
||||
# Invalid syntax
|
||||
logger.error(
|
||||
'Invalid apprise yaml version specified {}.'.format(version))
|
||||
return list()
|
||||
|
||||
#
|
||||
# global asset object
|
||||
#
|
||||
asset = asset if isinstance(asset, AppriseAsset) else AppriseAsset()
|
||||
tokens = result.get('asset', None)
|
||||
if tokens and isinstance(tokens, dict):
|
||||
for k, v in tokens.items():
|
||||
|
||||
if k.startswith('_') or k.endswith('_'):
|
||||
# Entries are considered reserved if they start or end
|
||||
# with an underscore
|
||||
logger.warning('Ignored asset key "{}".'.format(k))
|
||||
continue
|
||||
|
||||
if not (hasattr(asset, k) and
|
||||
isinstance(getattr(asset, k), six.string_types)):
|
||||
# We can't set a function or non-string set value
|
||||
logger.warning('Invalid asset key "{}".'.format(k))
|
||||
continue
|
||||
|
||||
if v is None:
|
||||
# Convert to an empty string
|
||||
v = ''
|
||||
|
||||
if not isinstance(v, six.string_types):
|
||||
# we must set strings with a string
|
||||
logger.warning('Invalid asset value to "{}".'.format(k))
|
||||
continue
|
||||
|
||||
# Set our asset object with the new value
|
||||
setattr(asset, k, v.strip())
|
||||
|
||||
#
|
||||
# global tag root directive
|
||||
#
|
||||
global_tags = set()
|
||||
|
||||
tags = result.get('tag', None)
|
||||
if tags and isinstance(tags, (list, tuple, six.string_types)):
|
||||
# Store any preset tags
|
||||
global_tags = set(parse_list(tags))
|
||||
|
||||
#
|
||||
# urls root directive
|
||||
#
|
||||
urls = result.get('urls', None)
|
||||
if not isinstance(urls, (list, tuple)):
|
||||
# Unsupported
|
||||
logger.error('Missing "urls" directive in apprise yaml.')
|
||||
return list()
|
||||
|
||||
# Iterate over each URL
|
||||
for no, url in enumerate(urls):
|
||||
|
||||
# Our results object is what we use to instantiate our object if
|
||||
# we can. Reset it to None on each iteration
|
||||
results = list()
|
||||
|
||||
if isinstance(url, six.string_types):
|
||||
# We're just a simple URL string
|
||||
|
||||
# swap hash (#) tag values with their html version
|
||||
_url = url.replace('/#', '/%23')
|
||||
|
||||
# Attempt to acquire the schema at the very least to allow our
|
||||
# plugins to determine if they can make a better
|
||||
# interpretation of a URL geared for them
|
||||
schema = GET_SCHEMA_RE.match(_url)
|
||||
if schema is None:
|
||||
logger.warning(
|
||||
'Unsupported schema in urls entry #{}'.format(no))
|
||||
continue
|
||||
|
||||
# Ensure our schema is always in lower case
|
||||
schema = schema.group('schema').lower()
|
||||
|
||||
# Some basic validation
|
||||
if schema not in plugins.SCHEMA_MAP:
|
||||
logger.warning(
|
||||
'Unsupported schema {} in urls entry #{}'.format(
|
||||
schema, no))
|
||||
continue
|
||||
|
||||
# Parse our url details of the server object as dictionary
|
||||
# containing all of the information parsed from our URL
|
||||
_results = plugins.SCHEMA_MAP[schema].parse_url(_url)
|
||||
if _results is None:
|
||||
logger.warning(
|
||||
'Unparseable {} based url; entry #{}'.format(
|
||||
schema, no))
|
||||
continue
|
||||
|
||||
# add our results to our global set
|
||||
results.append(_results)
|
||||
|
||||
elif isinstance(url, dict):
|
||||
# We are a url string with additional unescaped options
|
||||
if six.PY2:
|
||||
_url, tokens = next(url.iteritems())
|
||||
else: # six.PY3
|
||||
_url, tokens = next(iter(url.items()))
|
||||
|
||||
# swap hash (#) tag values with their html version
|
||||
_url = _url.replace('/#', '/%23')
|
||||
|
||||
# Get our schema
|
||||
schema = GET_SCHEMA_RE.match(_url)
|
||||
if schema is None:
|
||||
logger.warning(
|
||||
'Unsupported schema in urls entry #{}'.format(no))
|
||||
continue
|
||||
|
||||
# Ensure our schema is always in lower case
|
||||
schema = schema.group('schema').lower()
|
||||
|
||||
# Some basic validation
|
||||
if schema not in plugins.SCHEMA_MAP:
|
||||
logger.warning(
|
||||
'Unsupported schema {} in urls entry #{}'.format(
|
||||
schema, no))
|
||||
continue
|
||||
|
||||
# Parse our url details of the server object as dictionary
|
||||
# containing all of the information parsed from our URL
|
||||
_results = plugins.SCHEMA_MAP[schema].parse_url(_url)
|
||||
if _results is None:
|
||||
# Setup dictionary
|
||||
_results = {
|
||||
# Minimum requirements
|
||||
'schema': schema,
|
||||
}
|
||||
|
||||
if tokens is not None:
|
||||
# populate and/or override any results populated by
|
||||
# parse_url()
|
||||
for entries in tokens:
|
||||
# Copy ourselves a template of our parsed URL as a base
|
||||
# to work with
|
||||
r = _results.copy()
|
||||
|
||||
# We are a url string with additional unescaped options
|
||||
if isinstance(entries, dict):
|
||||
if six.PY2:
|
||||
_url, tokens = next(url.iteritems())
|
||||
else: # six.PY3
|
||||
_url, tokens = next(iter(url.items()))
|
||||
|
||||
# Tags you just can't over-ride
|
||||
if 'schema' in entries:
|
||||
del entries['schema']
|
||||
|
||||
# Extend our dictionary with our new entries
|
||||
r.update(entries)
|
||||
|
||||
# add our results to our global set
|
||||
results.append(r)
|
||||
|
||||
else:
|
||||
# add our results to our global set
|
||||
results.append(_results)
|
||||
|
||||
else:
|
||||
# Unsupported
|
||||
logger.warning(
|
||||
'Unsupported apprise yaml entry #{}'.format(no))
|
||||
continue
|
||||
|
||||
# Track our entries
|
||||
entry = 0
|
||||
|
||||
while len(results):
|
||||
# Increment our entry count
|
||||
entry += 1
|
||||
|
||||
# Grab our first item
|
||||
_results = results.pop(0)
|
||||
|
||||
# tag is a special keyword that is managed by apprise object.
|
||||
# The below ensures our tags are set correctly
|
||||
if 'tag' in _results:
|
||||
# Tidy our list up
|
||||
_results['tag'] = \
|
||||
set(parse_list(_results['tag'])) | global_tags
|
||||
|
||||
else:
|
||||
# Just use the global settings
|
||||
_results['tag'] = global_tags
|
||||
|
||||
# Prepare our Asset Object
|
||||
_results['asset'] = asset
|
||||
|
||||
try:
|
||||
# Attempt to create an instance of our plugin using the
|
||||
# parsed URL information
|
||||
plugin = plugins.SCHEMA_MAP[_results['schema']](**_results)
|
||||
|
||||
except Exception:
|
||||
# the arguments are invalid or can not be used.
|
||||
logger.warning(
|
||||
'Could not load apprise yaml entry #{}, item #{}'
|
||||
.format(no, entry))
|
||||
continue
|
||||
|
||||
# if we reach here, we successfully loaded our data
|
||||
response.append(plugin)
|
||||
|
||||
return response
|
||||
|
||||
|
||||
@ -36,6 +36,11 @@ from ..common import ConfigFormat
|
||||
# application/x-yaml
|
||||
MIME_IS_YAML = re.compile('(text|application)/(x-)?yaml', re.I)
|
||||
|
||||
# Support TEXT formats
|
||||
# text/plain
|
||||
# text/html
|
||||
MIME_IS_TEXT = re.compile('text/(plain|html)', re.I)
|
||||
|
||||
|
||||
class ConfigHTTP(ConfigBase):
|
||||
"""
|
||||
@ -224,12 +229,21 @@ class ConfigHTTP(ConfigBase):
|
||||
|
||||
# Detect config format based on mime if the format isn't
|
||||
# already enforced
|
||||
if self.config_format is None \
|
||||
and MIME_IS_YAML.match(r.headers.get(
|
||||
'Content-Type', 'text/plain')) is not None:
|
||||
content_type = r.headers.get(
|
||||
'Content-Type', 'application/octet-stream')
|
||||
if self.config_format is None and content_type:
|
||||
if MIME_IS_YAML.match(content_type) is not None:
|
||||
|
||||
# YAML data detected based on header content
|
||||
self.default_config_format = ConfigFormat.YAML
|
||||
# YAML data detected based on header content
|
||||
self.default_config_format = ConfigFormat.YAML
|
||||
|
||||
elif MIME_IS_TEXT.match(content_type) is not None:
|
||||
|
||||
# TEXT data detected based on header content
|
||||
self.default_config_format = ConfigFormat.TEXT
|
||||
|
||||
# else do nothing; fall back to whatever default is
|
||||
# already set.
|
||||
|
||||
except requests.RequestException as e:
|
||||
self.logger.warning(
|
||||
|
||||
@ -136,7 +136,11 @@ def is_email(address):
|
||||
and False if it isn't.
|
||||
"""
|
||||
|
||||
return GET_EMAIL_RE.match(address) is not None
|
||||
try:
|
||||
return GET_EMAIL_RE.match(address) is not None
|
||||
except TypeError:
|
||||
# invalid syntax
|
||||
return False
|
||||
|
||||
|
||||
def tidy_path(path):
|
||||
|
||||
@ -6,3 +6,4 @@ urllib3
|
||||
six
|
||||
click >= 5.0
|
||||
markdown
|
||||
PyYAML
|
||||
|
||||
@ -23,6 +23,7 @@
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
# THE SOFTWARE.
|
||||
|
||||
import six
|
||||
from apprise.AppriseAsset import AppriseAsset
|
||||
from apprise.config.ConfigBase import ConfigBase
|
||||
|
||||
@ -107,6 +108,12 @@ def test_config_base_config_parse_text():
|
||||
|
||||
"""
|
||||
|
||||
# Garbage Handling
|
||||
assert isinstance(ConfigBase.config_parse_text(object()), list)
|
||||
assert isinstance(ConfigBase.config_parse_text(None), list)
|
||||
assert isinstance(ConfigBase.config_parse_text(''), list)
|
||||
|
||||
# Valid Configuration
|
||||
result = ConfigBase.config_parse_text("""
|
||||
# A comment line over top of a URL
|
||||
mailto://userb:pass@gmail.com
|
||||
@ -166,3 +173,421 @@ def test_config_base_config_parse_text():
|
||||
# We expect to parse 0 entries from the above
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
|
||||
def test_config_base_config_parse_yaml():
|
||||
"""
|
||||
API: ConfigBase.config_parse_yaml object
|
||||
|
||||
"""
|
||||
|
||||
# general reference used below
|
||||
asset = AppriseAsset()
|
||||
|
||||
# Garbage Handling
|
||||
assert isinstance(ConfigBase.config_parse_yaml(object()), list)
|
||||
assert isinstance(ConfigBase.config_parse_yaml(None), list)
|
||||
assert isinstance(ConfigBase.config_parse_yaml(''), list)
|
||||
|
||||
# Invalid Version
|
||||
result = ConfigBase.config_parse_yaml("version: 2a", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid Syntax (throws a ScannerError)
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
urls
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Missing url token
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# No urls defined
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
urls:
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid url defined
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
# Invalid URL definition; yet the answer to life at the same time
|
||||
urls: 43
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid url/schema
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
urls:
|
||||
- invalid://
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid url/schema
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
urls:
|
||||
- invalid://:
|
||||
- a: b
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid url/schema
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
urls:
|
||||
- just some free text that isn't valid:
|
||||
- a garbage entry to go with it
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid url/schema
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
urls:
|
||||
- not even a proper url
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid url/schema
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# no lists... just no
|
||||
urls: [milk, pumpkin pie, eggs, juice]
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Invalid url/schema
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
urls:
|
||||
# a very invalid sns entry
|
||||
- sns://T1JJ3T3L2/
|
||||
- sns://:@/:
|
||||
- invalid: test
|
||||
- sns://T1JJ3T3L2/:
|
||||
- invalid: test
|
||||
|
||||
# some strangness
|
||||
-
|
||||
-
|
||||
- test
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# Invalid data gets us an empty result set
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# Valid Configuration
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed
|
||||
version: 1
|
||||
|
||||
#
|
||||
# Define your notification urls:
|
||||
#
|
||||
urls:
|
||||
- pbul://o.gn5kj6nfhv736I7jC3cj3QLRiyhgl98b
|
||||
- mailto://test:password@gmail.com
|
||||
""", asset=asset)
|
||||
|
||||
# We expect to parse 2 entries from the above
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 2
|
||||
assert len(result[0].tags) == 0
|
||||
|
||||
# Valid Configuration
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
urls:
|
||||
- json://localhost:
|
||||
- tag: my-custom-tag, my-other-tag
|
||||
|
||||
# How to stack multiple entries:
|
||||
- mailto://:
|
||||
- user: jeff
|
||||
pass: 123abc
|
||||
from: jeff@yahoo.ca
|
||||
|
||||
- user: jack
|
||||
pass: pass123
|
||||
from: jack@hotmail.com
|
||||
|
||||
# This is an illegal entry; the schema can not be changed
|
||||
schema: json
|
||||
|
||||
# accidently left a colon at the end of the url; no problem
|
||||
# we'll accept it
|
||||
- mailto://oscar:pass@gmail.com:
|
||||
|
||||
# A telegram entry (returns a None in parse_url())
|
||||
- tgram://invalid
|
||||
|
||||
""", asset=asset)
|
||||
|
||||
# We expect to parse 4 entries from the above because the tgram:// entry
|
||||
# would have failed to be loaded
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 4
|
||||
assert len(result[0].tags) == 2
|
||||
|
||||
# Global Tags
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# Global Tags stacked as a list
|
||||
tag:
|
||||
- admin
|
||||
- devops
|
||||
|
||||
urls:
|
||||
- json://localhost
|
||||
- dbus://
|
||||
""", asset=asset)
|
||||
|
||||
# We expect to parse 3 entries from the above
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 2
|
||||
|
||||
# all entries will have our global tags defined in them
|
||||
for entry in result:
|
||||
assert 'admin' in entry.tags
|
||||
assert 'devops' in entry.tags
|
||||
|
||||
# Global Tags
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# Global Tags
|
||||
tag: admin, devops
|
||||
|
||||
urls:
|
||||
# The following tags will get added to the global set
|
||||
- json://localhost:
|
||||
- tag: string-tag, my-other-tag, text
|
||||
|
||||
# Tags can be presented in this list format too:
|
||||
- dbus://:
|
||||
- tag:
|
||||
- list-tag
|
||||
- dbus
|
||||
""", asset=asset)
|
||||
|
||||
# all entries will have our global tags defined in them
|
||||
for entry in result:
|
||||
assert 'admin' in entry.tags
|
||||
assert 'devops' in entry.tags
|
||||
|
||||
# We expect to parse 3 entries from the above
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 2
|
||||
|
||||
# json:// has 2 globals + 3 defined
|
||||
assert len(result[0].tags) == 5
|
||||
assert 'text' in result[0].tags
|
||||
|
||||
# json:// has 2 globals + 2 defined
|
||||
assert len(result[1].tags) == 4
|
||||
assert 'list-tag' in result[1].tags
|
||||
|
||||
# An invalid set of entries
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
urls:
|
||||
# The following tags will get added to the global set
|
||||
- json://localhost:
|
||||
-
|
||||
-
|
||||
- entry
|
||||
""", asset=asset)
|
||||
|
||||
# We expect to parse 3 entries from the above
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 0
|
||||
|
||||
# An asset we'll manipulate
|
||||
asset = AppriseAsset()
|
||||
|
||||
# Global Tags
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# Test the creation of our apprise asset object
|
||||
asset:
|
||||
app_id: AppriseTest
|
||||
app_desc: Apprise Test Notifications
|
||||
app_url: http://nuxref.com
|
||||
|
||||
# Support setting empty values
|
||||
image_url_mask:
|
||||
image_url_logo:
|
||||
|
||||
image_path_mask: tmp/path
|
||||
|
||||
# invalid entry
|
||||
theme:
|
||||
-
|
||||
-
|
||||
- entry
|
||||
|
||||
# Now for some invalid entries
|
||||
invalid: entry
|
||||
__init__: can't be over-ridden
|
||||
nolists:
|
||||
- we don't support these entries
|
||||
- in the apprise object
|
||||
|
||||
urls:
|
||||
- json://localhost:
|
||||
""", asset=asset)
|
||||
|
||||
# We expect to parse 3 entries from the above
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert asset.app_id == "AppriseTest"
|
||||
assert asset.app_desc == "Apprise Test Notifications"
|
||||
assert asset.app_url == "http://nuxref.com"
|
||||
|
||||
# the theme was not updated and remains the same as it was
|
||||
assert asset.theme == AppriseAsset().theme
|
||||
|
||||
# Empty string assignment
|
||||
assert isinstance(asset.image_url_mask, six.string_types) is True
|
||||
assert asset.image_url_mask == ""
|
||||
assert isinstance(asset.image_url_logo, six.string_types) is True
|
||||
assert asset.image_url_logo == ""
|
||||
|
||||
# For on-lookers looking through this file; here is a perfectly formatted
|
||||
# YAML configuration file for your reference so you can see it without
|
||||
# all of the errors like the ones identified above
|
||||
result = ConfigBase.config_parse_yaml("""
|
||||
# if no version is specified then version 1 is presumed. Thus this is a
|
||||
# completely optional field. It's a good idea to just add this line because it
|
||||
# will help with future ambiguity (if it ever occurs).
|
||||
version: 1
|
||||
|
||||
# Define an Asset object if you wish (Optional)
|
||||
asset:
|
||||
app_id: AppriseTest
|
||||
app_desc: Apprise Test Notifications
|
||||
app_url: http://nuxref.com
|
||||
|
||||
# Optionally define some global tags to associate with ALL of your
|
||||
# urls below.
|
||||
tag: admin, devops
|
||||
|
||||
# Define your URLs (Mandatory!)
|
||||
urls:
|
||||
# Either on-line each entry like this:
|
||||
- json://localhost
|
||||
|
||||
# Or add a colon to the end of the URL where you can optionally provide
|
||||
# over-ride entries. One of the most likely entry to be used here
|
||||
# is the tag entry. This gets extended to the global tag (if defined)
|
||||
# above
|
||||
- xml://localhost:
|
||||
- tag: customer
|
||||
|
||||
# The more elements you specify under a URL the more times the URL will
|
||||
# get replicated and used. Hence this entry actually could be considered
|
||||
# 2 URLs being called with just the destination email address changed:
|
||||
- mailto://george:password@gmail.com:
|
||||
- to: jason@hotmail.com
|
||||
- to: fred@live.com
|
||||
|
||||
# Again... to re-iterate, the above mailto:// would actually fire two (2)
|
||||
# separate emails each with a different destination address specified.
|
||||
# Be careful when defining your arguments and differentiating between
|
||||
# when to use the dash (-) and when not to. Each time you do, you will
|
||||
# cause another instance to be created.
|
||||
|
||||
# Defining more then 1 element to a muti-set is easy, it looks like this:
|
||||
- mailto://jackson:abc123@hotmail.com:
|
||||
- to: jeff@gmail.com
|
||||
tag: jeff, customer
|
||||
|
||||
- to: chris@yahoo.com
|
||||
tag: chris, customer
|
||||
""", asset=asset)
|
||||
|
||||
# okay, here is how we get our total based on the above (read top-down)
|
||||
# +1 json:// entry
|
||||
# +1 xml:// entry
|
||||
# +2 mailto:// entry to jason@hotmail.com and fred@live.com
|
||||
# +2 mailto:// entry to jeff@gmail.com and chris@yahoo.com
|
||||
# = 6
|
||||
assert len(result) == 6
|
||||
|
||||
# all six entries will have our global tags defined in them
|
||||
for entry in result:
|
||||
assert 'admin' in entry.tags
|
||||
assert 'devops' in entry.tags
|
||||
|
||||
# Entries can be directly accessed as they were added
|
||||
|
||||
# our json:// had no additional tags added; so just the global ones
|
||||
# So just 2; admin and devops (these were already validated above in the
|
||||
# for loop
|
||||
assert len(result[0].tags) == 2
|
||||
|
||||
# our xml:// object has 1 tag added (customer)
|
||||
assert len(result[1].tags) == 3
|
||||
assert 'customer' in result[1].tags
|
||||
|
||||
# You get the idea, here is just a direct mapping to the remaining entries
|
||||
# in the same order they appear above
|
||||
assert len(result[2].tags) == 2
|
||||
assert len(result[3].tags) == 2
|
||||
|
||||
assert len(result[4].tags) == 4
|
||||
assert 'customer' in result[4].tags
|
||||
assert 'jeff' in result[4].tags
|
||||
|
||||
assert len(result[5].tags) == 4
|
||||
assert 'customer' in result[5].tags
|
||||
assert 'chris' in result[5].tags
|
||||
|
||||
@ -94,18 +94,27 @@ def test_config_http(mock_post, mock_get):
|
||||
assert isinstance(ch.url(), six.string_types) is True
|
||||
assert isinstance(ch.read(), six.string_types) is True
|
||||
|
||||
# one entry added
|
||||
assert len(ch) == 1
|
||||
|
||||
results = ConfigHTTP.parse_url('http://localhost:8080/path/')
|
||||
assert isinstance(results, dict)
|
||||
ch = ConfigHTTP(**results)
|
||||
assert isinstance(ch.url(), six.string_types) is True
|
||||
assert isinstance(ch.read(), six.string_types) is True
|
||||
|
||||
# one entry added
|
||||
assert len(ch) == 1
|
||||
|
||||
results = ConfigHTTP.parse_url('http://user@localhost?format=text')
|
||||
assert isinstance(results, dict)
|
||||
ch = ConfigHTTP(**results)
|
||||
assert isinstance(ch.url(), six.string_types) is True
|
||||
assert isinstance(ch.read(), six.string_types) is True
|
||||
|
||||
# one entry added
|
||||
assert len(ch) == 1
|
||||
|
||||
results = ConfigHTTP.parse_url('https://localhost')
|
||||
assert isinstance(results, dict)
|
||||
ch = ConfigHTTP(**results)
|
||||
@ -153,10 +162,41 @@ def test_config_http(mock_post, mock_get):
|
||||
|
||||
for st in yaml_supported_types:
|
||||
dummy_request.headers['Content-Type'] = st
|
||||
ch.default_config_format = ConfigFormat.TEXT
|
||||
ch.default_config_format = None
|
||||
assert isinstance(ch.read(), six.string_types) is True
|
||||
# Set to YAML
|
||||
assert ch.default_config_format == ConfigFormat.YAML
|
||||
|
||||
# Test TEXT detection
|
||||
text_supported_types = ('text/plain', 'text/html')
|
||||
|
||||
for st in text_supported_types:
|
||||
dummy_request.headers['Content-Type'] = st
|
||||
ch.default_config_format = None
|
||||
assert isinstance(ch.read(), six.string_types) is True
|
||||
# Set to TEXT
|
||||
assert ch.default_config_format == ConfigFormat.TEXT
|
||||
|
||||
# The type is never adjusted to mime types we don't understand
|
||||
ukwn_supported_types = ('text/css', 'application/zip')
|
||||
|
||||
for st in ukwn_supported_types:
|
||||
dummy_request.headers['Content-Type'] = st
|
||||
ch.default_config_format = None
|
||||
assert isinstance(ch.read(), six.string_types) is True
|
||||
# Remains unchanged
|
||||
assert ch.default_config_format is None
|
||||
|
||||
# When the entry is missing; we handle this too
|
||||
del dummy_request.headers['Content-Type']
|
||||
ch.default_config_format = None
|
||||
assert isinstance(ch.read(), six.string_types) is True
|
||||
# Remains unchanged
|
||||
assert ch.default_config_format is None
|
||||
|
||||
# Restore our content type object for lower tests
|
||||
dummy_request.headers['Content-Type'] = 'text/plain'
|
||||
|
||||
ch.max_buffer_size = len(dummy_request.content) - 1
|
||||
assert ch.read() is None
|
||||
|
||||
|
||||
@ -398,9 +398,12 @@ def test_is_email():
|
||||
"""
|
||||
# Valid Emails
|
||||
assert utils.is_email('test@gmail.com') is True
|
||||
assert utils.is_email('tag+test@gmail.com') is True
|
||||
|
||||
# Invalid Emails
|
||||
assert utils.is_email('invalid.com') is False
|
||||
assert utils.is_email(object()) is False
|
||||
assert utils.is_email(None) is False
|
||||
|
||||
|
||||
def test_parse_list():
|
||||
|
||||
Loading…
Reference in New Issue
Block a user