apprise/test/test_apprise_config.py

# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

import six
import io
import mock
from apprise import NotifyFormat
from apprise.Apprise import Apprise
from apprise.AppriseConfig import AppriseConfig
from apprise.AppriseAsset import AppriseAsset
from apprise.config.ConfigBase import ConfigBase
from apprise.plugins.NotifyBase import NotifyBase

from apprise.config import SCHEMA_MAP as CONFIG_SCHEMA_MAP
from apprise.plugins import SCHEMA_MAP as NOTIFY_SCHEMA_MAP
from apprise.config import __load_matrix
from apprise.config.ConfigFile import ConfigFile

# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)


def test_apprise_config(tmpdir):
    """
    API: AppriseConfig basic testing

    """

    # Create ourselves a config object
    ac = AppriseConfig()

    # There are no servers loaded
    assert len(ac) == 0

    # lets try anyway
    assert len(ac.servers()) == 0

    t = tmpdir.mkdir("simple-formatting").join("apprise")
    t.write("""
    # A comment line over top of a URL
    mailto://usera:pass@gmail.com

    # A line with mulitiple tag assignments to it
    taga,tagb=gnome://

    # Event if there is accidental leading spaces, this configuation
    # is accepting of htat and will not exclude them
                tagc=kde://

    # A very poorly structured url
    sns://:@/

    # Just 1 token provided causes exception
    sns://T1JJ3T3L2/
    """)

    # Create ourselves a config object
    ac = AppriseConfig(paths=str(t))

    # One configuration file should have been found
    assert len(ac) == 1

    # We should be able to read our 3 servers from that
    assert len(ac.servers()) == 3

    # Get our URL back
    assert isinstance(ac[0].url(), six.string_types)

    # Test cases where our URL is invalid
    t = tmpdir.mkdir("strange-lines").join("apprise")
    t.write("""
    # basicly this consists of defined tags and no url
    tag=
    """)

    # Create ourselves a config object
    ac = AppriseConfig(paths=str(t), asset=AppriseAsset())

    # One configuration file should have been found
    assert len(ac) == 1

    # No urls were set
    assert len(ac.servers()) == 0

    # Create a ConfigBase object
    cb = ConfigBase()

    # Test adding of all entries
    assert ac.add(configs=cb, asset=AppriseAsset(), tag='test') is True

    # Test adding of all entries
    assert ac.add(
        configs=['file://?', ], asset=AppriseAsset(), tag='test') is False

    # Test the adding of garbage
    assert ac.add(configs=object()) is False

    # Try again but enforce our format
    ac = AppriseConfig(paths='file://{}?format=text'.format(str(t)))

    # One configuration file should have been found
    assert len(ac) == 1

    # No urls were set
    assert len(ac.servers()) == 0

    #
    # Test Internatialization and the handling of unicode characters
    #
    istr = """
        # Iñtërnâtiônàlization Testing
        windows://"""

    if six.PY2:
        # decode string into unicode
        istr = istr.decode('utf-8')

    # Write our content to our file
    t = tmpdir.mkdir("internationalization").join("apprise")
    with io.open(str(t), 'wb') as f:
        f.write(istr.encode('latin-1'))

    # Create ourselves a config object
    ac = AppriseConfig(paths=str(t))

    # One configuration file should have been found
    assert len(ac) == 1

    # This will fail because our default encoding is utf-8; however the file
    # we opened was not; it was latin-1 and could not be parsed.
    assert len(ac.servers()) == 0

    # Test iterator
    count = 0
    for entry in ac:
        count += 1
    assert len(ac) == count

    # We can fix this though; set our encoding to latin-1
    ac = AppriseConfig(paths='file://{}?encoding=latin-1'.format(str(t)))

    # One configuration file should have been found
    assert len(ac) == 1

    # Our URL should be found
    assert len(ac.servers()) == 1

    # Get our URL back
    assert isinstance(ac[0].url(), six.string_types)

    # pop an entry from our list
    assert isinstance(ac.pop(0), ConfigBase) is True

    # Determine we have no more configuration entries loaded
    assert len(ac) == 0

    #
    # Test buffer handling (and overflow)
    t = tmpdir.mkdir("buffer-handling").join("apprise")
    buf = "gnome://"
    t.write(buf)

    # Reset our config object
    ac.clear()

    # Create ourselves a config object
    ac = AppriseConfig(paths=str(t))

    # update our length to be the size of our actual file
    ac[0].max_buffer_size = len(buf)

    # One configuration file should have been found
    assert len(ac) == 1

    assert len(ac.servers()) == 1

    # update our buffer size to be slightly smaller then what we allow
    ac[0].max_buffer_size = len(buf) - 1

    # Content is automatically cached; so even though we adjusted the buffer
    # above, our results have been cached so we get a 1 response.
    assert len(ac.servers()) == 1

    # Now do the same check but force a flushed cache
    assert len(ac.servers(cache=False)) == 0


