diff --git a/Dockerfile.py310 b/Dockerfile.py310 index c9b13d8c32..c7992f6b7a 100644 --- a/Dockerfile.py310 +++ b/Dockerfile.py310 @@ -31,7 +31,7 @@ FROM python:3.10-buster RUN apt-get update && \ apt-get install -y --no-install-recommends libdbus-1-dev libgirepository1.0-dev build-essential musl-dev bash dbus && \ rm -rf /var/lib/apt/lists/* -RUN pip install --no-cache-dir dbus-python "PyGObject==3.44.2" +RUN pip install --no-cache-dir "PyGObject==3.44.2" # Apprise Setup VOLUME ["/apprise"] diff --git a/Dockerfile.py311 b/Dockerfile.py311 index 5363d25907..614ec28e70 100644 --- a/Dockerfile.py311 +++ b/Dockerfile.py311 @@ -31,7 +31,7 @@ FROM python:3.11-buster RUN apt-get update && \ apt-get install -y --no-install-recommends libdbus-1-dev libgirepository1.0-dev build-essential musl-dev bash dbus && \ rm -rf /var/lib/apt/lists/* -RUN pip install --no-cache-dir dbus-python "PyGObject==3.44.2" +RUN pip install --no-cache-dir "PyGObject==3.44.2" # Apprise Setup VOLUME ["/apprise"] diff --git a/apprise/plugins/NotifyDBus.py b/apprise/plugins/NotifyDBus.py index 9e7429a638..0f2a2322fb 100644 --- a/apprise/plugins/NotifyDBus.py +++ b/apprise/plugins/NotifyDBus.py @@ -27,11 +27,11 @@ # POSSIBILITY OF SUCH DAMAGE. import sys - -from ..AppriseLocale import gettext_lazy as _ -from ..common import NotifyImageSize, NotifyType -from ..utils import parse_bool from .NotifyBase import NotifyBase +from ..common import NotifyImageSize +from ..common import NotifyType +from ..utils import parse_bool +from ..AppriseLocale import gettext_lazy as _ # Default our global support flag NOTIFY_DBUS_SUPPORT_ENABLED = False @@ -39,16 +39,49 @@ # Image support is dependant on the GdkPixbuf library being available NOTIFY_DBUS_IMAGE_SUPPORT = False +# Initialize our mainloops +LOOP_GLIB = None +LOOP_QT = None + try: - # dbus essentials - import gi - gi.require_version("Gio", "2.0") - gi.require_version("GLib", "2.0") - from gi.repository import Gio, GLib + # D-Bus Message Bus Daemon 1.12.XX Essentials + from dbus import SessionBus + from dbus import Interface + from dbus import Byte + from dbus import ByteArray + from dbus import DBusException + + # + # now we try to determine which mainloop(s) we can access + # + + # glib/dbus + try: + from dbus.mainloop.glib import DBusGMainLoop + LOOP_GLIB = DBusGMainLoop() + + except ImportError: # pragma: no cover + # No problem + pass - # We're good - NOTIFY_DBUS_SUPPORT_ENABLED = True + # qt + try: + from dbus.mainloop.qt import DBusQtMainLoop + LOOP_QT = DBusQtMainLoop(set_as_default=True) + + except ImportError: + try: + from dbus.mainloop.pyqt5 import DBusQtMainLoop + LOOP_QT = DBusQtMainLoop(set_as_default=True) + + except ImportError: + # No problem + pass + + # We're good as long as at least one + NOTIFY_DBUS_SUPPORT_ENABLED = ( + LOOP_GLIB is not None or LOOP_QT is not None) # ImportError: When using gi.repository you must not import static modules # like "gobject". Please change all occurrences of "import gobject" to @@ -59,6 +92,7 @@ try: # The following is required for Image/Icon loading only + import gi gi.require_version('GdkPixbuf', '2.0') from gi.repository import GdkPixbuf NOTIFY_DBUS_IMAGE_SUPPORT = True @@ -76,6 +110,15 @@ # library available to us (or maybe one we don't support)? pass +# Define our supported protocols and the loop to assign them. +# The key to value pairs are the actual supported schema's matched +# up with the Main Loop they should reference when accessed. +MAINLOOP_MAP = { + 'qt': LOOP_QT, + 'kde': LOOP_QT, + 'dbus': LOOP_QT if LOOP_QT else LOOP_GLIB, +} + # Urgencies class DBusUrgency: @@ -116,11 +159,11 @@ class NotifyDBus(NotifyBase): """ # Set our global enabled flag - enabled = NOTIFY_DBUS_SUPPORT_ENABLED + enabled = bool(NOTIFY_DBUS_SUPPORT_ENABLED) requirements = { # Define our required packaging in order to work - 'details': _('libdbus-1.so.x or libdbus-2.so.x must be installed.') + 'details': _('libdbus-1.so.x, dbus-python and/or pyqt must be installed.') } # The default descriptive name associated with the Notification @@ -130,7 +173,12 @@ class NotifyDBus(NotifyBase): service_url = 'http://www.freedesktop.org/Software/dbus/' # The default protocols - protocol = ('glib', 'dbus') + # Python 3 keys() does not return a list object, it is its own dict_keys() + # object if we were to reference, we wouldn't be backwards compatible with + # Python v2. So converting the result set back into a list makes us + # compatible + # TODO: Review after dropping support for Python 2. + protocol = list(MAINLOOP_MAP.keys()) # A URL that takes you to the setup/help of the specific protocol setup_url = 'https://github.com/caronc/apprise/wiki/Notify_dbus' @@ -202,6 +250,15 @@ def __init__(self, urgency=None, x_axis=None, y_axis=None, # Track our notifications self.registry = {} + # Store our schema; default to dbus + self.schema = kwargs.get('schema', 'dbus') + + if self.schema not in MAINLOOP_MAP: + msg = 'The schema specified ({}) is not supported.' \ + .format(self.schema) + self.logger.warning(msg) + raise TypeError(msg) + # The urgency of the message self.urgency = int( NotifyDBus.template_args['urgency']['default'] @@ -234,19 +291,11 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): """ Perform DBus Notification """ - # Acquire our dbus interface + # Acquire our session try: - dbus_iface = Gio.DBusProxy.new_for_bus_sync( - Gio.BusType.SESSION, - Gio.DBusProxyFlags.NONE, - None, - self.dbus_interface, - self.dbus_setting_location, - self.dbus_interface, - None, - ) + session = SessionBus(mainloop=MAINLOOP_MAP[self.schema]) - except GLib.Error as e: + except DBusException as e: # Handle exception self.logger.warning('Failed to send DBus notification.') self.logger.debug(f'DBus Exception: {e}') @@ -258,19 +307,31 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): title = body body = '' + # acquire our dbus object + dbus_obj = session.get_object( + self.dbus_interface, + self.dbus_setting_location, + ) + + # Acquire our dbus interface + dbus_iface = Interface( + dbus_obj, + dbus_interface=self.dbus_interface, + ) + # image path icon_path = None if not self.include_image \ else self.image_path(notify_type, extension='.ico') # Our meta payload meta_payload = { - "urgency": GLib.Variant("y", self.urgency), + "urgency": Byte(self.urgency) } if not (self.x_axis is None and self.y_axis is None): # Set x/y access if these were set - meta_payload['x'] = GLib.Variant("i", self.x_axis) - meta_payload['y'] = GLib.Variant("i", self.y_axis) + meta_payload['x'] = self.x_axis + meta_payload['y'] = self.y_axis if NOTIFY_DBUS_IMAGE_SUPPORT and icon_path: try: @@ -278,17 +339,14 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): image = GdkPixbuf.Pixbuf.new_from_file(icon_path) # Associate our image to our notification - meta_payload['icon_data'] = GLib.Variant( - "(iiibiiay)", - ( - image.get_width(), - image.get_height(), - image.get_rowstride(), - image.get_has_alpha(), - image.get_bits_per_sample(), - image.get_n_channels(), - image.get_pixels(), - ), + meta_payload['icon_data'] = ( + image.get_width(), + image.get_height(), + image.get_rowstride(), + image.get_has_alpha(), + image.get_bits_per_sample(), + image.get_n_channels(), + ByteArray(image.get_pixels()) ) except Exception as e: @@ -301,7 +359,6 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): self.throttle() dbus_iface.Notify( - "(susssasa{sv}i)", # Application Identifier self.app_id, # Message ID (0 = New Message) @@ -354,13 +411,15 @@ def url(self, privacy=False, *args, **kwargs): if self.y_axis: params['y'] = str(self.y_axis) - schema = self.protocol[0] - return f'{schema}://_/?{NotifyDBus.urlencode(params)}' + return '{schema}://_/?{params}'.format( + schema=self.schema, + params=NotifyDBus.urlencode(params), + ) @staticmethod def parse_url(url): """ - There are no parameters necessary for this protocol; simply having + There are no parameters nessisary for this protocol; simply having gnome:// is all you need. This function just makes sure that is in place. diff --git a/apprise/plugins/NotifyQT.py b/apprise/plugins/NotifyGLib.py similarity index 73% rename from apprise/plugins/NotifyQT.py rename to apprise/plugins/NotifyGLib.py index bba5c3ba5a..7c8b8b6926 100644 --- a/apprise/plugins/NotifyQT.py +++ b/apprise/plugins/NotifyGLib.py @@ -27,39 +27,28 @@ # POSSIBILITY OF SUCH DAMAGE. import sys -from .NotifyBase import NotifyBase -from ..common import NotifyImageSize -from ..common import NotifyType -from ..utils import parse_bool + from ..AppriseLocale import gettext_lazy as _ +from ..common import NotifyImageSize, NotifyType +from ..utils import parse_bool +from .NotifyBase import NotifyBase + +# Default our global support flag +NOTIFY_GLIB_SUPPORT_ENABLED = False # Image support is dependant on the GdkPixbuf library being available -NOTIFY_QT_IMAGE_SUPPORT = False +NOTIFY_GLIB_IMAGE_SUPPORT = False -# Initialize our mainloops -LOOP_QT = None try: - # D-Bus Message Bus Daemon 1.12.XX Essentials - from dbus import SessionBus - from dbus import Interface - from dbus import Byte - from dbus import ByteArray - from dbus import DBusException - - # QT - try: - from dbus.mainloop.qt import DBusQtMainLoop - LOOP_QT = DBusQtMainLoop(set_as_default=True) + # glib essentials + import gi + gi.require_version("Gio", "2.0") + gi.require_version("GLib", "2.0") + from gi.repository import Gio, GLib - except ImportError: - try: - from dbus.mainloop.pyqt5 import DBusQtMainLoop - LOOP_QT = DBusQtMainLoop(set_as_default=True) - - except ImportError: - # No problem - pass + # We're good + NOTIFY_GLIB_SUPPORT_ENABLED = True # ImportError: When using gi.repository you must not import static modules # like "gobject". Please change all occurrences of "import gobject" to @@ -70,10 +59,9 @@ try: # The following is required for Image/Icon loading only - import gi gi.require_version('GdkPixbuf', '2.0') from gi.repository import GdkPixbuf - NOTIFY_QT_IMAGE_SUPPORT = True + NOTIFY_GLIB_IMAGE_SUPPORT = True except (ImportError, ValueError, AttributeError): # No problem; this will get caught in outer try/catch @@ -88,10 +76,6 @@ # library available to us (or maybe one we don't support)? pass -# Define our supported protocols and the loop to assign them. -# The key to value pairs are the actual supported schema's matched -# up with the Main Loop they should reference when accessed. - # Urgencies class DBusUrgency: @@ -126,30 +110,31 @@ class DBusUrgency: } -class NotifyQT(NotifyBase): +class NotifyGLib(NotifyBase): """ - A wrapper for local QT Notifications + A wrapper for local GLib/Gio Notifications """ # Set our global enabled flag - enabled = LOOP_QT is not None + enabled = NOTIFY_GLIB_SUPPORT_ENABLED requirements = { # Define our required packaging in order to work - 'details': _('libdbus-1.so.x and pyqt must be installed.') + 'details': _('libdbus-1.so.x or libdbus-2.so.x must be installed.') } # The default descriptive name associated with the Notification service_name = _('DBus Notification') # The services URL - service_url = 'http://www.freedesktop.org/Software/dbus/' + service_url = \ + 'https://lazka.github.io/pgi-docs/Gio-2.0/classes/DBusProxy.html' # The default protocols - protocol = ('qt', 'kde') + protocol = ('glib', 'gio') # A URL that takes you to the setup/help of the specific protocol - setup_url = 'https://github.com/caronc/apprise/wiki/Notify_dbus' + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_glib' # No throttling required for DBus queries request_rate_per_sec = 0 @@ -165,8 +150,8 @@ class NotifyQT(NotifyBase): body_max_line_count = 10 # The following are required to hook into the notifications: - qt_interface = 'org.freedesktop.Notifications' - qt_setting_location = '/org/freedesktop/Notifications' + glib_interface = 'org.freedesktop.Notifications' + glib_setting_location = '/org/freedesktop/Notifications' # Define object templates templates = ( @@ -220,12 +205,12 @@ def __init__(self, urgency=None, x_axis=None, y_axis=None, # The urgency of the message self.urgency = int( - NotifyQT.template_args['urgency']['default'] + NotifyGLib.template_args['urgency']['default'] if urgency is None else next(( v for k, v in DBUS_URGENCY_MAP.items() if str(urgency).lower().startswith(k)), - NotifyQT.template_args['urgency']['default'])) + NotifyGLib.template_args['urgency']['default'])) # Our x/y axis settings if x_axis or y_axis: @@ -248,16 +233,24 @@ def __init__(self, urgency=None, x_axis=None, y_axis=None, def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): """ - Perform QT Notification + Perform GLib/Gio Notification """ - # Acquire our session + # Acquire our gio interface try: - session = SessionBus(mainloop=LOOP_QT) + gio_iface = Gio.DBusProxy.new_for_bus_sync( + Gio.BusType.SESSION, + Gio.DBusProxyFlags.NONE, + None, + self.glib_interface, + self.glib_setting_location, + self.glib_interface, + None, + ) - except DBusException as e: + except GLib.Error as e: # Handle exception - self.logger.warning('Failed to send QT notification.') - self.logger.debug(f'QT Exception: {e}') + self.logger.warning('Failed to send GLib/Gio notification.') + self.logger.debug(f'GLib/Gio Exception: {e}') return False # If there is no title, but there is a body, swap the two to get rid @@ -266,58 +259,50 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): title = body body = '' - # acquire our qt/dbus object - qt_obj = session.get_object( - self.qt_interface, - self.qt_setting_location, - ) - - # Acquire our qt/dbus interface - qt_iface = Interface( - qt_obj, - qt_interface=self.qt_interface, - ) - # image path icon_path = None if not self.include_image \ else self.image_path(notify_type, extension='.ico') # Our meta payload meta_payload = { - "urgency": Byte(self.urgency) + "urgency": GLib.Variant("y", self.urgency), } if not (self.x_axis is None and self.y_axis is None): # Set x/y access if these were set - meta_payload['x'] = self.x_axis - meta_payload['y'] = self.y_axis + meta_payload['x'] = GLib.Variant("i", self.x_axis) + meta_payload['y'] = GLib.Variant("i", self.y_axis) - if NOTIFY_QT_IMAGE_SUPPORT and icon_path: + if NOTIFY_GLIB_IMAGE_SUPPORT and icon_path: try: # Use Pixbuf to create the proper image type image = GdkPixbuf.Pixbuf.new_from_file(icon_path) # Associate our image to our notification - meta_payload['icon_data'] = ( - image.get_width(), - image.get_height(), - image.get_rowstride(), - image.get_has_alpha(), - image.get_bits_per_sample(), - image.get_n_channels(), - ByteArray(image.get_pixels()) + meta_payload['icon_data'] = GLib.Variant( + "(iiibiiay)", + ( + image.get_width(), + image.get_height(), + image.get_rowstride(), + image.get_has_alpha(), + image.get_bits_per_sample(), + image.get_n_channels(), + image.get_pixels(), + ), ) except Exception as e: self.logger.warning( "Could not load notification icon (%s).", icon_path) - self.logger.debug(f'QT Exception: {e}') + self.logger.debug(f'GLib/Gio Exception: {e}') try: # Always call throttle() before any remote execution is made self.throttle() - qt_iface.Notify( + gio_iface.Notify( + "(susssasa{sv}i)", # Application Identifier self.app_id, # Message ID (0 = New Message) @@ -336,11 +321,11 @@ def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): self.message_timeout_ms, ) - self.logger.info('Sent QT notification.') + self.logger.info('Sent GLib/Gio notification.') except Exception as e: - self.logger.warning('Failed to send QT notification.') - self.logger.debug(f'QT Exception: {e}') + self.logger.warning('Failed to send GLib/Gio notification.') + self.logger.debug(f'GLib/Gio Exception: {e}') return False return True @@ -370,15 +355,13 @@ def url(self, privacy=False, *args, **kwargs): if self.y_axis: params['y'] = str(self.y_axis) - return '{schema}://_/?{params}'.format( - schema=self.protocol[0], - params=NotifyQT.urlencode(params), - ) + schema = self.protocol[0] + return f'{schema}://_/?{NotifyGLib.urlencode(params)}' @staticmethod def parse_url(url): """ - There are no parameters nessisary for this protocol; simply having + There are no parameters necessary for this protocol; simply having gnome:// is all you need. This function just makes sure that is in place. @@ -390,22 +373,22 @@ def parse_url(url): results['include_image'] = \ parse_bool(results['qsd'].get('image', True)) - # QT supports urgency, but we we also support the keyword priority - # so that it is consistent with some of the other plugins + # GLib/Gio supports urgency, but we we also support the keyword + # priority so that it is consistent with some of the other plugins if 'priority' in results['qsd'] and len(results['qsd']['priority']): # We intentionally store the priority in the urgency section results['urgency'] = \ - NotifyQT.unquote(results['qsd']['priority']) + NotifyGLib.unquote(results['qsd']['priority']) if 'urgency' in results['qsd'] and len(results['qsd']['urgency']): results['urgency'] = \ - NotifyQT.unquote(results['qsd']['urgency']) + NotifyGLib.unquote(results['qsd']['urgency']) # handle x,y coordinates if 'x' in results['qsd'] and len(results['qsd']['x']): - results['x_axis'] = NotifyQT.unquote(results['qsd'].get('x')) + results['x_axis'] = NotifyGLib.unquote(results['qsd'].get('x')) if 'y' in results['qsd'] and len(results['qsd']['y']): - results['y_axis'] = NotifyQT.unquote(results['qsd'].get('y')) + results['y_axis'] = NotifyGLib.unquote(results['qsd'].get('y')) return results diff --git a/test/test_plugin_dbus.py b/test/test_plugin_dbus.py new file mode 100644 index 0000000000..0932aa2868 --- /dev/null +++ b/test/test_plugin_dbus.py @@ -0,0 +1,514 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2023, Chris Caron +# +# 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. + +import importlib +import logging +import re +import sys +import types +from unittest.mock import Mock, call, ANY + +import pytest + +import apprise +from helpers import reload_plugin + + +# Disable logging for a cleaner testing output +logging.disable(logging.CRITICAL) + +try: + import dbus + +except ImportError: + pass + +# Skip tests when Python environment does not provide the `dbus` package. +if 'dbus' not in sys.modules: + pytest.skip("Skipping dbus-python based tests", allow_module_level=True) + + +from dbus import DBusException # noqa E402 +from apprise.plugins.NotifyDBus import DBusUrgency, NotifyDBus # noqa E402 + + +def setup_dbus_environment(): + """ + Setup a heavily mocked Glib environment. + """ + mock_mainloop = Mock() + + # Our module base + gi_name = 'gi' + + # First we do an import without the gi library available to ensure + # we can handle cases when the library simply isn't available + + if gi_name in sys.modules: + # Test cases where the gi 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[gi_name] + importlib.reload(sys.modules['apprise.plugins.NotifyDBus']) + + # We need to fake our dbus environment for testing purposes since + # the gi library isn't available on CI + gi = types.ModuleType(gi_name) + gi.repository = types.ModuleType(gi_name + '.repository') + + mock_pixbuf = Mock() + mock_image = Mock() + mock_pixbuf.new_from_file.return_value = mock_image + + mock_image.get_width.return_value = 100 + mock_image.get_height.return_value = 100 + mock_image.get_rowstride.return_value = 1 + mock_image.get_has_alpha.return_value = 0 + mock_image.get_bits_per_sample.return_value = 8 + mock_image.get_n_channels.return_value = 1 + mock_image.get_pixels.return_value = '' + + gi.repository.GdkPixbuf = \ + types.ModuleType(gi_name + '.repository.GdkPixbuf') + gi.repository.GdkPixbuf.Pixbuf = mock_pixbuf + + # Emulate require_version function: + gi.require_version = Mock( + name=gi_name + '.require_version') + + # Force the fake module to exist + sys.modules[gi_name] = gi + sys.modules[gi_name + '.repository'] = gi.repository + + # Exception Handling + mock_mainloop.qt.DBusQtMainLoop.return_value = True + mock_mainloop.qt.DBusQtMainLoop.side_effect = ImportError + sys.modules['dbus.mainloop.qt'] = mock_mainloop.qt + mock_mainloop.qt.DBusQtMainLoop.side_effect = None + + mock_mainloop.glib.NativeMainLoop.return_value = True + mock_mainloop.glib.NativeMainLoop.side_effect = ImportError() + sys.modules['dbus.mainloop.glib'] = mock_mainloop.glib + mock_mainloop.glib.DBusGMainLoop.side_effect = None + mock_mainloop.glib.NativeMainLoop.side_effect = None + + # When patching something which has a side effect on the module-level code + # of a plugin, make sure to reload it. + reload_plugin('NotifyDBus') + + +@pytest.fixture +def dbus_environment(mocker): + """ + Fixture to provide a mocked Dbus environment to test case functions. + """ + interface_mock = mocker.patch('dbus.Interface', spec=True, + Notify=Mock()) + mocker.patch('dbus.SessionBus', spec=True, + **{"get_object.return_value": interface_mock}) + + +@pytest.fixture +def glib_environment(): + """ + Fixture to provide a mocked DBus environment to test case functions. + """ + setup_dbus_environment() + + +@pytest.fixture +def dbus_glib_environment(dbus_environment, glib_environment): + """ + Fixture to provide a mocked Glib/DBus environment to test case functions. + """ + pass + + +def test_plugin_dbus_general_success(mocker, dbus_glib_environment): + """ + NotifyDBus() general tests + + Test class loading using different arguments, provided via URL. + """ + + # Create our instance (identify all supported types) + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.url().startswith('dbus://_/') + obj = apprise.Apprise.instantiate('kde://', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.url().startswith('kde://_/') + obj = apprise.Apprise.instantiate('qt://', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.url().startswith('qt://_/') + obj.duration = 0 + + # Set our X and Y coordinate and try the notification + assert NotifyDBus( + x_axis=0, y_axis=0, **{'schema': 'dbus'})\ + .notify(title='', body='body', + notify_type=apprise.NotifyType.INFO) is True + + # 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 our arguments through the instantiate call + obj = apprise.Apprise.instantiate( + 'dbus://_/?image=True', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.url().startswith('dbus://_/') + assert re.search('image=yes', obj.url()) + + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + obj = apprise.Apprise.instantiate( + 'dbus://_/?image=False', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.url().startswith('dbus://_/') + assert re.search('image=no', obj.url()) + + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + # Test priority (alias to urgency) handling + obj = apprise.Apprise.instantiate( + 'dbus://_/?priority=invalid', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + obj = apprise.Apprise.instantiate( + 'dbus://_/?priority=high', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + obj = apprise.Apprise.instantiate( + 'dbus://_/?priority=2', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + # Test urgency handling + obj = apprise.Apprise.instantiate( + 'dbus://_/?urgency=invalid', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + obj = apprise.Apprise.instantiate( + 'dbus://_/?urgency=high', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + obj = apprise.Apprise.instantiate( + 'dbus://_/?urgency=2', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + obj = apprise.Apprise.instantiate( + 'dbus://_/?urgency=', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + # Test x/y + obj = apprise.Apprise.instantiate( + 'dbus://_/?x=5&y=5', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj.url(), str) is True + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + +def test_plugin_dbus_general_failure(dbus_glib_environment): + """ + Verify a few failure conditions. + """ + + with pytest.raises(TypeError): + apprise.Apprise.instantiate('dbus://_/?x=invalid&y=invalid', + suppress_exceptions=False) + + +def test_plugin_dbus_parse_configuration(dbus_glib_environment): + + # Test configuration parsing + content = """ + urls: + - dbus://: + - priority: 0 + tag: dbus_int low + - priority: "0" + tag: dbus_str_int low + - priority: low + tag: dbus_str low + - urgency: 0 + tag: dbus_int low + - urgency: "0" + tag: dbus_str_int low + - urgency: low + tag: dbus_str low + + # These will take on normal (default) urgency + - priority: invalid + tag: dbus_invalid + - urgency: invalid + tag: dbus_invalid + + - dbus://: + - priority: 2 + tag: dbus_int high + - priority: "2" + tag: dbus_str_int high + - priority: high + tag: dbus_str high + - urgency: 2 + tag: dbus_int high + - urgency: "2" + tag: dbus_str_int high + - urgency: high + tag: dbus_str high + """ + + # 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) + + # We should be able to read our 14 servers from that + # 6x low + # 6x high + # 2x invalid (so takes on normal urgency) + assert len(ac.servers()) == 14 + assert len(aobj) == 14 + assert len([x for x in aobj.find(tag='low')]) == 6 + for s in aobj.find(tag='low'): + assert s.urgency == DBusUrgency.LOW + + assert len([x for x in aobj.find(tag='high')]) == 6 + for s in aobj.find(tag='high'): + assert s.urgency == DBusUrgency.HIGH + + assert len([x for x in aobj.find(tag='dbus_str')]) == 4 + assert len([x for x in aobj.find(tag='dbus_str_int')]) == 4 + assert len([x for x in aobj.find(tag='dbus_int')]) == 4 + + assert len([x for x in aobj.find(tag='dbus_invalid')]) == 2 + for s in aobj.find(tag='dbus_invalid'): + assert s.urgency == DBusUrgency.NORMAL + + +def test_plugin_dbus_missing_icon(mocker, dbus_glib_environment): + """ + Test exception when loading icon; the notification will still be sent. + """ + + # Inject error when loading icon. + gi = importlib.import_module("gi") + gi.repository.GdkPixbuf.Pixbuf.new_from_file.side_effect = \ + AttributeError("Something failed") + + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + logger: Mock = mocker.spy(obj, "logger") + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + assert logger.mock_calls == [ + call.warning('Could not load notification icon (%s).', ANY), + call.debug('DBus Exception: Something failed'), + call.info('Sent DBus notification.'), + ] + + +def test_plugin_dbus_disabled_plugin(dbus_glib_environment): + """ + Verify notification will not be submitted if plugin is disabled. + """ + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + + obj.enabled = False + + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is False + + +def test_plugin_dbus_set_urgency(): + """ + Test the setting of an urgency. + """ + NotifyDBus(urgency=0) + + +def test_plugin_dbus_gi_missing(dbus_glib_environment): + """ + Verify notification succeeds even if the `gi` package is not available. + """ + + # Make `require_version` function raise an ImportError. + gi = importlib.import_module("gi") + gi.require_version.side_effect = ImportError() + + # When patching something which has a side effect on the module-level code + # of a plugin, make sure to reload it. + reload_plugin('NotifyDBus') + + # Create the instance. + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + obj.duration = 0 + + # Test url() call. + assert isinstance(obj.url(), str) is True + + # The notification succeeds even though the gi library was not loaded. + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + +def test_plugin_dbus_gi_require_version_error(dbus_glib_environment): + """ + Verify notification succeeds even if `gi.require_version()` croaks. + """ + + # Make `require_version` function raise a ValueError. + gi = importlib.import_module("gi") + gi.require_version.side_effect = ValueError("Something failed") + + # When patching something which has a side effect on the module-level code + # of a plugin, make sure to reload it. + reload_plugin('NotifyDBus') + + # Create instance. + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + obj.duration = 0 + + # Test url() call. + assert isinstance(obj.url(), str) is True + + # The notification succeeds even though the gi library was not loaded. + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is True + + +def test_plugin_dbus_module_croaks(mocker, dbus_glib_environment): + """ + Verify plugin is not available when `dbus` module is missing. + """ + + # Make importing `dbus` raise an ImportError. + mocker.patch.dict( + sys.modules, {'dbus': compile('raise ImportError()', 'dbus', 'exec')}) + + # When patching something which has a side effect on the module-level code + # of a plugin, make sure to reload it. + reload_plugin('NotifyDBus') + + # Verify plugin is not available. + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + assert obj is None + + +def test_plugin_dbus_session_croaks(mocker, dbus_glib_environment): + """ + Verify notification fails if DBus croaks. + """ + + mocker.patch('dbus.SessionBus', side_effect=DBusException('test')) + setup_dbus_environment() + + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + + # Emulate DBus session initialization error. + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is False + + +def test_plugin_dbus_interface_notify_croaks(mocker): + """ + Fail gracefully if underlying object croaks for whatever reason. + """ + + # Inject an error when invoking `dbus.Interface().Notify()`. + mocker.patch('dbus.SessionBus', spec=True) + mocker.patch('dbus.Interface', spec=True, + Notify=Mock(side_effect=AttributeError("Something failed"))) + setup_dbus_environment() + + obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) is True + + logger: Mock = mocker.spy(obj, "logger") + assert obj.notify( + title='title', body='body', + notify_type=apprise.NotifyType.INFO) is False + assert [ + call.warning('Failed to send DBus notification.'), + call.debug('DBus Exception: Something failed'), + ] in logger.mock_calls diff --git a/test/test_plugin_glib.py b/test/test_plugin_glib.py index aed4871114..a308721a2f 100644 --- a/test/test_plugin_glib.py +++ b/test/test_plugin_glib.py @@ -38,20 +38,22 @@ import apprise from helpers import reload_plugin +try: + from gi.repository import Gio # noqa F401 + +except ImportError: + pass # Disable logging for a cleaner testing output logging.disable(logging.CRITICAL) # Skip tests when Python environment does not provide the `gio` package. if 'gi.repository.Gio' not in sys.modules: - pytest.skip("Skipping dbus-python based tests", allow_module_level=True) + pytest.skip("Skipping glib based tests", allow_module_level=True) - -import gi # noqa E402 -gi.require_version("GLib", "2.0") from gi.repository import GLib # noqa E402 -from apprise.plugins.NotifyDBus import DBusUrgency, NotifyDBus # noqa E402 +from apprise.plugins.NotifyGLib import DBusUrgency, NotifyGLib # noqa E402 def setup_glib_environment(): @@ -70,9 +72,9 @@ def setup_glib_environment(): # for the purpose of testing and capture the handling of the # library when it is missing del sys.modules[gi_name] - importlib.reload(sys.modules['apprise.plugins.NotifyDBus']) + importlib.reload(sys.modules['apprise.plugins.NotifyGLib']) - # We need to fake our dbus environment for testing purposes since + # We need to fake our gio environment for testing purposes since # the gi library isn't available on CI gi = types.ModuleType(gi_name) gi.repository = types.ModuleType(gi_name + '.repository') @@ -103,61 +105,53 @@ def setup_glib_environment(): # When patching something which has a side effect on the module-level code # of a plugin, make sure to reload it. - reload_plugin('NotifyDBus') + reload_plugin('NotifyGLib') @pytest.fixture -def dbus_environment(mocker): +def gio_environment(mocker): """ Fixture to provide a mocked Dbus environment to test case functions. """ - mocker.patch('gi.repository.Gio.DBusProxy', spec=True, Notify=Mock()) + mocker.patch('gi.repository', spec=True, Notify=Mock()) @pytest.fixture def glib_environment(): """ - Fixture to provide a mocked Glib environment to test case functions. + Fixture to provide a mocked DBus environment to test case functions. """ setup_glib_environment() @pytest.fixture -def dbus_glib_environment(dbus_environment, glib_environment): +def gio_glib_environment(glib_environment, gio_environment): """ - Fixture to provide a mocked Glib/DBus environment to test case functions. + Fixture to provide a mocked Glib/Gio environment to test case functions. """ pass -def test_plugin_dbus_general_success(mocker, dbus_glib_environment): +def test_plugin_glib_general_success(mocker, gio_glib_environment): """ - NotifyDBus() general tests + NotifyGLib() general tests Test class loading using different arguments, provided via URL. """ # Create our instance (identify all supported types) - obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True - assert isinstance(obj.url(), str) is True - assert obj.url().startswith('dbus://_/') - obj = apprise.Apprise.instantiate('kde://', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True - assert isinstance(obj.url(), str) is True - assert obj.url().startswith('kde://_/') - obj = apprise.Apprise.instantiate('qt://', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True - assert isinstance(obj.url(), str) is True - assert obj.url().startswith('qt://_/') obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + assert isinstance(obj, NotifyGLib) is True + assert isinstance(obj.url(), str) is True + assert obj.url().startswith('glib://_/') + obj = apprise.Apprise.instantiate('gio://', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.url().startswith('glib://_/') obj.duration = 0 # Set our X and Y coordinate and try the notification - assert NotifyDBus( + assert NotifyGLib( x_axis=0, y_axis=0, **{'schema': 'dbus'})\ .notify(title='', body='body', notify_type=apprise.NotifyType.INFO) is True @@ -174,10 +168,10 @@ def test_plugin_dbus_general_success(mocker, dbus_glib_environment): # Test our arguments through the instantiate call obj = apprise.Apprise.instantiate( - 'dbus://_/?image=True', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?image=True', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True - assert obj.url().startswith('dbus://_/') + assert obj.url().startswith('glib://_/') assert re.search('image=yes', obj.url()) assert obj.notify( @@ -185,10 +179,10 @@ def test_plugin_dbus_general_success(mocker, dbus_glib_environment): notify_type=apprise.NotifyType.INFO) is True obj = apprise.Apprise.instantiate( - 'dbus://_/?image=False', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?image=False', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True - assert obj.url().startswith('dbus://_/') + assert obj.url().startswith('glib://_/') assert re.search('image=no', obj.url()) assert obj.notify( @@ -197,24 +191,24 @@ def test_plugin_dbus_general_success(mocker, dbus_glib_environment): # Test priority (alias to urgency) handling obj = apprise.Apprise.instantiate( - 'dbus://_/?priority=invalid', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?priority=invalid', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', notify_type=apprise.NotifyType.INFO) is True obj = apprise.Apprise.instantiate( - 'dbus://_/?priority=high', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?priority=high', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', notify_type=apprise.NotifyType.INFO) is True obj = apprise.Apprise.instantiate( - 'dbus://_/?priority=2', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?priority=2', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', @@ -222,32 +216,32 @@ def test_plugin_dbus_general_success(mocker, dbus_glib_environment): # Test urgency handling obj = apprise.Apprise.instantiate( - 'dbus://_/?urgency=invalid', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?urgency=invalid', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', notify_type=apprise.NotifyType.INFO) is True obj = apprise.Apprise.instantiate( - 'dbus://_/?urgency=high', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?urgency=high', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', notify_type=apprise.NotifyType.INFO) is True obj = apprise.Apprise.instantiate( - 'dbus://_/?urgency=2', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?urgency=2', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', notify_type=apprise.NotifyType.INFO) is True obj = apprise.Apprise.instantiate( - 'dbus://_/?urgency=', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?urgency=', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', @@ -255,33 +249,30 @@ def test_plugin_dbus_general_success(mocker, dbus_glib_environment): # Test x/y obj = apprise.Apprise.instantiate( - 'dbus://_/?x=5&y=5', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + 'glib://_/?x=5&y=5', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True assert isinstance(obj.url(), str) is True assert obj.notify( title='title', body='body', notify_type=apprise.NotifyType.INFO) is True -def test_plugin_dbus_general_failure(dbus_glib_environment): +def test_plugin_glib_general_failure(gio_glib_environment): """ Verify a few failure conditions. """ with pytest.raises(TypeError): - NotifyDBus(**{'schema': 'invalid'}) - - with pytest.raises(TypeError): - apprise.Apprise.instantiate('dbus://_/?x=invalid&y=invalid', + apprise.Apprise.instantiate('glib://_/?x=invalid&y=invalid', suppress_exceptions=False) -def test_plugin_dbus_parse_configuration(dbus_glib_environment): +def test_plugin_glib_parse_configuration(gio_glib_environment): # Test configuration parsing content = """ urls: - - dbus://: + - glib://: - priority: 0 tag: dbus_int low - priority: "0" @@ -301,7 +292,7 @@ def test_plugin_dbus_parse_configuration(dbus_glib_environment): - urgency: invalid tag: dbus_invalid - - dbus://: + - glib://: - priority: 2 tag: dbus_int high - priority: "2" @@ -348,7 +339,7 @@ def test_plugin_dbus_parse_configuration(dbus_glib_environment): assert s.urgency == DBusUrgency.NORMAL -def test_plugin_dbus_missing_icon(mocker, dbus_glib_environment): +def test_plugin_glib_missing_icon(mocker, gio_glib_environment): """ Test exception when loading icon; the notification will still be sent. """ @@ -358,7 +349,7 @@ def test_plugin_dbus_missing_icon(mocker, dbus_glib_environment): gi.repository.GdkPixbuf.Pixbuf.new_from_file.side_effect = \ AttributeError("Something failed") - obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) logger: Mock = mocker.spy(obj, "logger") assert obj.notify( title='title', body='body', @@ -370,11 +361,11 @@ def test_plugin_dbus_missing_icon(mocker, dbus_glib_environment): ] -def test_plugin_dbus_disabled_plugin(dbus_glib_environment): +def test_plugin_glib_disabled_plugin(gio_glib_environment): """ Verify notification will not be submitted if plugin is disabled. """ - obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) obj.enabled = False @@ -383,14 +374,14 @@ def test_plugin_dbus_disabled_plugin(dbus_glib_environment): notify_type=apprise.NotifyType.INFO) is False -def test_plugin_dbus_set_urgency(): +def test_plugin_glib_set_urgency(): """ Test the setting of an urgency. """ - NotifyDBus(urgency=0) + NotifyGLib(urgency=0) -def test_plugin_dbus_gi_missing(dbus_glib_environment): +def test_plugin_glib_gi_missing(gio_glib_environment): """ Verify plugin is not available when the `gi` package is not available. """ @@ -401,15 +392,14 @@ def test_plugin_dbus_gi_missing(dbus_glib_environment): # When patching something which has a side effect on the module-level code # of a plugin, make sure to reload it. - reload_plugin('NotifyDBus') + reload_plugin('NotifyGLib') # Create the instance. obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is False + assert isinstance(obj, NotifyGLib) is False -@pytest.mark.skip('temporary') -def test_plugin_dbus_gi_require_version_error(dbus_glib_environment): +def test_plugin_glib_gi_require_version_error(gio_glib_environment): """ Verify plugin is not available when the `gi.require_version()` croaks. """ @@ -420,36 +410,14 @@ def test_plugin_dbus_gi_require_version_error(dbus_glib_environment): # When patching something which has a side effect on the module-level code # of a plugin, make sure to reload it. - reload_plugin('NotifyDBus') + reload_plugin('NotifyGLib') # Create instance. obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is False + assert isinstance(obj, NotifyGLib) is False -@pytest.mark.skip('temporary') -def test_plugin_dbus_module_croaks(mocker, dbus_glib_environment): - """ - Verify plugin is not available when `gi.repository.Gio` module is missing. - """ - - # Make importing `gi.repository.Gio` raise an ImportError. - mocker.patch.dict( - sys.modules, - {'gi.repository.Gio': compile('raise ImportError()', - 'gi.repository.Gio', - 'exec')}) - - # When patching something which has a side effect on the module-level code - # of a plugin, make sure to reload it. - reload_plugin('NotifyDBus') - - # Verify plugin is not available. - obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) - assert obj is None - - -def test_plugin_dbus_session_croaks(mocker, dbus_glib_environment): +def test_plugin_glib_session_croaks(mocker, gio_glib_environment): """ Verify notification fails if DBus croaks. """ @@ -458,7 +426,7 @@ def test_plugin_dbus_session_croaks(mocker, dbus_glib_environment): side_effect=GLib.Error('test')) setup_glib_environment() - obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) + obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) # Emulate DBus session initialization error. assert obj.notify( @@ -466,8 +434,7 @@ def test_plugin_dbus_session_croaks(mocker, dbus_glib_environment): notify_type=apprise.NotifyType.INFO) is False -@pytest.mark.skip('temporary') -def test_plugin_dbus_interface_notify_croaks(mocker): +def test_plugin_glib_interface_notify_croaks(mocker, gio_glib_environment): """ Fail gracefully if underlying object croaks for whatever reason. """ @@ -477,14 +444,14 @@ def test_plugin_dbus_interface_notify_croaks(mocker): Notify=Mock(side_effect=AttributeError("Something failed"))) setup_glib_environment() - obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) is True + obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) is True logger: Mock = mocker.spy(obj, "logger") assert obj.notify( title='title', body='body', notify_type=apprise.NotifyType.INFO) is False assert [ - call.warning('Failed to send DBus notification.'), - call.debug('DBus Exception: Something failed'), + call.warning('Failed to send GLib/Gio notification.'), + call.debug('GLib/Gio Exception: Something failed'), ] in logger.mock_calls