extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-10-04 17:22:18 +02:00
Code Issues Packages Projects Releases Wiki Activity
master
apprise/test/test_config_base.py

1488 lines
41 KiB
Python
Raw Permalink Blame History

# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import pytest
from apprise import AppriseAsset
from apprise.config import ConfigBase
from apprise import Apprise
from apprise import ConfigFormat
from inspect import cleandoc
import yaml
# Disable logging for a cleaner testing output
import logging
from apprise.plugins.email import NotifyEmail
logging.disable(logging.CRITICAL)
def test_config_base():
"""
API: ConfigBase() object
"""
# invalid types throw exceptions
with pytest.raises(TypeError):
ConfigBase(**{'format': 'invalid'})
# Config format types are not the same as ConfigBase ones
with pytest.raises(TypeError):
ConfigBase(**{'format': 'markdown'})
cb = ConfigBase(**{'format': 'yaml'})
assert isinstance(cb, ConfigBase)
cb = ConfigBase(**{'format': 'text'})
assert isinstance(cb, ConfigBase)
# Set encoding
cb = ConfigBase(encoding='utf-8', format='text')
assert isinstance(cb, ConfigBase)
# read is not supported in the base object; only the children
assert cb.read() is None
# There are no servers loaded on a freshly created object
assert len(cb.servers()) == 0
# Unsupported URLs are not parsed
assert ConfigBase.parse_url(url='invalid://') is None
# Valid URL & Valid Format
results = ConfigBase.parse_url(
url='file://relative/path?format=yaml&encoding=latin-1')
assert isinstance(results, dict)
# These are moved into the root
assert results.get('format') == 'yaml'
assert results.get('encoding') == 'latin-1'
# But they also exist in the qsd location
assert isinstance(results.get('qsd'), dict)
assert results['qsd'].get('encoding') == 'latin-1'
assert results['qsd'].get('format') == 'yaml'
# Valid URL & Invalid Format
results = ConfigBase.parse_url(
url='file://relative/path?format=invalid&encoding=latin-1')
assert isinstance(results, dict)
# Only encoding is moved into the root
assert 'format' not in results
assert results.get('encoding') == 'latin-1'
# But they will always exist in the qsd location
assert isinstance(results.get('qsd'), dict)
assert results['qsd'].get('encoding') == 'latin-1'
assert results['qsd'].get('format') == 'invalid'
def test_config_base_detect_config_format():
"""
API: ConfigBase.detect_config_format
"""
# Garbage Handling
for garbage in (object(), None, 42):
# A response is always correctly returned
assert ConfigBase.detect_config_format(garbage) is None
# Empty files are valid
assert ConfigBase.detect_config_format('') is ConfigFormat.TEXT
# Valid Text Configuration
assert ConfigBase.detect_config_format("""
# A comment line over top of a URL
mailto://userb:pass@gmail.com
""") is ConfigFormat.TEXT
# A text file that has semi-colon as comment characters
# is valid too
assert ConfigBase.detect_config_format("""
; A comment line over top of a URL
mailto://userb:pass@gmail.com
""") is ConfigFormat.TEXT
# Valid YAML Configuration
assert ConfigBase.detect_config_format("""
# A comment line over top of a URL
version: 1
""") is ConfigFormat.YAML
# Just a whole lot of blank lines...
assert ConfigBase.detect_config_format('\n\n\n') is ConfigFormat.TEXT
# Invalid Config
assert ConfigBase.detect_config_format("3") is None
def test_config_base_config_parse():
"""
API: ConfigBase.config_parse
"""
# Garbage Handling
for garbage in (object(), None, 42):
# A response is always correctly returned
result = ConfigBase.config_parse(garbage)
# response is a tuple...
assert isinstance(result, tuple)
# containing 2 items (plugins, config)
assert len(result) == 2
# In the case of garbage in, we get garbage out; both lists are empty
assert result == (list(), list())
# Valid Text Configuration
result = ConfigBase.config_parse("""
# A comment line over top of a URL
mailto://userb:pass@gmail.com
""", asset=AppriseAsset())
# We expect to parse 1 entry from the above
assert isinstance(result, tuple)
assert len(result) == 2
# The first element is the number of notification services processed
assert len(result[0]) == 1
# If we index into the item, we can check to see the tags associate
# with it
assert len(result[0][0].tags) == 0
# The second is the number of configuration include lines parsed
assert len(result[1]) == 0
# Valid Configuration
result = ConfigBase.config_parse("""
# 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
- json://localhost:
- tag: devops, admin
""", asset=AppriseAsset())
# We expect to parse 2 entries from the above
assert isinstance(result, tuple)
assert len(result) == 2
assert isinstance(result[0], list)
assert len(result[0]) == 3
assert len(result[0][0].tags) == 0
assert len(result[0][1].tags) == 0
assert len(result[0][2].tags) == 2
# Test case where we pass in a bad format
result = ConfigBase.config_parse("""
; A comment line over top of a URL
mailto://userb:pass@gmail.com
""", config_format='invalid-format')
# This is not parseable despite the valid text
assert isinstance(result, tuple)
assert isinstance(result[0], list)
assert len(result[0]) == 0
result, _ = ConfigBase.config_parse("""
; A comment line over top of a URL
mailto://userb:pass@gmail.com
""", config_format=ConfigFormat.TEXT)
# Parseable
assert isinstance(result, list)
assert len(result) == 1
def test_config_base_discord_bug_report_01():
"""
API: ConfigBase.config_parse user feedback
A Discord report that a tag was not correctly assigned to a URL when
presented in the following format
urls:
- json://myhost:
- tag: test
userid: test
"""
result, config = ConfigBase.config_parse("""
urls:
- json://myhost:
- tag: test
userid: test
""", asset=AppriseAsset())
# We expect to parse 4 entries from the above
assert isinstance(result, list)
assert isinstance(config, list)
assert len(result) == 1
assert len(result[0].tags) == 1
assert 'test' in result[0].tags
def test_config_base_config_parse_text():
"""
API: ConfigBase.config_parse_text object
"""
# Garbage Handling
for garbage in (object(), None, 42):
# A response is always correctly returned
result = ConfigBase.config_parse_text(garbage)
# response is a tuple...
assert isinstance(result, tuple)
# containing 2 items (plugins, config)
assert len(result) == 2
# In the case of garbage in, we get garbage out; both lists are empty
assert result == (list(), list())
# Valid Configuration
result, config = ConfigBase.config_parse_text("""
# A completely invalid token on json string (it gets ignored)
# but the URL is still valid
json://localhost?invalid-token=nodashes
# A comment line over top of a URL
mailto://userb:pass@gmail.com
# Test a URL using it's native format; in this case Ryver
https://apprise.ryver.com/application/webhook/ckhrjW8w672m6HG
# Invalid URL as it's not associated with a plugin
# or a native url
https://not.a.native.url/
# A line with mulitiple tag assignments to it
taga,tagb=kde://
# An include statement to Apprise API with trailing spaces:
include http://localhost:8080/notify/apprise
# A relative include statement (with trailing spaces)
include apprise.cfg """, asset=AppriseAsset())
# We expect to parse 4 entries from the above
assert isinstance(result, list)
assert isinstance(config, list)
assert len(result) == 4
assert len(result[0].tags) == 0
# Our last element will have 2 tags associated with it
assert len(result[-1].tags) == 2
assert 'taga' in result[-1].tags
assert 'tagb' in result[-1].tags
assert len(config) == 2
assert 'http://localhost:8080/notify/apprise' in config
assert 'apprise.cfg' in config
# Here is a similar result set however this one has an invalid line
# in it which invalidates the entire file
result, config = ConfigBase.config_parse_text("""
# A comment line over top of a URL
mailto://userc:pass@gmail.com
# A line with mulitiple tag assignments to it
taga,tagb=windows://
I am an invalid line that does not follow any of the Apprise file rules!
""")
# We expect to parse 0 entries from the above because the invalid line
# invalidates the entire configuration file. This is for security reasons;
# we don't want to point at files load content in them just because they
# resemble an Apprise configuration.
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
# More invalid data
result, config = ConfigBase.config_parse_text("""
# An invalid URL
invalid://user:pass@gmail.com
# A tag without a url
taga=
# A very poorly structured url
sns://:@/
# Just 1 token provided
sns://T1JJ3T3L2/
# Even with the above invalid entries, we can still
# have valid include lines
include file:///etc/apprise.cfg
# An invalid include (nothing specified afterwards)
include
# An include of a config type we don't support
include invalid://
""")
# We expect to parse 0 entries from the above
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
# Test case where a comment is on it's own line with nothing else
result, config = ConfigBase.config_parse_text("#")
# We expect to parse 0 entries from the above
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
# Verify our tagging works when multiple tags are provided
result, config = ConfigBase.config_parse_text("""
tag1, tag2, tag3=json://user:pass@localhost
""")
assert isinstance(result, list)
assert len(result) == 1
assert len(result[0].tags) == 3
assert 'tag1' in result[0].tags
assert 'tag2' in result[0].tags
assert 'tag3' in result[0].tags
def test_config_base_config_tag_groups_text():
"""
API: ConfigBase.config_tag_groups_text object
"""
# Valid Configuration
result, config = ConfigBase.config_parse_text("""
# Tag assignments
groupA, groupB = tagB, tagC
# groupB doubles down as it takes the entries initialized above
# plus the added ones defined below
groupB = tagA, tagB, tagD
groupC = groupA, groupB, groupC, tagE
# Tag that recursively looks to more tags
groupD = groupC
# Assigned ourselves
groupX = groupX
# Set up a recursive loop
groupE = groupF
groupF = groupE
# Set up a larger recursive loop
groupG = groupH
groupH = groupI
groupI = groupJ
groupJ = groupK
groupK = groupG
# Bad assignments
groupM = , , ,
, , = , , ,
# int's and floats are okay
1 = 2
a = 5
# A comment line over top of a URL
4, groupB = mailto://userb:pass@gmail.com
# Tag Assignments
tagA,groupB=json://localhost
# More Tag Assignments
tagC,groupB=xml://localhost
# More Tag Assignments
groupD=form://localhost
""", asset=AppriseAsset())
# We expect to parse 4 entries from the above
assert isinstance(result, list)
assert isinstance(config, list)
assert len(result) == 4
# Our first element is our group tags
assert len(result[0].tags) == 2
assert 'groupB' in result[0].tags
assert '4' in result[0].tags
# No additional configuration is loaded
assert len(config) == 0
apobj = Apprise()
assert apobj.add(result)
# We match against 1 entry
assert len([x for x in apobj.find('tagA')]) == 1
assert len([x for x in apobj.find('tagB')]) == 0
assert len([x for x in apobj.find('groupA')]) == 1
assert len([x for x in apobj.find('groupB')]) == 3
assert len([x for x in apobj.find('groupC')]) == 2
assert len([x for x in apobj.find('groupD')]) == 3
# Invalid Assignment
result, config = ConfigBase.config_parse_text("""
# Must have something to equal or it's a bad line
group =
# A tag Assignments that is never gotten to as the line
# above is bad
groupD=form://localhost
""")
# We expect to parse 0 entries from the above
assert isinstance(result, list)
assert isinstance(config, list)
assert len(result) == 0
assert len(config) == 0
# Invalid Assignment
result, config = ConfigBase.config_parse_text("""
# Rundant assignment
group = group
# Our group assignment
group=windows://
""")
# the redundant assignment does us no harm; but it doesn't grant us any
# value either
assert isinstance(result, list)
assert len(result) == 1
# Our first element is our group tags
assert len(result[0].tags) == 1
assert 'group' in result[0].tags
# There were no include entries defined
assert len(config) == 0
# More invalid data
result, config = ConfigBase.config_parse_text("""
# A tag without a url or group assignment
taga=
""")
# We expect to parse 0 entries from the above
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
result, config = ConfigBase.config_parse_text("""
# A tag without a url or group assignment
taga= %%INVALID
""")
# We expect to parse 0 entries from the above
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
def test_config_base_config_parse_text_with_url():
"""
API: ConfigBase.config_parse_text object_with_url
"""
# Here is a similar result set however this one has an invalid line
# in it which invalidates the entire file
result, config = ConfigBase.config_parse_text("""
# Test a URL that has a URL as an argument
json://user:pass@localhost?+arg=http://example.com?arg2=1&arg3=3
""")
# No tag is parsed, but our URL successfully parses as is
assert isinstance(result, list)
assert len(result) == 1
assert len(result[0].tags) == 0
# Verify our URL is correctly captured
assert '%2Barg=http%3A%2F%2Fexample.com%3Farg2%3D1' in result[0].url()
assert 'json://user:pass@localhost/' in result[0].url()
# There were no include entries defined
assert len(config) == 0
# Pass in our configuration again
result, config = ConfigBase.config_parse_text(result[0].url())
# Verify that our results repeat themselves
assert isinstance(result, list)
assert len(result) == 1
assert len(result[0].tags) == 0
assert '%2Barg=http%3A%2F%2Fexample.com%3Farg2%3D1' in result[0].url()
assert 'json://user:pass@localhost/' in result[0].url()
assert len(config) == 0
def test_config_base_config_parse_yaml():
"""
API: ConfigBase.config_parse_yaml object
"""
# general reference used below
asset = AppriseAsset()
# Garbage Handling
for garbage in (object(), None, '', 42):
# A response is always correctly returned
result = ConfigBase.config_parse_yaml(garbage)
# response is a tuple...
assert isinstance(result, tuple)
# containing 2 items (plugins, config)
assert len(result) == 2
# In the case of garbage in, we get garbage out; both lists are empty
assert result == (list(), list())
# Invalid Version
result, config = ConfigBase.config_parse_yaml("version: 2a", asset=asset)
# Invalid data gets us an empty result set
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
# Invalid Syntax (throws a ScannerError)
result, config = 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
# There were no include entries defined
assert len(config) == 0
# Missing url token
result, config = 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
# There were no include entries defined
assert len(config) == 0
# No urls defined
result, config = 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
# There were no include entries defined
assert len(config) == 0
# Invalid url defined
result, config = 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
# There were no include entries defined
assert len(config) == 0
# Invalid url/schema
result, config = 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
# There were no include entries defined
assert len(config) == 0
# Invalid url/schema
result, config = 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
# There were no include entries defined
assert len(config) == 0
# Invalid url/schema
result, config = ConfigBase.config_parse_yaml("""
# Include entry with nothing associated with it
include:
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
# There were no include entries defined
assert len(config) == 0
# Invalid url/schema
result, config = 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
# There were no include entries defined
assert len(config) == 0
# Invalid url/schema
result, config = ConfigBase.config_parse_yaml("""
urls:
# a very invalid sns entry
- sns://T1JJ3T3L2/
- sns://:@/:
- invalid: test
- sns://T1JJ3T3L2/:
- invalid: test
- _invalid: Token can not start with an underscore
# some strangeness
-
-
- test
""", asset=asset)
# Invalid data gets us an empty result set
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
# Valid Configuration
result, config = ConfigBase.config_parse_yaml("""
# if no version is specified then version 1 is presumed
version: 1
# Including by dict
include:
# File includes
- file:///absolute/path/
- relative/path
# Trailing colon shouldn't disrupt include
- http://test.com:
# invalid (numeric)
- 4
# some strangeness
-
-
- test
#
# Define your notification urls:
#
urls:
- pbul://o.gn5kj6nfhv736I7jC3cj3QLRiyhgl98b
- mailto://test:password@gmail.com
- https://apprise.ryver.com/application/webhook/ckhrjW8w672m6HG
- https://not.a.native.url/
# A completely invalid token on json string (it gets ignored)
# but the URL is still valid
- json://localhost?invalid-token=nodashes
""", asset=asset)
# We expect to parse 4 entries from the above
# The Ryver one is in a native form and the 4th one is invalid
assert isinstance(result, list)
assert len(result) == 4
assert len(result[0].tags) == 0
# There were 3 include entries
assert len(config) == 3
assert 'file:///absolute/path/' in config
assert 'relative/path' in config
assert 'http://test.com' in config
# Valid Configuration
result, config = ConfigBase.config_parse_yaml("""
# A single line include is supported
include: http://localhost:8080/notify/apprise
urls:
# The following generates 1 service
- json://localhost:
tag: my-custom-tag, my-other-tag
# The following also generates 1 service
- json://localhost:
- tag: my-custom-tag, my-other-tag
# How to stack multiple entries (this generates 2):
- mailto://user:123abc@yahoo.ca:
- to: test@examle.com
- to: test2@examle.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 Ryver URL (using Native format); still accepted
- https://apprise.ryver.com/application/webhook/ckhrjW8w672m6HG:
# An invalid URL with colon (ignored)
- https://not.a.native.url/:
# A telegram entry (returns a None in parse_url())
- tgram://invalid
""", asset=asset)
# We expect to parse 6 entries from the above because the tgram:// entry
# would have failed to be loaded
assert isinstance(result, list)
assert len(result) == 6
assert len(result[0].tags) == 2
# Our single line included
assert len(config) == 1
assert 'http://localhost:8080/notify/apprise' in config
# Global Tags
result, config = ConfigBase.config_parse_yaml("""
# Global Tags stacked as a list
tag:
- admin
- devops
urls:
- json://localhost
- dbus://
""", asset=asset)
# We expect to parse 2 entries from the above
assert isinstance(result, list)
assert len(result) == 2
# There were no include entries defined
assert len(config) == 0
# 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, config = 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 2 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
# There were no include entries defined
assert len(config) == 0
# An invalid set of entries
result, config = ConfigBase.config_parse_yaml("""
urls:
# The following tags will get added to the global set
- json://localhost:
-
-
- entry
""", asset=asset)
# We expect to parse 0 entries from the above
assert isinstance(result, list)
assert len(result) == 0
# There were no include entries defined
assert len(config) == 0
# An asset we'll manipulate; set some system flags
asset = AppriseAsset(_uid="abc123", _recursion=1)
# Global Tags
result, config = 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
async_mode: no
# System flags should never get set
_uid: custom_id
_recursion: 100
# 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 1 entries from the above
assert isinstance(result, list)
assert len(result) == 1
# There were no include entries defined
assert len(config) == 0
assert asset.app_id == "AppriseTest"
assert asset.app_desc == "Apprise Test Notifications"
assert asset.app_url == "http://nuxref.com"
# Verify our system flags retain only the value they were initialized to
assert asset._uid == "abc123"
assert asset._recursion == 1
# Boolean types stay boolean
assert asset.async_mode is False
# 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, str)
assert asset.image_url_mask == ""
assert isinstance(asset.image_url_logo, str)
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, config = 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
# There were no include entries defined
assert len(config) == 0
# Valid Configuration (multi inline configuration entries)
result, config = ConfigBase.config_parse_yaml("""
# A configuration file that contains 2 includes separated by a comma and/or
# space:
include: http://localhost:8080/notify/apprise, http://localhost/apprise/cfg
""", asset=asset)
# We will have loaded no results
assert isinstance(result, list)
assert len(result) == 0
# But our two configuration files will be present:
assert len(config) == 2
assert 'http://localhost:8080/notify/apprise' in config
assert 'http://localhost/apprise/cfg' in config
# Valid Configuration (another way of specifying more then one include)
result, config = ConfigBase.config_parse_yaml("""
# A configuration file that contains 4 includes on their own
# lines beneath the keyword `include`:
include:
http://localhost:8080/notify/apprise
http://localhost/apprise/cfg01
http://localhost/apprise/cfg02
http://localhost/apprise/cfg03
""", asset=asset)
# We will have loaded no results
assert isinstance(result, list)
assert len(result) == 0
# But our 4 configuration files will be present:
assert len(config) == 4
assert 'http://localhost:8080/notify/apprise' in config
assert 'http://localhost/apprise/cfg01' in config
assert 'http://localhost/apprise/cfg02' in config
assert 'http://localhost/apprise/cfg03' in config
# Test a configuration with an invalid schema with options
result, config = ConfigBase.config_parse_yaml("""
urls:
- invalid://:
tag: 'invalid'
:name: 'Testing2'
:body: 'test body2'
:title: 'test title2'
""", asset=asset)
# We will have loaded no results
assert isinstance(result, list)
assert len(result) == 0
# Valid Configuration (we allow comma separated entries for
# each defined bullet)
result, config = ConfigBase.config_parse_yaml("""
# A configuration file that contains 4 includes on their own
# lines beneath the keyword `include`:
include:
- http://localhost:8080/notify/apprise, http://localhost/apprise/cfg01
http://localhost/apprise/cfg02
- http://localhost/apprise/cfg03
""", asset=asset)
# We will have loaded no results
assert isinstance(result, list)
assert len(result) == 0
# But our 4 configuration files will be present:
assert len(config) == 4
assert 'http://localhost:8080/notify/apprise' in config
assert 'http://localhost/apprise/cfg01' in config
assert 'http://localhost/apprise/cfg02' in config
assert 'http://localhost/apprise/cfg03' in config
def test_yaml_vs_text_tagging():
"""
API: ConfigBase YAML vs TEXT tagging
"""
yaml_result, _ = ConfigBase.config_parse_yaml("""
urls:
- mailtos://lead2gold:yesqbrulvaelyxve@gmail.com:
tag: mytag
""")
assert yaml_result
text_result, _ = ConfigBase.config_parse_text("""
mytag=mailtos://lead2gold:yesqbrulvaelyxve@gmail.com
""")
assert text_result
# Now we compare our results and verify they are the same
assert len(yaml_result) == len(text_result)
assert isinstance(yaml_result[0], NotifyEmail)
assert isinstance(text_result[0], NotifyEmail)
assert 'mytag' in text_result[0]
assert 'mytag' in yaml_result[0]
def test_config_base_config_tag_groups_yaml_01():
"""
API: ConfigBase.config_tag_groups_yaml #1 object
"""
# general reference used below
asset = AppriseAsset()
# Valid Configuration
result, config = ConfigBase.config_parse_yaml("""
# if no version is specified then version 1 is presumed
version: 1
groups:
- group1: tagB, tagC, tagNotAssigned
- group2:
- tagA
- tagC
- group3:
- tagD: optional comment
- tagA: optional comment #2
# No assignment
- group4
# No assignment type 2
- group5:
# Integer assignment
- group6: 3
- group6: 3, 4, 5, test
- group6: 3.5, tagC
# Recursion
- groupA: groupB
- groupB: groupA
# And Again... (just because)
- groupA: groupB
- groupB: groupA
# Self assignment
- groupX: groupX
# Set up a larger recursive loop
- groupG: groupH
- groupH: groupI, groupJ
- groupI: groupJ, groupG
- groupJ: groupK, groupH, groupI
- groupK: groupG
# No tags assigned
- groupK: ",, , ,"
- " , ": ",, , ,"
# Multi Assignments
- groupL, groupM: tagD, tagA
- 4, groupN:
- tagD
- tagE, TagA
# Add one more tag to groupL making it different then GroupM by 1
- groupL: tagB
#
# Define your notification urls:
#
urls:
- form://localhost:
- tag: tagA
- mailto://test:password@gmail.com:
- tag: tagB
- xml://localhost:
- tag: tagC
- json://localhost:
- tag: tagD, tagA
""", asset=asset)
# We expect to parse 4 entries from the above
assert isinstance(result, list)
assert isinstance(config, list)
assert len(result) == 4
# Our first element is our group tags
assert len(result[0].tags) == 5
assert 'group2' in result[0].tags
assert 'group3' in result[0].tags
assert 'groupL' in result[0].tags
assert 'groupM' in result[0].tags
assert 'tagA' in result[0].tags
# No additional configuration is loaded
assert len(config) == 0
apobj = Apprise()
assert apobj.add(result)
# We match against 1 entry
assert len([x for x in apobj.find('tagA')]) == 2
assert len([x for x in apobj.find('tagB')]) == 1
assert len([x for x in apobj.find('tagC')]) == 1
assert len([x for x in apobj.find('tagD')]) == 1
assert len([x for x in apobj.find('group1')]) == 2
assert len([x for x in apobj.find('group2')]) == 3
assert len([x for x in apobj.find('group3')]) == 2
assert len([x for x in apobj.find('group4')]) == 0
assert len([x for x in apobj.find('group5')]) == 0
# json:// -- group6 -> 4 -> TagA
# xml:// -- group6 -> TagC
assert len([x for x in apobj.find('group6')]) == 2
assert len([x for x in apobj.find('4')]) == 1
assert len([x for x in apobj.find('groupN')]) == 1
def test_config_base_config_tag_groups_yaml_02():
"""
API: ConfigBase.config_tag_groups_yaml #2 object
"""
# general reference used below
asset = AppriseAsset()
# Valid Configuration
result, config = ConfigBase.config_parse_yaml("""
# if no version is specified then version 1 is presumed
version: 1
groups:
group1: tagB, tagC, tagNotAssigned
group2:
- tagA
- tagC
group3:
- tagD: optional comment
- tagA: optional comment #2
# No assignment type 2
group5:
# Integer assignment (since it's not a list, the last element prevails
# and replaces the above); '4' does not get appended as it would in
# the event this was a list instead
group6: 3
group6: 3, 4, 5, test
group6: 3.5, tagC
# Recursion
groupA: groupB
groupB: groupA
# And Again... (just because)
groupA: groupB
groupB: groupA
# Self assignment
groupX: groupX
# Set up a larger recursive loop
groupG: groupH
groupH: groupI, groupJ
groupI: groupJ, groupG
groupJ: groupK, groupH, groupI
groupK: groupG
# No tags assigned
groupK: ",, , ,"
" , ": ",, , ,"
# Multi Assignments
groupL, groupM: tagD, tagA
4, groupN:
- tagD
- tagE, TagA
# Add one more tag to groupL making it different then GroupM by 1
groupL: tagB
#
# Define your notification urls:
#
urls:
- form://localhost:
- tag: tagA
- mailto://test:password@gmail.com:
- tag: tagB
- xml://localhost:
- tag: tagC
- json://localhost:
- tag: tagD, tagA
""", asset=asset)
# We expect to parse 4 entries from the above
assert isinstance(result, list)
assert isinstance(config, list)
assert len(result) == 4
# Our first element is our group tags
assert len(result[0].tags) == 5
assert 'group2' in result[0].tags
assert 'group3' in result[0].tags
assert 'groupL' in result[0].tags
assert 'groupM' in result[0].tags
assert 'tagA' in result[0].tags
# No additional configuration is loaded
assert len(config) == 0
apobj = Apprise()
assert apobj.add(result)
# We match against 1 entry
assert len([x for x in apobj.find('tagA')]) == 2
assert len([x for x in apobj.find('tagB')]) == 1
assert len([x for x in apobj.find('tagC')]) == 1
assert len([x for x in apobj.find('tagD')]) == 1
assert len([x for x in apobj.find('group1')]) == 2
assert len([x for x in apobj.find('group2')]) == 3
assert len([x for x in apobj.find('group3')]) == 2
assert len([x for x in apobj.find('group4')]) == 0
assert len([x for x in apobj.find('group5')]) == 0
# NOT json:// -- group6 -> 4 -> TagA (not appended because dict storage)
# ^
# |
# See: test_config_base_config_tag_groups_yaml_01 (above)
# dict storage (as this tests for) causes last entry to
# prevail; previous assignments are lost
#
# xml:// -- group6 -> TagC
assert len([x for x in apobj.find('group6')]) == 1
assert len([x for x in apobj.find('4')]) == 1
assert len([x for x in apobj.find('groupN')]) == 1
assert len([x for x in apobj.find('groupK')]) == 0
def test_config_base_config_parse_yaml_globals():
"""
API: ConfigBase.config_parse_yaml globals
"""
# general reference used below
asset = AppriseAsset()
# Invalid Syntax (throws a ScannerError)
results, config = ConfigBase.config_parse_yaml(cleandoc("""
urls:
- jsons://localhost1:
- to: jeff@gmail.com
tag: jeff, customer
cto: 30
rto: 30
verify: no
- jsons://localhost2?cto=30&rto=30&verify=no:
- to: json@gmail.com
tag: json, customer
"""), asset=asset)
# Invalid data gets us an empty result set
assert isinstance(results, list)
# Our results loaded
assert len(results) == 2
assert len(config) == 0
# Now verify that our global variables correctly initialized
for entry in results:
assert entry.verify_certificate is False
assert entry.socket_read_timeout == 30
assert entry.socket_connect_timeout == 30
# This test fails on CentOS 8.x so it was moved into it's own function
# so it could be bypassed. The ability to use lists in YAML files didn't
# appear to happen until later on; it's certainly not available in v3.12
# which was what shipped with CentOS v8 at the time.
@pytest.mark.skipif(int(yaml.__version__.split('.')[0]) <= 3,
reason="requires pyaml v4.x or higher.")
def test_config_base_config_parse_yaml_list():
"""
API: ConfigBase.config_parse_yaml list parsing
"""
# general reference used below
asset = AppriseAsset()
# Invalid url/schema
result, config = ConfigBase.config_parse_yaml("""
# no lists... just no
urls: [milk, pumpkin pie, eggs, juice]
# Including by list is okay
include: [file:///absolute/path/, relative/path, http://test.com]
""", asset=asset)
# Invalid data gets us an empty result set
assert isinstance(result, list)
assert len(result) == 0
# There were 3 include entries
assert len(config) == 3
assert 'file:///absolute/path/' in config
assert 'relative/path' in config
assert 'http://test.com' in config
Reference in New Issue View Git Blame Copy Permalink