github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity
apprise/test/test_xmpp_plugin.py

400 lines
14 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
#
# Copyright (C) 2020 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 os
import sys
import ssl
import mock
import pytest
import apprise
try:
# Python v3.4+
from importlib import reload
except ImportError:
try:
# Python v3.0-v3.3
from imp import reload
except ImportError:
# Python v2.7
pass
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
@pytest.mark.skipif(
'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
def test_xmpp_plugin_import_error(tmpdir):
"""
API: NotifyXMPP Plugin() Import Error
"""
# This is a really confusing test case; it can probably be done better,
# but this was all I could come up with. Effectively Apprise is will
# still work flawlessly without the sleekxmpp dependancy. Since
# sleekxmpp is actually required to be installed to run these unit tests
# we need to do some hacky tricks into fooling our test cases that the
# package isn't available.
# So we create a temporary directory called sleekxmpp (simulating the
# library itself) and writing an __init__.py in it that does nothing
# but throw an ImportError exception (simulating that the library
# isn't found).
suite = tmpdir.mkdir("sleekxmpp")
suite.join("__init__.py").write('')
module_name = 'sleekxmpp'
suite.join("{}.py".format(module_name)).write('raise ImportError()')
# The second part of the test is to update our PYTHON_PATH to look
# into this new directory first (before looking where the actual
# valid paths are). This will allow us to override 'JUST' the sleekxmpp
# path.
# Update our path to point to our new test suite
sys.path.insert(0, str(suite))
# We need to remove the sleekxmpp modules that have already been loaded
# in memory otherwise they'll just be used instead. Python is smart and
# won't go try and reload everything again if it doesn't have to.
for name in list(sys.modules.keys()):
if name.startswith('{}.'.format(module_name)):
del sys.modules[name]
del sys.modules[module_name]
del sys.modules['apprise.plugins.NotifyXMPP.SleekXmppAdapter']
# The following libraries need to be reloaded to prevent
# TypeError: super(type, obj): obj must be an instance or subtype of type
# This is better explained in this StackOverflow post:
# https://stackoverflow.com/questions/31363311/\
# any-way-to-manually-fix-operation-of-\
# super-after-ipython-reload-avoiding-ty
#
reload(sys.modules['apprise.plugins.NotifyXMPP'])
reload(sys.modules['apprise.plugins'])
reload(sys.modules['apprise.Apprise'])
reload(sys.modules['apprise'])
# This tests that Apprise still works without sleekxmpp.
# XMPP objects can still be instantiated in these cases.
obj = apprise.Apprise.instantiate('xmpp://user:pass@localhost')
assert obj is not None
# Tidy-up / restore things to how they were
# Remove our garbage library
os.unlink(str(suite.join("{}.py".format(module_name))))
# Remove our custom entry into the path
sys.path.remove(str(suite))
# Reload the libraries we care about
reload(sys.modules['apprise.plugins.NotifyXMPP.SleekXmppAdapter'])
reload(sys.modules['apprise.plugins.NotifyXMPP'])
reload(sys.modules['apprise.plugins'])
reload(sys.modules['apprise.Apprise'])
reload(sys.modules['apprise'])
@pytest.mark.skipif(
'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
def test_xmpp_plugin(tmpdir):
"""
API: NotifyXMPP Plugin()
"""
# Set success flag
apprise.plugins.SleekXmppAdapter.success = True
# Create a restore point
ca_backup = apprise.plugins.SleekXmppAdapter\
.CA_CERTIFICATE_FILE_LOCATIONS
# Clear CA Certificates
apprise.plugins.SleekXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = []
# Disable Throttling to speed testing
apprise.plugins.NotifyBase.request_rate_per_sec = 0
# Create our instance
obj = apprise.Apprise.instantiate('xmpp://', suppress_exceptions=False)
# Not possible because no password or host was specified
assert obj is None
with pytest.raises(TypeError):
apprise.Apprise.instantiate(
'xmpp://hostname', suppress_exceptions=False)
# SSL Flags
if hasattr(ssl, "PROTOCOL_TLS"):
# Test cases where PROTOCOL_TLS simply isn't available
ssl_temp_swap = ssl.PROTOCOL_TLS
del ssl.PROTOCOL_TLS
# Test our URL
url = 'xmpps://user:pass@127.0.0.1'
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
# Test we loaded
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
# Check that it found our mocked environments
assert obj._enabled is True
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
client_stream = mock.Mock()
client_stream.connect.return_value = True
mock_stream.return_value = client_stream
# We fail because we could not verify the host
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is False
# Restore the variable for remaining tests
setattr(ssl, 'PROTOCOL_TLS', ssl_temp_swap)
else:
# Handle case where it is not missing
setattr(ssl, 'PROTOCOL_TLS', ssl.PROTOCOL_TLSv1)
# Test our URL
url = 'xmpps://user:pass@localhost'
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
# Test we loaded
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
# Check that it found our mocked environments
assert obj._enabled is True
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
client_stream = mock.Mock()
client_stream.connect.return_value = True
mock_stream.return_value = client_stream
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True
# Restore settings as they were
del ssl.PROTOCOL_TLS
urls = (
{
'u': 'xmpp://user:pass@localhost',
'p': 'xmpp://user:****@localhost',
}, {
'u': 'xmpp://user:pass@localhost?'
'xep=30,199,garbage,xep_99999999',
'p': 'xmpp://user:****@localhost',
}, {
'u': 'xmpps://user:pass@localhost?xep=ignored&verify=no',
'p': 'xmpps://user:****@localhost',
}, {
'u': 'xmpps://pass@localhost/?verify=false'
'user@test.com, user2@test.com/resource',
'p': 'xmpps://****@localhost',
}, {
'u': 'xmpps://pass@localhost:5226?jid=user@test.com&verify=no',
'p': 'xmpps://****@localhost:5226',
}, {
'u': 'xmpps://pass@localhost?jid=user@test.com&verify=False',
'p': 'xmpps://****@localhost',
}, {
'u': 'xmpps://user:pass@localhost?verify=False',
'p': 'xmpps://user:****@localhost',
}, {
'u': 'xmpp://user:pass@localhost?to=user@test.com&verify=no',
'p': 'xmpp://user:****@localhost',
}
)
# Try Different Variations of our URL
for entry in urls:
url = entry['u']
privacy_url = entry['p']
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
# Test we loaded
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
# Check that it found our mocked environments
assert obj._enabled is True
# Test url() call
assert isinstance(obj.url(), six.string_types) is True
# Test url(privacy=True) call
assert isinstance(obj.url(privacy=True), six.string_types) is True
assert obj.url(privacy=True).startswith(privacy_url)
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
client_stream = mock.Mock()
client_stream.connect.return_value = True
mock_stream.return_value = client_stream
# test notifications
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True
# test notification without a title
assert obj.notify(
title='', body='body',
notify_type=apprise.NotifyType.INFO) is True
# Test Connection Failure
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
client_stream = mock.Mock()
client_stream.connect.return_value = False
mock_stream.return_value = client_stream
# test notifications
assert obj.notify(
title='title', body='body',
notify_type=apprise.NotifyType.INFO) is False
# Toggle our _enabled flag
obj._enabled = False
with mock.patch('sleekxmpp.ClientXMPP') as mock_client:
# Allow a connection to succeed
mock_client.connect.return_value = True
# Verify that we can't send content now
assert obj.notify(
title='', body='body',
notify_type=apprise.NotifyType.INFO) is False
# Toggle it back so it doesn't disrupt other testing
obj._enabled = True
# create an empty file for now
ca_cert = tmpdir.mkdir("apprise_xmpp_test").join('ca_cert')
ca_cert.write('')
# Update our path
apprise.plugins.SleekXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = \
[str(ca_cert), ]
obj = apprise.Apprise.instantiate(
'xmpps://pass@localhost/user@test.com?verify=yes',
suppress_exceptions=False)
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
with mock.patch('sleekxmpp.ClientXMPP') as mock_client:
# Allow a connection to succeed
mock_client.connect.return_value = True
# Our notification now should be able to get a ca_cert to reference
assert obj.notify(
title='', body='body', notify_type=apprise.NotifyType.INFO) is True
# Restore our CA Certificates from backup
apprise.plugins.SleekXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = \
ca_backup
@pytest.mark.skipif(
'sleekxmpp' not in sys.modules, reason="requires sleekxmpp")
def test_sleekxmpp_callbacks():
"""
API: NotifyXMPP Plugin() Sleekxmpp callback tests
The tests identified here just test the basic callbacks defined for
sleekxmpp. Emulating a full xmpp server in order to test this plugin
proved to be difficult so just here are some basic tests to make sure code
doesn't produce any exceptions. This is a perfect solution to get 100%
test coverage of the NotifyXMPP plugin, but it's better than nothing at
all.
"""
def dummy_before_message():
# Just a dummy function for testing purposes
return
kwargs = {
'host': 'localhost',
'port': 5555,
'secure': False,
'verify_certificate': False,
'xep': [
# xep_0030: Service Discovery
30,
# xep_0199: XMPP Ping
199,
],
'jid': 'user@localhost',
'password': 'secret!',
'body': 'my message to delivery!',
'targets': ['user2@localhost'],
'before_message': dummy_before_message,
'logger': None,
}
# Set success flag
apprise.plugins.SleekXmppAdapter.success = False
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
client_stream = mock.Mock()
client_stream.send_message.return_value = True
mock_stream.return_value = client_stream
adapter = apprise.plugins.SleekXmppAdapter(**kwargs)
assert isinstance(adapter, apprise.plugins.SleekXmppAdapter)
# Ensure we are initialized in a failure state; our return flag after
# we actually attempt to send the notification(s). This get's toggled
# to true only after a session_start() call is done successfully
assert adapter.success is False
adapter.session_start()
assert adapter.success is True
# Now we'll do a test with no one to notify
kwargs['targets'] = []
adapter = apprise.plugins.SleekXmppAdapter(**kwargs)
assert isinstance(adapter, apprise.plugins.SleekXmppAdapter)
# success flag should be back to a False state
assert adapter.success is False
with mock.patch('sleekxmpp.ClientXMPP') as mock_stream:
client_stream = mock.Mock()
client_stream.send_message.return_value = True
mock_stream.return_value = client_stream
adapter.session_start()
# success flag changes to True
assert adapter.success is True
# Restore our target, but set up invalid xep codes
kwargs['targets'] = ['user2@localhost']
kwargs['xep'] = [1, 999]
with pytest.raises(ValueError):
apprise.plugins.SleekXmppAdapter(**kwargs)
Reference in New Issue View Git Blame Copy Permalink