mirror of
https://github.com/caronc/apprise.git
synced 2024-10-04 17:22:18 +02:00
375 lines
13 KiB
Python
375 lines
13 KiB
Python
# -*- 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.
|
|
|
|
from unittest import mock
|
|
import socket
|
|
import apprise
|
|
from apprise.plugins.aprs import NotifyAprs
|
|
|
|
# Disable logging for a cleaner testing output
|
|
import logging
|
|
|
|
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 = socket.error('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 = socket.error('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 = socket.error('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 = socket.error('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
|