mirror of https://github.com/caronc/apprise
349 lines
11 KiB
Python
349 lines
11 KiB
Python
# BSD 2-Clause License
|
|
#
|
|
# Apprise - Push Notification Library.
|
|
# Copyright (c) 2025, 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.
|
|
|
|
# Disable logging for a cleaner testing output
|
|
import logging
|
|
import time
|
|
from typing import ClassVar
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from apprise import NotificationManager
|
|
from apprise.common import ConfigFormat
|
|
from apprise.config.http import ConfigHTTP
|
|
from apprise.plugins import NotifyBase
|
|
|
|
logging.disable(logging.CRITICAL)
|
|
|
|
# Grant access to our Notification Manager Singleton
|
|
N_MGR = NotificationManager()
|
|
|
|
# Some exception handling we'll use
|
|
REQUEST_EXCEPTIONS = (
|
|
requests.ConnectionError(0, "requests.ConnectionError() not handled"),
|
|
requests.RequestException(0, "requests.RequestException() not handled"),
|
|
requests.HTTPError(0, "requests.HTTPError() not handled"),
|
|
requests.ReadTimeout(0, "requests.ReadTimeout() not handled"),
|
|
requests.TooManyRedirects(0, "requests.TooManyRedirects() not handled"),
|
|
)
|
|
|
|
|
|
@mock.patch("requests.post")
|
|
def test_config_http(mock_post):
|
|
"""
|
|
API: ConfigHTTP() object
|
|
|
|
"""
|
|
|
|
# Define our good:// url
|
|
class GoodNotification(NotifyBase):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def notify(self, *args, **kwargs):
|
|
# Pretend everything is okay
|
|
return True
|
|
|
|
def url(self, **kwargs):
|
|
# Support url() function
|
|
return ""
|
|
|
|
# Store our good notification in our schema map
|
|
N_MGR["good"] = GoodNotification
|
|
|
|
# Our default content
|
|
default_content = """taga,tagb=good://server01"""
|
|
|
|
class DummyResponse:
|
|
"""A dummy response used to manage our object."""
|
|
|
|
status_code = requests.codes.ok
|
|
headers: ClassVar[dict[str, str]] = {
|
|
"Content-Length": str(len(default_content)),
|
|
"Content-Type": "text/plain",
|
|
}
|
|
|
|
text = default_content
|
|
|
|
# Pointer to file
|
|
ptr = None
|
|
|
|
def close(self):
|
|
return
|
|
|
|
def raise_for_status(self):
|
|
return
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args, **kwargs):
|
|
return
|
|
|
|
# Prepare Mock
|
|
dummy_response = DummyResponse()
|
|
mock_post.return_value = dummy_response
|
|
|
|
assert ConfigHTTP.parse_url("garbage://") is None
|
|
|
|
results = ConfigHTTP.parse_url("http://user:pass@localhost?+key=value")
|
|
assert isinstance(results, dict)
|
|
ch = ConfigHTTP(**results)
|
|
assert isinstance(ch.url(), str) is True
|
|
assert isinstance(ch.read(), str) 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(), str) is True
|
|
assert isinstance(ch.read(), str) is True
|
|
|
|
# one entry added
|
|
assert len(ch) == 1
|
|
|
|
# Clear all our mock counters
|
|
mock_post.reset_mock()
|
|
|
|
# Cache Handling; cache each request for 30 seconds
|
|
results = ConfigHTTP.parse_url("http://localhost:8080/path/?cache=30")
|
|
assert mock_post.call_count == 0
|
|
assert isinstance(ch.url(), str) is True
|
|
|
|
assert isinstance(results, dict)
|
|
ch = ConfigHTTP(**results)
|
|
assert mock_post.call_count == 0
|
|
|
|
assert isinstance(ch.url(), str) is True
|
|
assert mock_post.call_count == 0
|
|
|
|
assert isinstance(ch.read(), str) is True
|
|
assert mock_post.call_count == 1
|
|
|
|
# Clear all our mock counters
|
|
mock_post.reset_mock()
|
|
|
|
# Behind the scenes we haven't actually made a fetch yet. We can consider
|
|
# our content expired at this point
|
|
assert ch.expired() is True
|
|
|
|
# Test using boolean check; this will force a remote fetch
|
|
assert ch
|
|
|
|
# Now a call was made
|
|
assert mock_post.call_count == 1
|
|
mock_post.reset_mock()
|
|
|
|
# Our content hasn't expired yet (it's good for 30 seconds)
|
|
assert ch.expired() is False
|
|
assert len(ch) == 1
|
|
assert mock_post.call_count == 0
|
|
|
|
# Test using boolean check; we will re-use our cache and not
|
|
# make another remote request
|
|
mock_post.reset_mock()
|
|
assert ch
|
|
assert len(ch.servers()) == 1
|
|
assert len(ch) == 1
|
|
|
|
# No remote post has been made
|
|
assert mock_post.call_count == 0
|
|
|
|
with mock.patch("time.time", return_value=time.time() + 10):
|
|
# even with 10 seconds elapsed, no fetch will be made
|
|
assert ch.expired() is False
|
|
assert ch
|
|
assert len(ch.servers()) == 1
|
|
assert len(ch) == 1
|
|
|
|
# No remote post has been made
|
|
assert mock_post.call_count == 0
|
|
|
|
with mock.patch("time.time", return_value=time.time() + 31):
|
|
# but 30+ seconds from now is considered expired
|
|
assert ch.expired() is True
|
|
assert ch
|
|
assert len(ch.servers()) == 1
|
|
assert len(ch) == 1
|
|
|
|
# Our content would have been renewed with a single new fetch
|
|
assert mock_post.call_count == 1
|
|
|
|
# one entry added
|
|
assert len(ch) == 1
|
|
|
|
# Invalid cache
|
|
results = ConfigHTTP.parse_url("http://localhost:8080/path/?cache=False")
|
|
assert isinstance(results, dict)
|
|
assert isinstance(ch.url(), str) is True
|
|
|
|
results = ConfigHTTP.parse_url("http://localhost:8080/path/?cache=-10")
|
|
assert isinstance(results, dict)
|
|
with pytest.raises(TypeError):
|
|
ch = ConfigHTTP(**results)
|
|
|
|
results = ConfigHTTP.parse_url("http://user@localhost?format=text")
|
|
assert isinstance(results, dict)
|
|
ch = ConfigHTTP(**results)
|
|
assert isinstance(ch.url(), str) is True
|
|
assert isinstance(ch.read(), str) is True
|
|
|
|
# one entry added
|
|
assert len(ch) == 1
|
|
|
|
results = ConfigHTTP.parse_url("https://localhost")
|
|
assert isinstance(results, dict)
|
|
ch = ConfigHTTP(**results)
|
|
assert isinstance(ch.url(), str) is True
|
|
assert isinstance(ch.read(), str) is True
|
|
|
|
# one entry added
|
|
assert len(ch) == 1
|
|
|
|
# Testing of pop
|
|
ch = ConfigHTTP(**results)
|
|
|
|
ref = ch[0]
|
|
assert isinstance(ref, NotifyBase) is True
|
|
|
|
ref_popped = ch.pop(0)
|
|
assert isinstance(ref_popped, NotifyBase) is True
|
|
|
|
assert ref == ref_popped
|
|
|
|
assert len(ch) == 0
|
|
|
|
# reference to calls on initial reference
|
|
ch = ConfigHTTP(**results)
|
|
assert isinstance(ch.pop(0), NotifyBase) is True
|
|
|
|
ch = ConfigHTTP(**results)
|
|
assert isinstance(ch[0], NotifyBase) is True
|
|
# Second reference actually uses cache
|
|
assert isinstance(ch[0], NotifyBase) is True
|
|
|
|
ch = ConfigHTTP(**results)
|
|
# Itereator creation (nothing needed to assert here)
|
|
iter(ch)
|
|
# Second reference actually uses cache
|
|
iter(ch)
|
|
|
|
# Test a buffer size limit reach
|
|
ch.max_buffer_size = len(dummy_response.text)
|
|
assert isinstance(ch.read(), str) is True
|
|
|
|
# Test YAML detection
|
|
yaml_supported_types = (
|
|
"text/yaml",
|
|
"text/x-yaml",
|
|
"application/yaml",
|
|
"application/x-yaml",
|
|
)
|
|
|
|
for st in yaml_supported_types:
|
|
dummy_response.headers["Content-Type"] = st
|
|
ch.default_config_format = None
|
|
assert isinstance(ch.read(), str) 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_response.headers["Content-Type"] = st
|
|
ch.default_config_format = None
|
|
assert isinstance(ch.read(), str) 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_response.headers["Content-Type"] = st
|
|
ch.default_config_format = None
|
|
assert isinstance(ch.read(), str) is True
|
|
# Remains unchanged
|
|
assert ch.default_config_format is None
|
|
|
|
# When the entry is missing; we handle this too
|
|
del dummy_response.headers["Content-Type"]
|
|
ch.default_config_format = None
|
|
assert isinstance(ch.read(), str) is True
|
|
# Remains unchanged
|
|
assert ch.default_config_format is None
|
|
|
|
# Restore our content type object for lower tests
|
|
dummy_response.headers["Content-Type"] = "text/plain"
|
|
|
|
# Take a snapshot
|
|
max_buffer_size = ch.max_buffer_size
|
|
|
|
ch.max_buffer_size = len(dummy_response.text) - 1
|
|
assert ch.read() is None
|
|
|
|
# Restore buffer size count
|
|
ch.max_buffer_size = max_buffer_size
|
|
|
|
# Test erroneous Content-Length
|
|
# Our content is still within the limits, so we're okay
|
|
dummy_response.headers["Content-Length"] = "garbage"
|
|
|
|
assert isinstance(ch.read(), str) is True
|
|
|
|
dummy_response.headers["Content-Length"] = "None"
|
|
# Our content is still within the limits, so we're okay
|
|
assert isinstance(ch.read(), str) is True
|
|
|
|
# Handle cases where the content length is exactly at our limit
|
|
dummy_response.text = "a" * ch.max_buffer_size
|
|
# This is acceptable
|
|
assert isinstance(ch.read(), str) is True
|
|
|
|
# If we are over our limit though..
|
|
dummy_response.text = "b" * (ch.max_buffer_size + 1)
|
|
assert ch.read() is None
|
|
|
|
# Test an invalid return code
|
|
dummy_response.status_code = 400
|
|
assert ch.read() is None
|
|
ch.max_error_buffer_size = 0
|
|
assert ch.read() is None
|
|
|
|
# Exception handling
|
|
for _exception in REQUEST_EXCEPTIONS:
|
|
mock_post.side_effect = _exception
|
|
assert ch.read() is None
|
|
|
|
# Restore buffer size count
|
|
ch.max_buffer_size = max_buffer_size
|