def test_apprise_multi_config_entries(tmpdir):
    """
    API: AppriseConfig basic multi-adding functionality

    """
    # temporary file to work with
    t = tmpdir.mkdir("apprise-multi-add").join("apprise")
    buf = """
    good://hostname
    """
    t.write(buf)

    # temporary empty file to work with
    te = tmpdir.join("apprise-multi-add", "apprise-empty")
    te.write("")

    # Define our good:// url
    class GoodNotification(NotifyBase):
        def __init__(self, **kwargs):
            super(GoodNotification, self).__init__(
                notify_format=NotifyFormat.HTML, **kwargs)

        def notify(self, **kwargs):
            # Pretend everything is okay
            return True

        def url(self):
            # support url()
            return ''

    # Store our good notification in our schema map
    NOTIFY_SCHEMA_MAP['good'] = GoodNotification

    # Create ourselves a config object
    ac = AppriseConfig()

    # There are no servers loaded
    assert len(ac) == 0

    # Support adding of muilt strings and objects:
    assert ac.add(configs=(str(t), str(t))) is True
    assert ac.add(configs=(
        ConfigFile(path=str(te)), ConfigFile(path=str(t)))) is True

    # don't support the adding of invalid content
    assert ac.add(configs=(object(), object())) is False
    assert ac.add(configs=object()) is False

    # Try to pop an element out of range
    try:
        ac.server_pop(len(ac.servers()))
        # We should have thrown an exception here
        assert False

    except IndexError:
        # We expect to be here
        assert True

    # Pop our elements
    while len(ac.servers()) > 0:
        assert isinstance(
            ac.server_pop(len(ac.servers()) - 1), NotifyBase) is True


def test_apprise_config_tagging(tmpdir):
    """
    API: AppriseConfig tagging

    """

    # temporary file to work with
    t = tmpdir.mkdir("tagging").join("apprise")
    buf = "gnome://"
    t.write(buf)

    # Create ourselves a config object
    ac = AppriseConfig()

    # Add an item associated with tag a
    assert ac.add(configs=str(t), asset=AppriseAsset(), tag='a') is True
    # Add an item associated with tag b
    assert ac.add(configs=str(t), asset=AppriseAsset(), tag='b') is True
    # Add an item associated with tag a or b
    assert ac.add(configs=str(t), asset=AppriseAsset(), tag='a,b') is True

    # Now filter: a:
    assert len(ac.servers(tag='a')) == 2
    # Now filter: a or b:
    assert len(ac.servers(tag='a,b')) == 3
    # Now filter: a and b
    assert len(ac.servers(tag=[('a', 'b')])) == 1


