mirror of https://github.com/caronc/apprise
Adjust XMPP plugin fixes
Chris Caron
Andreas Motl
parent
3c89b0e4c7
commit
4e86b44fa7
|
|
@ -83,6 +83,9 @@ class NotifyXMPP(NotifyBase):
|
|||
# A URL that takes you to the setup/help of the specific protocol
|
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_xmpp'
|
||||
|
||||
# Lower throttle rate for XMPP
|
||||
request_rate_per_sec = 0.5
|
||||
|
||||
# The default XMPP port
|
||||
default_unsecure_port = 5222
|
||||
|
||||
|
|
@ -378,13 +381,15 @@ class NotifyXMPP(NotifyBase):
|
|||
return results
|
||||
|
||||
|
||||
class SleekXmppAdapter:
|
||||
class SleekXmppAdapter(object):
|
||||
"""
|
||||
Wrapper to SleekXmpp
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
host=None, port=None, secure=None, verify_certificate=None,
|
||||
xep=None, jid=None, password=None,
|
||||
body=None, targets=None, before_message=None,
|
||||
logger=None):
|
||||
xep=None, jid=None, password=None, body=None, targets=None,
|
||||
before_message=None, logger=None):
|
||||
|
||||
self.host = host
|
||||
self.port = port
|
||||
|
|
@ -462,9 +467,8 @@ class SleekXmppAdapter:
|
|||
# Establish connection to XMPP server.
|
||||
# To speed up sending messages, don't use the "reattempt" feature,
|
||||
# it will add a nasty delay even before connecting to XMPP server.
|
||||
use_ssl = self.port == NotifyXMPP.default_secure_port
|
||||
if not self.xmpp.connect((self.host, self.port),
|
||||
use_ssl=use_ssl, reattempt=False):
|
||||
use_ssl=self.secure, reattempt=False):
|
||||
return False
|
||||
|
||||
# Process XMPP communication.
|
||||
|
|
@ -473,27 +477,8 @@ class SleekXmppAdapter:
|
|||
return self.success
|
||||
|
||||
def session_start(self, event):
|
||||
|
||||
# [amo, 2020-03-19]
|
||||
# To speed up sending messages, let's skip presence signalling and
|
||||
# roster inquiry. I believe both XMPP features resonate more with
|
||||
# human users than bots.
|
||||
"""
|
||||
self.xmpp.send_presence()
|
||||
|
||||
try:
|
||||
self.xmpp.get_roster()
|
||||
|
||||
except sleekxmpp.exceptions.IqError as e:
|
||||
self.logger.warning('There was an error getting the XMPP roster.')
|
||||
self.logger.debug(e.iq['error']['condition'])
|
||||
self.xmpp.disconnect()
|
||||
return False
|
||||
|
||||
except sleekxmpp.exceptions.IqTimeout:
|
||||
self.logger.warning('XMPP Server is taking too long to respond.')
|
||||
self.xmpp.disconnect()
|
||||
return False
|
||||
Session Manager
|
||||
"""
|
||||
|
||||
targets = list(self.targets)
|
||||
|
|
|
|||
|
|
@ -6,3 +6,6 @@ pytest-cov
|
|||
tox
|
||||
babel
|
||||
cryptography
|
||||
|
||||
# Plugin Dependencies
|
||||
sleekxmpp
|
||||
|
|
|
|||
|
|
@ -24,100 +24,27 @@
|
|||
# THE SOFTWARE.
|
||||
|
||||
import six
|
||||
import mock
|
||||
import sys
|
||||
import ssl
|
||||
|
||||
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)
|
||||
|
||||
|
||||
# Mock the XMPP adapter to override "self.success".
|
||||
from apprise.plugins.NotifyXMPP import SleekXmppAdapter
|
||||
class MockedSleekXmppAdapter(SleekXmppAdapter):
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
SleekXmppAdapter.__init__(self, *args, **kwargs)
|
||||
self.success = True
|
||||
|
||||
|
||||
def test_xmpp_plugin(tmpdir):
|
||||
"""
|
||||
API: NotifyXMPP Plugin()
|
||||
"""
|
||||
|
||||
# Our module base
|
||||
sleekxmpp_name = 'sleekxmpp'
|
||||
class MockedSleekXmppAdapter(apprise.plugins.NotifyXMPP.SleekXmppAdapter):
|
||||
|
||||
# First we do an import without the sleekxmpp library available to ensure
|
||||
# we can handle cases when the library simply isn't available
|
||||
|
||||
if sleekxmpp_name in sys.modules:
|
||||
# Test cases where the sleekxmpp library exists; we want to remove it
|
||||
# for the purpose of testing and capture the handling of the
|
||||
# library when it is missing
|
||||
del sys.modules[sleekxmpp_name]
|
||||
reload(sys.modules['apprise.plugins.NotifyXMPP'])
|
||||
|
||||
# We need to fake our gnome environment for testing purposes since
|
||||
# the sleekxmpp library isn't available in Travis CI
|
||||
sys.modules[sleekxmpp_name] = mock.MagicMock()
|
||||
|
||||
xmpp = mock.Mock()
|
||||
xmpp.register_plugin.return_value = True
|
||||
xmpp.send_message.return_value = True
|
||||
xmpp.connect.return_value = True
|
||||
xmpp.disconnect.return_value = True
|
||||
xmpp.send_presence.return_value = True
|
||||
xmpp.get_roster.return_value = True
|
||||
xmpp.ssl_version = None
|
||||
|
||||
class IqError(Exception):
|
||||
iq = {'error': {'condition': 'test'}}
|
||||
pass
|
||||
|
||||
class IqTimeout(Exception):
|
||||
pass
|
||||
|
||||
# Setup our Exceptions
|
||||
sys.modules[sleekxmpp_name].exceptions.IqError = IqError
|
||||
sys.modules[sleekxmpp_name].exceptions.IqTimeout = IqTimeout
|
||||
|
||||
sys.modules[sleekxmpp_name].ClientXMPP.return_value = xmpp
|
||||
|
||||
# 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'])
|
||||
|
||||
NotifyXMPP = sys.modules['apprise.plugins.NotifyXMPP']
|
||||
|
||||
# An empty CA list
|
||||
NotifyXMPP.CA_CERTIFICATE_FILE_LOCATIONS = []
|
||||
|
||||
# Mock the XMPP adapter to override "self.success"
|
||||
NotifyXMPP.SleekXmppAdapter = MockedSleekXmppAdapter
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(MockedSleekXmppAdapter, self).__init__(*args, **kwargs)
|
||||
# Mock the XMPP adapter to override "self.success".
|
||||
self.success = True
|
||||
|
||||
# Disable Throttling to speed testing
|
||||
apprise.plugins.NotifyBase.request_rate_per_sec = 0
|
||||
|
|
@ -148,7 +75,7 @@ def test_xmpp_plugin(tmpdir):
|
|||
del ssl.PROTOCOL_TLS
|
||||
|
||||
# Test our URL
|
||||
url = 'xmpps://user:pass@example.com'
|
||||
url = 'xmpps://user:pass@localhost'
|
||||
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
|
||||
# Test we loaded
|
||||
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
|
||||
|
|
@ -163,7 +90,7 @@ def test_xmpp_plugin(tmpdir):
|
|||
# Handle case where it is not missing
|
||||
setattr(ssl, 'PROTOCOL_TLS', ssl.PROTOCOL_TLSv1)
|
||||
# Test our URL
|
||||
url = 'xmpps://user:pass@example.com'
|
||||
url = 'xmpps://user:pass@localhost'
|
||||
obj = apprise.Apprise.instantiate(url, suppress_exceptions=False)
|
||||
# Test we loaded
|
||||
assert isinstance(obj, apprise.plugins.NotifyXMPP) is True
|
||||
|
|
@ -176,31 +103,31 @@ def test_xmpp_plugin(tmpdir):
|
|||
|
||||
urls = (
|
||||
{
|
||||
'u': 'xmpps://user:pass@example.com',
|
||||
'p': 'xmpps://user:****@example.com',
|
||||
'u': 'xmpps://user:pass@localhost',
|
||||
'p': 'xmpps://user:****@localhost',
|
||||
}, {
|
||||
'u': 'xmpps://user:pass@example.com?'
|
||||
'u': 'xmpps://user:pass@localhost?'
|
||||
'xep=30,199,garbage,xep_99999999',
|
||||
'p': 'xmpps://user:****@example.com',
|
||||
'p': 'xmpps://user:****@localhost',
|
||||
}, {
|
||||
'u': 'xmpps://user:pass@example.com?xep=ignored',
|
||||
'p': 'xmpps://user:****@example.com',
|
||||
'u': 'xmpps://user:pass@localhost?xep=ignored',
|
||||
'p': 'xmpps://user:****@localhost',
|
||||
}, {
|
||||
'u': 'xmpps://pass@example.com/'
|
||||
'u': 'xmpps://pass@localhost/'
|
||||
'user@test.com, user2@test.com/resource',
|
||||
'p': 'xmpps://****@example.com',
|
||||
'p': 'xmpps://****@localhost',
|
||||
}, {
|
||||
'u': 'xmpps://pass@example.com:5226?jid=user@test.com',
|
||||
'p': 'xmpps://****@example.com:5226',
|
||||
'u': 'xmpps://pass@localhost:5226?jid=user@test.com',
|
||||
'p': 'xmpps://****@localhost:5226',
|
||||
}, {
|
||||
'u': 'xmpps://pass@example.com?jid=user@test.com&verify=False',
|
||||
'p': 'xmpps://****@example.com',
|
||||
'u': 'xmpps://pass@localhost?jid=user@test.com&verify=False',
|
||||
'p': 'xmpps://****@localhost',
|
||||
}, {
|
||||
'u': 'xmpps://user:pass@example.com?verify=False',
|
||||
'p': 'xmpps://user:****@example.com',
|
||||
'u': 'xmpps://user:pass@localhost?verify=False',
|
||||
'p': 'xmpps://user:****@localhost',
|
||||
}, {
|
||||
'u': 'xmpp://user:pass@example.com?to=user@test.com',
|
||||
'p': 'xmpp://user:****@example.com',
|
||||
'u': 'xmpp://user:pass@localhost?to=user@test.com',
|
||||
'p': 'xmpp://user:****@localhost',
|
||||
}
|
||||
)
|
||||
|
||||
|
|
@ -253,35 +180,9 @@ def test_xmpp_plugin(tmpdir):
|
|||
.CA_CERTIFICATE_FILE_LOCATIONS = [str(ca_cert), ]
|
||||
|
||||
obj = apprise.Apprise.instantiate(
|
||||
'xmpps://pass@example.com/user@test.com',
|
||||
'xmpps://pass@localhost/user@test.com',
|
||||
suppress_exceptions=False)
|
||||
|
||||
# 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
|
||||
|
||||
# Test Connect Failures
|
||||
xmpp.connect.return_value = False
|
||||
assert obj.notify(
|
||||
title='', body='body', notify_type=apprise.NotifyType.INFO) is False
|
||||
|
||||
# Return our object value so we don't obstruct other tests
|
||||
xmpp.connect.return_value = True
|
||||
|
||||
# Test Exceptions
|
||||
# These stopped working after starting to
|
||||
# honor sleekxmpp's asynchronous nature.
|
||||
"""
|
||||
xmpp.get_roster.side_effect = \
|
||||
sys.modules[sleekxmpp_name].exceptions.IqTimeout()
|
||||
|
||||
assert obj.notify(
|
||||
title='', body='body', notify_type=apprise.NotifyType.INFO) is False
|
||||
xmpp.get_roster.side_effect = None
|
||||
|
||||
xmpp.get_roster.side_effect = \
|
||||
sys.modules[sleekxmpp_name].exceptions.IqError()
|
||||
assert obj.notify(
|
||||
title='', body='body', notify_type=apprise.NotifyType.INFO) is False
|
||||
xmpp.get_roster.side_effect = None
|
||||
"""
|
||||
|
|
|
|||
Loading…
Reference in New Issue