apprise/test/test_apprise_attachments.py

# -*- 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
import json
import requests
from unittest import mock
from os.path import getsize
from os.path import join
from os.path import dirname
from inspect import cleandoc
from apprise import Apprise, AppriseAsset
from apprise import AttachmentManager
from apprise.apprise_attachment import AppriseAttachment
from apprise.attachment import AttachBase
from apprise.common import ContentLocation

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

TEST_VAR_DIR = join(dirname(__file__), 'var')

# Grant access to our Attachment Manager Singleton
A_MGR = AttachmentManager()


def test_apprise_attachment():
    """
    API: AppriseAttachment basic testing

    """

    # Create ourselves an attachment object
    aa = AppriseAttachment()

    # There are no attachents loaded
    assert len(aa) == 0

    # Object can be directly checked as a boolean; response is False
    # when there are no entries loaded
    assert not aa

    # An attachment object using a custom Apprise Asset object
    # Set a cache expiry of 5 minutes (300 seconds)
    aa = AppriseAttachment(asset=AppriseAsset(), cache=300)

    # still no attachments added
    assert len(aa) == 0

    # Add a file by it's path
    path = join(TEST_VAR_DIR, 'apprise-test.gif')
    assert aa.add(path)

    # There is now 1 attachment
    assert len(aa) == 1

    # our attachment took on our cache value
    assert aa[0].cache == 300

    # we can test the object as a boolean and get a value of True now
    assert aa

    # Add another entry already in it's AttachBase format
    response = AppriseAttachment.instantiate(path, cache=True)
    assert isinstance(response, AttachBase)
    assert aa.add(response, asset=AppriseAsset())

    # There is now 2 attachments
    assert len(aa) == 2

    # Cache is initialized to True
    assert aa[1].cache is True

    # Reset our object
    aa = AppriseAttachment()

    # We can add by lists as well in a variety of formats
    attachments = (
        path,
        'file://{}?name=newfilename.gif?cache=120'.format(path),
        AppriseAttachment.instantiate(
            'file://{}?name=anotherfilename.gif'.format(path), cache=100),
    )

    # Add them
    assert aa.add(attachments, cache=False)

    # There is now 3 attachments
    assert len(aa) == 3

    # Take on our fixed cache value of False.
    # The last entry will have our set value of 100
    assert aa[0].cache is False
    # Even though we set a value of 120, we take on the value of False because
    # it was forced on the instantiate call
    assert aa[1].cache is False
    assert aa[2].cache == 100

    # We can pop the last element off of the list as well
    attachment = aa.pop()
    assert isinstance(attachment, AttachBase)
    # we can test of the attachment is valid using a boolean check:
    assert attachment
    assert len(aa) == 2
    assert attachment.path == path
    assert attachment.name == 'anotherfilename.gif'
    assert attachment.mimetype == 'image/gif'

    # elements can also be directly indexed
    assert isinstance(aa[0], AttachBase)
    assert isinstance(aa[1], AttachBase)

    with pytest.raises(IndexError):
        aa[2]

    # We can iterate over attachments too:
    for count, a in enumerate(aa):
        assert isinstance(a, AttachBase)

        # we'll never iterate more then the number of entries in our object
        assert count < len(aa)

    # Get the file-size of our image
    expected_size = getsize(path) * len(aa)

    # verify that's what we get as a result
    assert aa.size() == expected_size

    # Attachments can also be loaded during the instantiation of the
    # AppriseAttachment object
    aa = AppriseAttachment(attachments)

    # There is now 3 attachments
    assert len(aa) == 3

    # Reset our object
    aa.clear()
    assert len(aa) == 0
    assert not aa

    assert aa.add(AppriseAttachment.instantiate(
        'file://{}?name=andanother.png&cache=Yes'.format(path)))
    assert aa.add(AppriseAttachment.instantiate(
        'file://{}?name=andanother.png&cache=No'.format(path)))
    AppriseAttachment.instantiate(
        'file://{}?name=andanother.png&cache=600'.format(path))
    assert aa.add(AppriseAttachment.instantiate(
        'file://{}?name=andanother.png&cache=600'.format(path)))

    assert len(aa) == 3
    assert aa[0].cache is True
    assert aa[1].cache is False
    assert aa[2].cache == 600

    # Negative cache are not allowed
    assert not aa.add(AppriseAttachment.instantiate(
        'file://{}?name=andanother.png&cache=-600'.format(path)))

    # Invalid cache value
    assert not aa.add(AppriseAttachment.instantiate(
        'file://{}?name=andanother.png'.format(path), cache='invalid'))

    # No length change
    assert len(aa) == 3

    # Reset our object
    aa.clear()

    # Garbage in produces garbage out
    assert aa.add(None) is False
    assert aa.add(object()) is False
    assert aa.add(42) is False

    # length remains unchanged
    assert len(aa) == 0

    # We can add by lists as well in a variety of formats
    attachments = (
        None,
        object(),
        42,
        'garbage://',
    )

    # Add our attachments
    assert aa.add(attachments) is False

    # length remains unchanged
    assert len(aa) == 0

    # if instantiating attachments from the class, it will throw a TypeError
    # if attachments couldn't be loaded
    with pytest.raises(TypeError):
        AppriseAttachment('garbage://')

    # Load our other attachment types
    aa = AppriseAttachment(location=ContentLocation.LOCAL)

    # Hosted type won't allow us to import files
    aa = AppriseAttachment(location=ContentLocation.HOSTED)
    assert len(aa) == 0

    # Add our attachments defined a the head of this function
    aa.add(attachments)

    # Our length is still zero because we can't import files in
    # a hosted environment
    assert len(aa) == 0

    # Inaccessible type prevents the adding of new stuff
    aa = AppriseAttachment(location=ContentLocation.INACCESSIBLE)
    assert len(aa) == 0

    # Add our attachments defined a the head of this function
    aa.add(attachments)

    # Our length is still zero
    assert len(aa) == 0

    with pytest.raises(TypeError):
        # Invalid location specified
        AppriseAttachment(location="invalid")

    # test cases when file simply doesn't exist
    aa = AppriseAttachment('file://non-existant-file.png')
    # Our length is still 1
    assert len(aa) == 1
    # Our object will still return a True
    assert aa

    # However our indexed entry will not
    assert not aa[0]

    # length will return 0
    assert len(aa[0]) == 0

    # Total length will also return 0
    assert aa.size() == 0


