mirror of https://github.com/caronc/apprise
391 lines
13 KiB
Python
391 lines
13 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 socket
|
|
from unittest import mock
|
|
|
|
import apprise
|
|
from apprise.plugins.aprs import NotifyAprs
|
|
|
|
logging.disable(logging.CRITICAL)
|
|
|
|
|
|
@mock.patch("socket.create_connection")
|
|
def test_plugin_aprs_urls(mock_create_connection):
|
|
"""NotifyAprs() Apprise URLs."""
|
|
# A socket object
|
|
sobj = mock.Mock()
|
|
sobj.return_value = 1
|
|
sobj.getpeername.return_value = ("localhost", 1234)
|
|
sobj.socket_close.return_value = None
|
|
sobj.setblocking.return_value = True
|
|
sobj.recv.return_value = "ping\npong pong DF1JSL-15 verified pong".encode(
|
|
"latin-1"
|
|
)
|
|
sobj.sendall.return_value = True
|
|
sobj.settimeout.return_value = True
|
|
|
|
# Prepare Mock
|
|
mock_create_connection.return_value = sobj
|
|
|
|
# Test invalid URLs
|
|
assert apprise.Apprise.instantiate("aprs://") is None
|
|
assert apprise.Apprise.instantiate("aprs://:@/") is None
|
|
|
|
# No call-sign specified
|
|
assert apprise.Apprise.instantiate("aprs://DF1JSL-15:12345") is None
|
|
|
|
# Garbage
|
|
assert NotifyAprs.parse_url(None) is None
|
|
|
|
# Valid call-sign but no password
|
|
assert apprise.Apprise.instantiate("aprs://DF1JSL-15:@DF1ABC") is None
|
|
assert apprise.Apprise.instantiate("aprs://DF1JSL-15@DF1ABC") is None
|
|
# Password of -1 not supported
|
|
assert apprise.Apprise.instantiate("aprs://DF1JSL-15:-1@DF1ABC") is None
|
|
# Alpha Password not supported
|
|
assert apprise.Apprise.instantiate("aprs://DF1JSL-15:abcd@DF1ABC") is None
|
|
|
|
# Valid instances
|
|
instance = apprise.Apprise.instantiate("aprs://DF1JSL-15:12345@DF1ABC")
|
|
assert isinstance(instance, NotifyAprs)
|
|
assert instance.url(privacy=True).startswith(
|
|
"aprs://DF1JSL-15:****@D...C?"
|
|
)
|
|
assert instance.notify("test") is True
|
|
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?delay=3.0"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?delay=2"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?delay=-3.0"
|
|
)
|
|
assert instance is None
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?delay=40.0"
|
|
)
|
|
assert instance is None
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?delay=invalid"
|
|
)
|
|
assert instance is None
|
|
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC/DF1DEF"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
assert instance.url(privacy=True).startswith(
|
|
"aprs://DF1JSL-15:****@D...C/D...F?"
|
|
)
|
|
assert instance.notify("test") is True
|
|
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC-1/DF1ABC/DF1ABC-15"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
assert instance.url(privacy=True).startswith(
|
|
"aprs://DF1JSL-15:****@D...1/D...C/D...5?"
|
|
)
|
|
assert instance.notify("test") is True
|
|
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@?to=DF1ABC,DF1DEF"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
assert instance.url(privacy=True).startswith(
|
|
"aprs://DF1JSL-15:****@D...C/D...F?"
|
|
)
|
|
assert instance.notify("test") is True
|
|
|
|
# Test Locale settings
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?locale=EURO"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
assert instance.url(privacy=True).startswith(
|
|
"aprs://DF1JSL-15:****@D...C?"
|
|
)
|
|
# we used the default locale, so no setting
|
|
assert "locale=" not in instance.url(privacy=True)
|
|
assert instance.notify("test") is True
|
|
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?locale=NOAM"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
assert instance.url(privacy=True).startswith(
|
|
"aprs://DF1JSL-15:****@D...C?"
|
|
)
|
|
# locale is set in URL
|
|
assert "locale=NOAM" in instance.url(privacy=True)
|
|
assert instance.notify("test") is True
|
|
|
|
# Invalid locale
|
|
assert (
|
|
apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC?locale=invalid"
|
|
)
|
|
is None
|
|
)
|
|
|
|
# Invalid call signs
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@abcdefghi/a"
|
|
)
|
|
|
|
# We still instantiate
|
|
assert isinstance(instance, NotifyAprs)
|
|
|
|
# We still load our bad entries
|
|
assert instance.url(privacy=True).startswith(
|
|
"aprs://DF1JSL-15:****@A...I/A...A?"
|
|
)
|
|
|
|
# But with only bad entries, we have nothing to notify
|
|
assert instance.notify("test") is False
|
|
|
|
# Enforces a close
|
|
del instance
|
|
|
|
|
|
@mock.patch("socket.create_connection")
|
|
def test_plugin_aprs_edge_cases(mock_create_connection):
|
|
"""NotifyAprs() Edge Cases."""
|
|
|
|
# A socket object
|
|
sobj = mock.Mock()
|
|
sobj.return_value = 1
|
|
sobj.getpeername.return_value = ("localhost", 1234)
|
|
sobj.socket_close.return_value = None
|
|
sobj.setblocking.return_value = True
|
|
sobj.recv.return_value = "ping\npong pong DF1JSL-15 verified pong".encode(
|
|
"latin-1"
|
|
)
|
|
sobj.sendall.return_value = True
|
|
sobj.settimeout.return_value = True
|
|
|
|
# Prepare Mock
|
|
mock_create_connection.return_value = sobj
|
|
|
|
# Valid instances
|
|
instance = apprise.Apprise.instantiate(
|
|
"aprs://DF1JSL-15:12345@DF1ABC/DF1DEF"
|
|
)
|
|
assert isinstance(instance, NotifyAprs)
|
|
|
|
# our URL Identifier
|
|
assert isinstance(instance.url_id(), str)
|
|
|
|
# Objects read
|
|
assert len(instance) == 2
|
|
|
|
# Bad data
|
|
sobj.recv.return_value = "one line".encode("latin-1")
|
|
assert instance.notify(body="body", title="title") is False
|
|
sobj.recv.return_value = "\n\n\n".encode("latin-1")
|
|
assert instance.notify(body="body", title="title") is False
|
|
sobj.recv.return_value = "".encode("latin-1")
|
|
assert instance.notify(body="body", title="title") is False
|
|
sobj.recv.return_value = "\ndata".encode("latin-1")
|
|
assert instance.notify(body="body", title="title") is False
|
|
# Different Call-Sign then what we logged in as
|
|
sobj.recv.return_value = "ping\npong pong DF1JSL-14 verified, pong".encode(
|
|
"latin-1"
|
|
)
|
|
assert instance.notify(body="body", title="title") is False
|
|
# Unverified
|
|
sobj.recv.return_value = (
|
|
"ping\npong pong DF1JSL-15 unverified, pong".encode("latin-1")
|
|
)
|
|
assert instance.notify(body="body", title="title") is False
|
|
|
|
#
|
|
# Test Login edge cases
|
|
#
|
|
sobj.return_value = False
|
|
assert instance.aprsis_login() is False
|
|
sobj.return_value = 1
|
|
sobj.recv.return_value = "".encode("latin-1")
|
|
assert instance.aprsis_login() is False
|
|
sobj.recv.return_value = "ping\npong pong DF1JSL-15 verified pong".encode(
|
|
"latin-1"
|
|
)
|
|
|
|
#
|
|
# Test Socket Send Exceptions
|
|
#
|
|
sobj.sendall.return_value = None
|
|
sobj.sendall.side_effect = socket.gaierror("gaierror")
|
|
# No connection
|
|
assert instance.socket_send("data") is False
|
|
# Ensure we have a connection before calling socket_send()
|
|
assert instance.socket_open() is True
|
|
assert instance.socket_send("data") is False
|
|
sobj.sendall.side_effect = socket.timeout("timeout")
|
|
assert instance.socket_open() is True
|
|
assert instance.socket_send("data") is False
|
|
assert instance.socket_open() is True
|
|
sobj.sendall.side_effect = OSError("error")
|
|
assert instance.socket_send("data") is False
|
|
|
|
# Login is impacted by socket_send
|
|
sobj.return_value = 1
|
|
assert instance.socket_open() is True
|
|
assert instance.aprsis_login() is False
|
|
|
|
# Return some of our
|
|
sobj.sendall.side_effect = None
|
|
sobj.sendall.return_value = True
|
|
|
|
assert instance.socket_open() is True
|
|
sobj.close.return_value = None
|
|
sobj.close.side_effect = socket.gaierror("gaierror")
|
|
instance.socket_close()
|
|
sobj.close.side_effect = socket.timeout("timeout")
|
|
instance.socket_close()
|
|
sobj.close.side_effect = OSError("error")
|
|
instance.socket_close()
|
|
sobj.return_value = None
|
|
instance.socket_close()
|
|
# Socket isn't open; so we can't get content
|
|
assert instance.socket_receive(100) is False
|
|
sobj.close.side_effect = None
|
|
sobj.close.return_value = None
|
|
# Double close test
|
|
instance.socket_close()
|
|
|
|
sobj.return_value = 1
|
|
mock_create_connection.return_value = None
|
|
mock_create_connection.side_effect = socket.gaierror("gaierror")
|
|
assert instance.socket_open() is False
|
|
assert instance.notify("test") is False
|
|
mock_create_connection.side_effect = socket.timeout("timeout")
|
|
assert instance.socket_open() is False
|
|
assert instance.notify("test") is False
|
|
mock_create_connection.side_effect = OSError("error")
|
|
assert instance.socket_open() is False
|
|
assert instance.notify("test") is False
|
|
mock_create_connection.side_effect = ConnectionError("ConnectionError")
|
|
assert instance.socket_open() is False
|
|
assert instance.notify("test") is False
|
|
|
|
# Restore our good connection
|
|
mock_create_connection.return_value = sobj
|
|
mock_create_connection.side_effect = None
|
|
|
|
# Functionality has been restored
|
|
assert instance.socket_open() is True
|
|
|
|
# Now play with getpeername
|
|
sobj.getpeername.return_value = None
|
|
sobj.getpeername.side_effect = ValueError("getpeername ValueError")
|
|
assert instance.socket_open() is True
|
|
|
|
sobj.getpeername.return_value = ("localhost", 1234)
|
|
assert instance.socket_open() is True
|
|
# Test different receive settings
|
|
assert instance.socket_receive(0)
|
|
assert instance.socket_receive(-1)
|
|
assert instance.socket_receive(100)
|
|
|
|
sobj.recv.side_effect = socket.gaierror("gaierror")
|
|
assert instance.socket_open() is True
|
|
assert instance.socket_receive(100) is False
|
|
sobj.recv.side_effect = socket.timeout("timeout")
|
|
assert instance.socket_open() is True
|
|
assert instance.socket_receive(100) is False
|
|
sobj.recv.side_effect = OSError("error")
|
|
assert instance.socket_open() is True
|
|
assert instance.socket_receive(100) is False
|
|
|
|
# Restore
|
|
sobj.recv.side_effect = None
|
|
sobj.recv.return_value = "ping\npong pong DF1JSL-15 verified pong".encode(
|
|
"latin-1"
|
|
)
|
|
|
|
# Simulate a successful connection, but a failed notification
|
|
# To do this we need to have a login succeed, but the second call to send
|
|
# to fail
|
|
sobj.sendall.return_value = True
|
|
assert instance.notify("test") is True
|
|
|
|
sobj.sendall.return_value = None
|
|
sobj.sendall.side_effect = (True, socket.gaierror("gaierror"))
|
|
assert instance.notify("test") is False
|
|
|
|
sobj.sendall.return_value = True
|
|
sobj.sendall.side_effect = None
|
|
del sobj
|
|
|
|
|
|
def test_plugin_aprs_config_files():
|
|
"""NotifyAprs() Config File Cases."""
|
|
content = """
|
|
urls:
|
|
- aprs://DF1JSL-15:12345@DF1ABC":
|
|
- locale: NOAM
|
|
|
|
- aprs://DF1JSL-15:12345@DF1ABC:
|
|
- locale: SOAM
|
|
|
|
- aprs://DF1JSL-15:12345@DF1ABC:
|
|
- locale: EURO
|
|
|
|
- aprs://DF1JSL-15:12345@DF1ABC:
|
|
- locale: ASIA
|
|
|
|
- aprs://DF1JSL-15:12345@DF1ABC:
|
|
- locale: AUNZ
|
|
|
|
- aprs://DF1JSL-15:12345@DF1ABC:
|
|
- locale: ROTA
|
|
|
|
# This will fail to load because the locale is bad
|
|
- aprs://DF1JSL-15:12345@DF1ABC:
|
|
- locale: aprs_invalid
|
|
"""
|
|
|
|
# Create ourselves a config object
|
|
ac = apprise.AppriseConfig()
|
|
assert ac.add_config(content=content) is True
|
|
|
|
aobj = apprise.Apprise()
|
|
|
|
# Add our configuration
|
|
aobj.add(ac)
|
|
|
|
assert len(ac.servers()) == 6
|
|
assert len(aobj) == 6
|