def test_apprise_instantiate():
    """
    API: AppriseConfig.instantiate()

    """
    assert AppriseConfig.instantiate(
        'file://?', suppress_exceptions=True) is None

    assert AppriseConfig.instantiate(
        'invalid://?', suppress_exceptions=True) is None

    class BadConfig(ConfigBase):
        def __init__(self, **kwargs):
            super(BadConfig, self).__init__(**kwargs)

            # We fail whenever we're initialized
            raise TypeError()

    # Store our bad configuration in our schema map
    CONFIG_SCHEMA_MAP['bad'] = BadConfig

    try:
        AppriseConfig.instantiate(
            'bad://path', suppress_exceptions=False)
        # We should never make it to this line
        assert False

    except TypeError:
        # Exception caught as expected
        assert True

    # Same call but exceptions suppressed
    assert AppriseConfig.instantiate(
        'bad://path', suppress_exceptions=True) is None


def test_apprise_config_with_apprise_obj(tmpdir):
    """
    API: ConfigBase.parse_inaccessible_text_file

    """

    # temporary file to work with
    t = tmpdir.mkdir("apprise-obj").join("apprise")
    buf = """
    good://hostname
    localhost=good://localhost
    """
    t.write(buf)

    # Define our good:// url
    class GoodNotification(NotifyBase):
        def __init__(self, **kwargs):
            super(GoodNotification, self).__init__(
                notify_format=NotifyFormat.HTML, **kwargs)

        def notify(self, **kwargs):
            # Pretend everything is okay
            return True

        def url(self):
            # support url()
            return ''

    # Store our good notification in our schema map
    NOTIFY_SCHEMA_MAP['good'] = GoodNotification

    # Create ourselves a config object
    ac = AppriseConfig(cache=False)

    # Nothing loaded yet
    assert len(ac) == 0

    # Add an item associated with tag a
    assert ac.add(configs=str(t), asset=AppriseAsset(), tag='a') is True

    # One configuration file
    assert len(ac) == 1

    # 2 services found in it
    assert len(ac.servers()) == 2

    # Pop one of them (at index 0)
    ac.server_pop(0)

    # Verify that it no longer listed
    assert len(ac.servers()) == 1

    # Test our ability to add Config objects to our apprise object
    a = Apprise()

    # Add our configuration object
    assert a.add(servers=ac) is True

    # Detect our 1 entry (originally there were 2 but we deleted one)
    assert len(a) == 1

    # Notify our service
    assert a.notify(body='apprise configuration power!') is True

    # Add our configuration object
    assert a.add(
        servers=[AppriseConfig(str(t)), AppriseConfig(str(t))]) is True

    # Detect our 5 loaded entries now; 1 from first config, and another
    # 2x2 based on adding our list above
    assert len(a) == 5

    # We can't add garbage
    assert a.add(servers=object()) is False
    assert a.add(servers=[object(), object()]) is False

    # Our length is unchanged
    assert len(a) == 5

    # reference index 0 of our list
    ref = a[0]
    assert isinstance(ref, NotifyBase) is True

    # Our length is unchanged
    assert len(a) == 5

    # pop the index
    ref_popped = a.pop(0)

    # Verify our response
    assert isinstance(ref_popped, NotifyBase) is True

    # Our length drops by 1
    assert len(a) == 4

    # Content popped is the same as one referenced by index
    # earlier
    assert ref == ref_popped

    # pop an index out of range
    try:
        a.pop(len(a))
        # We'll thrown an IndexError and not make it this far
        assert False

    except IndexError:
        # As expected
        assert True

    # Our length remains unchanged
    assert len(a) == 4

    # Reference content out of range
    try:
        a[len(a)]

        # We'll thrown an IndexError and not make it this far
        assert False

    except IndexError:
        # As expected
        assert True

    # reference index at the end of our list
    ref = a[len(a) - 1]

    # Verify our response
    assert isinstance(ref, NotifyBase) is True

    # Our length stays the same
    assert len(a) == 4

    # We can pop from the back of the list without a problem too
    ref_popped = a.pop(len(a) - 1)

    # Verify our response
    assert isinstance(ref_popped, NotifyBase) is True

    # Content popped is the same as one referenced by index
    # earlier
    assert ref == ref_popped

    # Our length drops by 1
    assert len(a) == 3

    # Now we'll test adding another element to the list so that it mixes up
    # our response object.
    # Below we add 3 different types, a ConfigBase, NotifyBase, and URL
    assert a.add(
        servers=[
            ConfigFile(path=(str(t))),
            'good://another.host',
            GoodNotification(**{'host': 'nuxref.com'})]) is True

    # Our length increases by 4 (2 entries in the config file, + 2 others)
    assert len(a) == 7

    # reference index at the end of our list
    ref = a[len(a) - 1]

    # Verify our response
    assert isinstance(ref, NotifyBase) is True

    # We can pop from the back of the list without a problem too
    ref_popped = a.pop(len(a) - 1)

    # Verify our response
    assert isinstance(ref_popped, NotifyBase) is True

    # Content popped is the same as one referenced by index
    # earlier
    assert ref == ref_popped

    # Our length drops by 1
    assert len(a) == 6

    # pop our list
    while len(a) > 0:
        assert isinstance(a.pop(len(a) - 1), NotifyBase) is True