@mock.patch('requests.get')
def test_apprise_attachment_truncate(mock_get):
    """
    API: AppriseAttachment when truncation in place

    """

    # Prepare our response
    response = requests.Request()
    response.status_code = requests.codes.ok

    # Prepare Mock
    mock_get.return_value = response

    # our Apprise Object
    ap_obj = Apprise()

    # Add ourselves an object set to truncate
    ap_obj.add('json://localhost/?method=GET&overflow=truncate')

    # Create ourselves an attachment object
    aa = AppriseAttachment()

    # There are no attachents loaded
    assert len(aa) == 0

    # Object can be directly checked as a boolean; response is False
    # when there are no entries loaded
    assert not aa

    # Add 2 attachments
    assert aa.add(join(TEST_VAR_DIR, 'apprise-test.gif'))
    assert aa.add(join(TEST_VAR_DIR, 'apprise-test.png'))

    assert mock_get.call_count == 0
    assert ap_obj.notify(body='body', title='title', attach=aa)

    assert mock_get.call_count == 1

    # Our first item was truncated, so only 1 attachment
    details = mock_get.call_args_list[0]
    dataset = json.loads(details[1]['data'])
    assert len(dataset['attachments']) == 1

    # Reset our object
    mock_get.reset_mock()

    # our Apprise Object
    ap_obj = Apprise()

    # Add ourselves an object set to upstream
    ap_obj.add('json://localhost/?method=GET&overflow=upstream')

    # Create ourselves an attachment object
    aa = AppriseAttachment()

    # Add 2 attachments
    assert aa.add(join(TEST_VAR_DIR, 'apprise-test.gif'))
    assert aa.add(join(TEST_VAR_DIR, 'apprise-test.png'))

    assert mock_get.call_count == 0
    assert ap_obj.notify(body='body', title='title', attach=aa)

    assert mock_get.call_count == 1

    # Our item was not truncated, so all attachments
    details = mock_get.call_args_list[0]
    dataset = json.loads(details[1]['data'])
    assert len(dataset['attachments']) == 2


def test_apprise_attachment_instantiate():
    """
    API: AppriseAttachment.instantiate()

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

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

    class BadAttachType(AttachBase):
        def __init__(self, **kwargs):
            super().__init__(**kwargs)

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

    # Store our bad attachment type in our schema map
    A_MGR['bad'] = BadAttachType

    with pytest.raises(TypeError):
        AppriseAttachment.instantiate(
            'bad://path', suppress_exceptions=False)

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


def test_attachment_matrix_dynamic_importing(tmpdir):
    """
    API: Apprise() Attachment Matrix Importing

    """

    # Make our new path valid
    suite = tmpdir.mkdir("apprise_attach_test_suite")
    suite.join("__init__.py").write('')

    module_name = 'badattach'

    # Create a base area to work within
    base = suite.mkdir(module_name)
    base.join("__init__.py").write('')

    # Test no app_id
    base.join('AttachBadFile1.py').write(cleandoc(
        """
        class AttachBadFile1:
            pass
        """))

    # No class of the same name
    base.join('AttachBadFile2.py').write(cleandoc(
        """
        class BadClassName:
            pass
        """))

    # Exception thrown
    base.join('AttachBadFile3.py').write("""raise ImportError()""")

    # Utilizes a schema:// already occupied (as string)
    base.join('AttachGoober.py').write(cleandoc(
        """
        from apprise import AttachBase
        class AttachGoober(AttachBase):
            # This class tests the fact we have a new class name, but we're
            # trying to over-ride items previously used

            # The default simple (insecure) protocol
            protocol = 'http'

            # The default secure protocol
            secure_protocol = 'https'
        """))

    # Utilizes a schema:// already occupied (as tuple)
    base.join('AttachBugger.py').write(cleandoc(
        """
        from apprise import AttachBase
        class AttachBugger(AttachBase):
            # This class tests the fact we have a new class name, but we're
            # trying to over-ride items previously used

            # The default simple (insecure) protocol
            protocol = ('http', 'bugger-test' )

            # The default secure protocol
            secure_protocol = ('https', 'bugger-tests')
        """))

    A_MGR.load_modules(path=str(base), name=module_name)