# -*- coding: utf-8 -*- # # Copyright (C) 2020 Chris Caron # 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_sleekxmpp_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.SliXmppAdapter']) 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_sleekxmpp_plugin(tmpdir): """ API: NotifyXMPP Plugin() """ # Set success flag apprise.plugins.SleekXmppAdapter.success = True # Enforce Adapter apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SleekXmppAdapter # 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_sleekxmpp_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 # Enforce Adapter apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SleekXmppAdapter 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) @pytest.mark.skipif( 'slixmpp' not in sys.modules, reason="requires slixmpp") def test_slixmpp_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 slixmpp dependancy. Since # slixmpp 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 slixmpp (simulating the # library itself) and writing an __init__.py in it that does nothing # but throw an ImportError exception (simulating that hte library # isn't found). suite = tmpdir.mkdir("slixmpp") suite.join("__init__.py").write('') module_name = 'slixmpp' 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 slixmpp # path. # Update our path to point to our new test suite sys.path.insert(0, str(suite)) # We need to remove the slixmpp 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.SliXmppAdapter'] # 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 slixmpp. # XMPP objects can't be istantiated though. 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.SliXmppAdapter']) 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( 'slixmpp' not in sys.modules, reason="requires slixmpp") def test_slixmpp_plugin(tmpdir): """ API: NotifyXMPP Plugin() """ # Set success flag apprise.plugins.SliXmppAdapter.success = True # Enforce Adapter apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SliXmppAdapter # Create a restore point ca_backup = apprise.plugins.SliXmppAdapter\ .CA_CERTIFICATE_FILE_LOCATIONS # Clear CA Certificates apprise.plugins.SliXmppAdapter.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('slixmpp.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('slixmpp.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('slixmpp.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('slixmpp.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('slixmpp.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_slixmpp_test").join('ca_cert') ca_cert.write('') # Update our path apprise.plugins.SliXmppAdapter.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('slixmpp.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.SliXmppAdapter.CA_CERTIFICATE_FILE_LOCATIONS = \ ca_backup @pytest.mark.skipif( 'slixmpp' not in sys.modules, reason="requires slixmpp") def test_slixmpp_callbacks(): """ API: NotifyXMPP Plugin() Slixmpp callback tests The tests identified here just test the basic callbacks defined for slixmpp. 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.SliXmppAdapter.success = False # Enforce Adapter apprise.plugins.NotifyXMPP._adapter = apprise.plugins.SliXmppAdapter with mock.patch('slixmpp.ClientXMPP') as mock_stream: client_stream = mock.Mock() client_stream.send_message.return_value = True mock_stream.return_value = client_stream adapter = apprise.plugins.SliXmppAdapter(**kwargs) assert isinstance(adapter, apprise.plugins.SliXmppAdapter) # 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.SliXmppAdapter(**kwargs) assert isinstance(adapter, apprise.plugins.SliXmppAdapter) # success flag should be back to a False state assert adapter.success is False with mock.patch('slixmpp.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.SliXmppAdapter(**kwargs)