def test_apprise_config_matrix_load():
    """
    API: AppriseConfig() matrix initialization

    """

    import apprise

    class ConfigDummy(ConfigBase):
        """
        A dummy wrapper for testing the different options in the load_matrix
        function
        """

        # The default descriptive name associated with the Notification
        service_name = 'dummy'

        # protocol as tuple
        protocol = ('uh', 'oh')

        # secure protocol as tuple
        secure_protocol = ('no', 'yes')

    class ConfigDummy2(ConfigBase):
        """
        A dummy wrapper for testing the different options in the load_matrix
        function
        """

        # The default descriptive name associated with the Notification
        service_name = 'dummy2'

        # secure protocol as tuple
        secure_protocol = ('true', 'false')

    class ConfigDummy3(ConfigBase):
        """
        A dummy wrapper for testing the different options in the load_matrix
        function
        """

        # The default descriptive name associated with the Notification
        service_name = 'dummy3'

        # secure protocol as string
        secure_protocol = 'true'

    class ConfigDummy4(ConfigBase):
        """
        A dummy wrapper for testing the different options in the load_matrix
        function
        """

        # The default descriptive name associated with the Notification
        service_name = 'dummy4'

        # protocol as string
        protocol = 'true'

    # Generate ourselfs a fake entry
    apprise.config.ConfigDummy = ConfigDummy
    apprise.config.ConfigDummy2 = ConfigDummy2
    apprise.config.ConfigDummy3 = ConfigDummy3
    apprise.config.ConfigDummy4 = ConfigDummy4

    __load_matrix()

    # Call it again so we detect our entries already loaded
    __load_matrix()


@mock.patch('os.path.getsize')
def test_config_base_parse_inaccessible_text_file(mock_getsize, tmpdir):
    """
    API: ConfigBase.parse_inaccessible_text_file

    """

    # temporary file to work with
    t = tmpdir.mkdir("inaccessible").join("apprise")
    buf = "gnome://"
    t.write(buf)

    # Set getsize return value
    mock_getsize.return_value = None
    mock_getsize.side_effect = OSError

    # Create ourselves a config object
    ac = AppriseConfig(paths=str(t))

    # The following internally throws an exception but still counts
    # as a loaded configuration file
    assert len(ac) == 1

    # Thus no notifications are loaded
    assert len(ac.servers()) == 0


def test_config_base_parse_yaml_file(tmpdir):
    """
    API: ConfigBase.parse_yaml_file

    """
    t = tmpdir.mkdir("empty-file").join("apprise.yml")
    t.write("")

    # Create ourselves a config object
    ac = AppriseConfig(paths=str(t))

    # The number of configuration files that exist
    assert len(ac) == 1

    # no notifications are loaded
    assert len(ac.servers()) == 0