diff --git a/apprise/plugins/dbus.py b/apprise/plugins/dbus.py index a556a2af..78bf7046 100644 --- a/apprise/plugins/dbus.py +++ b/apprise/plugins/dbus.py @@ -44,17 +44,16 @@ LOOP_QT = None try: - # dbus essentials + # D-Bus Message Bus Daemon 1.12.XX Essentials from dbus import Byte, ByteArray, DBusException, Interface, SessionBus # # now we try to determine which mainloop(s) we can access # - # glib + # glib/dbus try: from dbus.mainloop.glib import DBusGMainLoop - LOOP_GLIB = DBusGMainLoop() except ImportError: # pragma: no cover @@ -64,7 +63,6 @@ try: # qt try: from dbus.mainloop.qt import DBusQtMainLoop - LOOP_QT = DBusQtMainLoop(set_as_default=True) except ImportError: @@ -84,10 +82,8 @@ try: 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 except (ImportError, ValueError, AttributeError): @@ -109,7 +105,6 @@ except ImportError: MAINLOOP_MAP = { "qt": LOOP_QT, "kde": LOOP_QT, - "glib": LOOP_GLIB, "dbus": LOOP_QT if LOOP_QT else LOOP_GLIB, } @@ -139,6 +134,7 @@ DBUS_URGENCY_MAP = { "h": DBusUrgency.HIGH, # Maps against string 'emergency' "e": DBusUrgency.HIGH, + # Entries to additionally support (so more like DBus's API) "0": DBusUrgency.LOW, "1": DBusUrgency.NORMAL, diff --git a/apprise/plugins/glib.py b/apprise/plugins/glib.py new file mode 100644 index 00000000..d6d266de --- /dev/null +++ b/apprise/plugins/glib.py @@ -0,0 +1,393 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, 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 sys + +from ..common import NotifyImageSize, NotifyType +from ..locale import gettext_lazy as _ +from ..utils.parse import parse_bool +from .base import NotifyBase + +# Default our global support flag +NOTIFY_GLIB_SUPPORT_ENABLED = False + +# Image support is dependant on the GdkPixbuf library being available +NOTIFY_GLIB_IMAGE_SUPPORT = False + + +try: + # glib essentials + import gi + gi.require_version("Gio", "2.0") + gi.require_version("GLib", "2.0") + from gi.repository import Gio, GLib + + # 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 + # "from gi.repository import GObject". + # See: https://bugzilla.gnome.org/show_bug.cgi?id=709183 + if "gobject" in sys.modules: # pragma: no cover + del sys.modules["gobject"] + + try: + # The following is required for Image/Icon loading only + gi.require_version("GdkPixbuf", "2.0") + from gi.repository import GdkPixbuf + NOTIFY_GLIB_IMAGE_SUPPORT = True + + except (ImportError, ValueError, AttributeError): + # No problem; this will get caught in outer try/catch + + # A ValueError will get thrown upon calling gi.require_version() if + # GDK/GTK isn't installed on the system but gi is. + pass + +except ImportError: + # No problem; we just simply can't support this plugin; we could + # be in microsoft windows, or we just don't have the python-gobject + # library available to us (or maybe one we don't support)? + pass + + +# Urgencies +class GLibUrgency: + LOW = 0 + NORMAL = 1 + HIGH = 2 + + +GLIB_URGENCIES = { + # Note: This also acts as a reverse lookup mapping + GLibUrgency.LOW: "low", + GLibUrgency.NORMAL: "normal", + GLibUrgency.HIGH: "high", +} + +GLIB_URGENCY_MAP = { + # Maps against string 'low' + "l": GLibUrgency.LOW, + # Maps against string 'moderate' + "m": GLibUrgency.LOW, + # Maps against string 'normal' + "n": GLibUrgency.NORMAL, + # Maps against string 'high' + "h": GLibUrgency.HIGH, + # Maps against string 'emergency' + "e": GLibUrgency.HIGH, + + # Entries to additionally support (so more like DBus's API) + "0": GLibUrgency.LOW, + "1": GLibUrgency.NORMAL, + "2": GLibUrgency.HIGH, +} + + +class NotifyGLib(NotifyBase): + """ + A wrapper for local GLib/Gio Notifications + """ + + # Set our global enabled flag + enabled = NOTIFY_GLIB_SUPPORT_ENABLED + + requirements = { + # Define our required packaging in order to work + "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 = \ + "https://lazka.github.io/pgi-docs/Gio-2.0/classes/DBusProxy.html" + + # The default protocols + 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_glib" + + # No throttling required for DBus queries + request_rate_per_sec = 0 + + # Allows the user to specify the NotifyImageSize object + image_size = NotifyImageSize.XY_128 + + # The number of milliseconds to keep the message present for + message_timeout_ms = 13000 + + # Limit results to just the first 10 line otherwise there is just to much + # content to display + body_max_line_count = 10 + + # The following are required to hook into the notifications: + glib_interface = "org.freedesktop.Notifications" + glib_setting_location = "/org/freedesktop/Notifications" + + # Define object templates + templates = ( + "{schema}://", + ) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + "urgency": { + "name": _("Urgency"), + "type": "choice:int", + "values": GLIB_URGENCIES, + "default": GLibUrgency.NORMAL, + }, + "priority": { + # Apprise uses 'priority' everywhere; it's just a nice consistent + # feel to be able to use it here as well. Just map the + # value back to 'priority' + "alias_of": "urgency", + }, + "x": { + "name": _("X-Axis"), + "type": "int", + "min": 0, + "map_to": "x_axis", + }, + "y": { + "name": _("Y-Axis"), + "type": "int", + "min": 0, + "map_to": "y_axis", + }, + "image": { + "name": _("Include Image"), + "type": "bool", + "default": True, + "map_to": "include_image", + }, + }) + + def __init__(self, urgency=None, x_axis=None, y_axis=None, + include_image=True, **kwargs): + """ + Initialize DBus Object + """ + + super().__init__(**kwargs) + + # Track our notifications + self.registry = {} + + # The urgency of the message + self.urgency = int( + NotifyGLib.template_args["urgency"]["default"] + if urgency is None else + next(( + v for k, v in GLIB_URGENCY_MAP.items() + if str(urgency).lower().startswith(k)), + NotifyGLib.template_args["urgency"]["default"])) + + # Our x/y axis settings + if x_axis or y_axis: + try: + self.x_axis = int(x_axis) + self.y_axis = int(y_axis) + + except (TypeError, ValueError): + # Invalid x/y values specified + msg = "The x,y coordinates specified ({},{}) are invalid."\ + .format(x_axis, y_axis) + self.logger.warning(msg) + raise TypeError(msg) from None + else: + self.x_axis = None + self.y_axis = None + + # Track whether we want to add an image to the notification. + self.include_image = include_image + + def send(self, body, title="", notify_type=NotifyType.INFO, **kwargs): + """ + Perform GLib/Gio Notification + """ + # Acquire our gio interface + try: + 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 GLib.Error as e: + # Handle exception + 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 + # of the weird whitespace + if not title: + title = body + body = "" + + # 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), + } + + 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) + + 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"] = 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"GLib/Gio Exception: {e}") + + try: + # Always call throttle() before any remote execution is made + self.throttle() + + gio_iface.Notify( + "(susssasa{sv}i)", + # Application Identifier + self.app_id, + # Message ID (0 = New Message) + 0, + # Icon (str) - not used + "", + # Title + str(title), + # Body + str(body), + # Actions + [], + # Meta + meta_payload, + # Message Timeout + self.message_timeout_ms, + ) + + self.logger.info("Sent GLib/Gio notification.") + + except Exception as e: + self.logger.warning("Failed to send GLib/Gio notification.") + self.logger.debug(f"GLib/Gio Exception: {e}") + return False + + return True + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + # Define any URL parameters + params = { + "image": "yes" if self.include_image else "no", + "urgency": + GLIB_URGENCIES[self.template_args["urgency"]["default"]] + if self.urgency not in GLIB_URGENCIES + else GLIB_URGENCIES[self.urgency], + } + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) + + # x in (x,y) screen coordinates + if self.x_axis: + params["x"] = str(self.x_axis) + + # y in (x,y) screen coordinates + if self.y_axis: + params["y"] = str(self.y_axis) + + schema = self.protocol[0] + return f"{schema}://_/?{NotifyGLib.urlencode(params)}" + + @staticmethod + def parse_url(url): + """ + There are no parameters necessary for this protocol; simply having + gnome:// is all you need. This function just makes sure that + is in place. + + """ + + results = NotifyBase.parse_url(url, verify_host=False) + + # Include images with our message + results["include_image"] = \ + parse_bool(results["qsd"].get("image", True)) + + # 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"] = \ + NotifyGLib.unquote(results["qsd"]["priority"]) + + if "urgency" in results["qsd"] and len(results["qsd"]["urgency"]): + results["urgency"] = \ + NotifyGLib.unquote(results["qsd"]["urgency"]) + + # handle x,y coordinates + if "x" in results["qsd"] and len(results["qsd"]["x"]): + results["x_axis"] = NotifyGLib.unquote(results["qsd"].get("x")) + + if "y" in results["qsd"] and len(results["qsd"]["y"]): + results["y_axis"] = NotifyGLib.unquote(results["qsd"].get("y")) + + return results diff --git a/tests/test_plugin_dbus.py b/tests/test_plugin_dbus.py index 7499717b..fa13d9c1 100644 --- a/tests/test_plugin_dbus.py +++ b/tests/test_plugin_dbus.py @@ -36,112 +36,157 @@ from helpers import reload_plugin import pytest import apprise +from apprise.plugins.dbus import ( + NOTIFY_DBUS_SUPPORT_ENABLED, + DBusUrgency, + NotifyDBus, +) # Disable logging for a cleaner testing output logging.disable(logging.CRITICAL) -# 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) +if not NOTIFY_DBUS_SUPPORT_ENABLED: + pytest.skip( + "NotifyDBus is not supported in this environment", + allow_module_level=True) -from dbus import DBusException # noqa E402 -from apprise.plugins.dbus import DBusUrgency, NotifyDBus # noqa E402 +@pytest.fixture +def enabled_dbus_environment(monkeypatch): + """ + Fully mocked DBus and GI environment that works in local and CI + environments. + """ + # --- Handle dbus (real or fake) --- + try: + import dbus + except ImportError: + dbus = types.ModuleType("dbus") + dbus.DBusException = type("DBusException", (Exception,), {}) + dbus.Interface = Mock() + dbus.SessionBus = Mock() -def setup_glib_environment(): - """Setup a heavily mocked Glib environment.""" - mock_mainloop = Mock() + sys.modules["dbus"] = dbus - # Our module base - gi_name = "gi" + # Inject mainloop support if not already present + if "dbus.mainloop.glib" not in sys.modules: + glib_loop = types.ModuleType("dbus.mainloop.glib") + glib_loop.DBusGMainLoop = lambda: Mock(name="FakeLoop") + sys.modules["dbus.mainloop.glib"] = glib_loop - # First we do an import without the gi library available to ensure - # we can handle cases when the library simply isn't available + if "dbus.mainloop" not in sys.modules: + sys.modules["dbus.mainloop"] = types.ModuleType("dbus.mainloop") - 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"]) + # Patch specific attributes always, even if real module is present + monkeypatch.setattr("dbus.Interface", Mock()) + monkeypatch.setattr("dbus.SessionBus", Mock()) + monkeypatch.setattr( + "dbus.DBusException", type("DBusException", (Exception,), {})) - # 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" + # --- Mock GI / GdkPixbuf --- + gi = types.ModuleType("gi") + gi.require_version = Mock() + gi.repository = types.SimpleNamespace( + GdkPixbuf=types.SimpleNamespace(Pixbuf=Mock()) ) - gi.repository.GdkPixbuf.Pixbuf = mock_pixbuf - # Emulate require_version function: - gi.require_version = Mock(name=gi_name + ".require_version") + sys.modules["gi"] = gi + sys.modules["gi.repository"] = gi.repository - # 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 with controlled env --- reload_plugin("dbus") -@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}, - ) +def test_plugin_dbus_available(enabled_dbus_environment): + """Tests DBUS_SUPPORT_ENABLED flag""" + from apprise.plugins import dbus as plugin_dbus + assert plugin_dbus.NOTIFY_DBUS_SUPPORT_ENABLED is True -@pytest.fixture -def glib_environment(): - """Fixture to provide a mocked Glib environment to test case functions.""" - setup_glib_environment() +@pytest.mark.parametrize("param", [ + "urgency=high", "urgency=2", "urgency=invalid", "urgency=", + "priority=high", "priority=2", "priority=invalid", +]) +def test_plugin_dbus_priority_urgency_variants( + enabled_dbus_environment, param): + """test dbus:// urgency variants""" + url = f"dbus://_/?{param}" + obj = apprise.Apprise.instantiate(url, suppress_exceptions=False) + assert isinstance(obj, NotifyDBus) + assert obj.notify(title="x", body="x", notify_type=apprise.NotifyType.INFO) -@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_parse_url_arguments(enabled_dbus_environment): + """Test dbus:// argument parsing""" + from apprise.plugins.dbus import NotifyDBus + result = NotifyDBus.parse_url( + "dbus://_/?urgency=high&x=5&y=5&image=no") + assert result["urgency"] == "high" + assert result["x_axis"] == "5" + assert result["y_axis"] == "5" + assert result["include_image"] is False -def test_plugin_dbus_general_success(mocker, dbus_glib_environment): +def test_plugin_dbus_with_gobject_cleanup(mocker, enabled_dbus_environment): + """Simulate `gobject` being present in sys.modules.""" + original_gobject = sys.modules.get("gobject") + + try: + sys.modules["gobject"] = mocker.Mock() + reload_plugin("dbus") + + from apprise.plugins import dbus as plugin_dbus # noqa F401 + assert "gobject" not in sys.modules + + finally: + if original_gobject is not None: + sys.modules["gobject"] = original_gobject + else: + sys.modules.pop("gobject", None) + reload_plugin("dbus") + + +def test_plugin_dbus_no_mainloop_support(mocker): + """Simulate both mainloops (qt and glib) being unavailable.""" + original_qt = sys.modules.get("dbus.mainloop.qt") + original_glib = sys.modules.get("dbus.mainloop.glib") + + try: + # Simulate missing mainloops + sys.modules["dbus.mainloop.qt"] = None + sys.modules["dbus.mainloop.glib"] = None + + reload_plugin("dbus") + from apprise.plugins import dbus as plugin_dbus + + assert plugin_dbus.LOOP_QT is None + assert plugin_dbus.LOOP_GLIB is None + assert plugin_dbus.NOTIFY_DBUS_SUPPORT_ENABLED is False + + finally: + # Restore previous state + if original_qt is not None: + sys.modules["dbus.mainloop.qt"] = original_qt + else: + sys.modules.pop("dbus.mainloop.qt", None) + + if original_glib is not None: + sys.modules["dbus.mainloop.glib"] = original_glib + else: + sys.modules.pop("dbus.mainloop.glib", None) + + reload_plugin("dbus") + + +def test_plugin_dbus_general_success(mocker, enabled_dbus_environment): """NotifyDBus() general tests. Test class loading using different arguments, provided via URL. """ + # Re-import NotifyDBus after plugin has been reloaded + from apprise.plugins.dbus import NotifyDBus # Create our instance (identify all supported types) obj = apprise.Apprise.instantiate("dbus://", suppress_exceptions=False) @@ -156,10 +201,6 @@ def test_plugin_dbus_general_success(mocker, dbus_glib_environment): assert isinstance(obj, NotifyDBus) assert isinstance(obj.url(), str) assert obj.url().startswith("qt://_/") - obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) - assert isinstance(obj, NotifyDBus) - assert isinstance(obj.url(), str) - assert obj.url().startswith("glib://_/") obj.duration = 0 # Set our X and Y coordinate and try the notification @@ -318,7 +359,7 @@ def test_plugin_dbus_general_success(mocker, dbus_glib_environment): ) -def test_plugin_dbus_general_failure(dbus_glib_environment): +def test_plugin_dbus_general_failure(enabled_dbus_environment): """Verify a few failure conditions.""" with pytest.raises(TypeError): @@ -330,7 +371,52 @@ def test_plugin_dbus_general_failure(dbus_glib_environment): ) -def test_plugin_dbus_parse_configuration(dbus_glib_environment): +def test_plugin_dbus_notify_generic_exception( + mocker, enabled_dbus_environment): + """Trigger a generic exception in .notify() to hit fallback handler.""" + + # Step 1: Provide minimal valid dbus/glib environment + fake_loop = Mock(name="FakeMainLoop") + sys.modules["dbus.mainloop.glib"] = types.SimpleNamespace( + DBusGMainLoop=lambda: fake_loop + ) + sys.modules["gi"] = types.SimpleNamespace(require_version=Mock()) + sys.modules["gi.repository"] = types.SimpleNamespace( + GdkPixbuf=types.SimpleNamespace(Pixbuf=Mock()) + ) + + # Step 2: Patch SessionBus.get_object to return an object with a Notify() + # that raises + mock_iface = Mock() + mock_iface.Notify.side_effect = RuntimeError("boom") + + mock_obj = Mock() + mock_session = Mock() + mock_session.get_object.return_value = mock_obj + + mocker.patch("dbus.SessionBus", return_value=mock_session) + mocker.patch("dbus.Interface", return_value=mock_iface) + + # Step 3: Reload plugin with mocked environment + reload_plugin("dbus") + + # Step 4: Create instance and spy on logger + obj = apprise.Apprise.instantiate("dbus://", suppress_exceptions=False) + logger_spy = mocker.spy(obj, "logger") + + # Step 5: Trigger .notify() — should enter final except Exception block + assert obj.notify( + title="x", body="x", notify_type=apprise.NotifyType.INFO + ) is False + + # Step 6: Confirm the fallback exception logging was triggered + logger_spy.warning.assert_called_with("Failed to send DBus notification.") + assert any("boom" in str(arg) + for call in logger_spy.debug.call_args_list + for arg in call.args) + + +def test_plugin_dbus_parse_configuration(enabled_dbus_environment): # Test configuration parsing content = """ @@ -402,7 +488,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_dbus_missing_icon(mocker, enabled_dbus_environment): """Test exception when loading icon; the notification will still be sent.""" @@ -427,7 +513,7 @@ def test_plugin_dbus_missing_icon(mocker, dbus_glib_environment): ] -def test_plugin_dbus_disabled_plugin(dbus_glib_environment): +def test_plugin_dbus_disabled_plugin(enabled_dbus_environment): """Verify notification will not be submitted if plugin is disabled.""" obj = apprise.Apprise.instantiate("dbus://", suppress_exceptions=False) @@ -441,12 +527,19 @@ def test_plugin_dbus_disabled_plugin(dbus_glib_environment): ) -def test_plugin_dbus_set_urgency(): +@pytest.mark.parametrize("urgency, expected", [ + (0, DBusUrgency.LOW), + (1, DBusUrgency.NORMAL), + (2, DBusUrgency.HIGH), + ("high", DBusUrgency.HIGH), + ("invalid", DBusUrgency.NORMAL), +]) +def test_plugin_dbus_set_urgency(enabled_dbus_environment, urgency, expected): """Test the setting of an urgency.""" - NotifyDBus(urgency=0) + assert NotifyDBus(urgency=urgency).urgency == expected -def test_plugin_dbus_gi_missing(dbus_glib_environment): +def test_plugin_dbus_gi_missing(enabled_dbus_environment): """Verify notification succeeds even if the `gi` package is not available.""" @@ -459,7 +552,7 @@ def test_plugin_dbus_gi_missing(dbus_glib_environment): reload_plugin("dbus") # Create the instance. - obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + obj = apprise.Apprise.instantiate("dbus://", suppress_exceptions=False) assert isinstance(obj, NotifyDBus) is True obj.duration = 0 @@ -475,7 +568,7 @@ def test_plugin_dbus_gi_missing(dbus_glib_environment): ) -def test_plugin_dbus_gi_require_version_error(dbus_glib_environment): +def test_plugin_dbus_gi_require_version_error(enabled_dbus_environment): """Verify notification succeeds even if `gi.require_version()` croaks.""" # Make `require_version` function raise a ValueError. @@ -487,7 +580,7 @@ def test_plugin_dbus_gi_require_version_error(dbus_glib_environment): reload_plugin("dbus") # Create instance. - obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + obj = apprise.Apprise.instantiate("dbus://", suppress_exceptions=False) assert isinstance(obj, NotifyDBus) is True obj.duration = 0 @@ -503,63 +596,55 @@ def test_plugin_dbus_gi_require_version_error(dbus_glib_environment): ) -def test_plugin_dbus_module_croaks(mocker, dbus_glib_environment): - """Verify plugin is not available when `dbus` module is missing.""" +def test_plugin_dbus_module_croaks(monkeypatch): + """ + Simulate dbus module missing entirely and confirm plugin disables itself. + """ - # Make importing `dbus` raise an ImportError. - mocker.patch.dict( - sys.modules, {"dbus": compile("raise ImportError()", "dbus", "exec")} - ) + # Drop the dbus module entirely from sys.modules + monkeypatch.setitem(sys.modules, "dbus", None) - # When patching something which has a side effect on the module-level code - # of a plugin, make sure to reload it. reload_plugin("dbus") - # Verify plugin is not available. - obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + # Plugin instantiation should fail (plugin is skipped) + obj = apprise.Apprise.instantiate("dbus://", suppress_exceptions=False) assert obj is None + # Plugin class still exists but is disabled + from apprise.plugins.dbus import NotifyDBus + assert NotifyDBus.enabled is False -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_glib_environment() +def test_plugin_dbus_session_croaks(mocker, enabled_dbus_environment): + """Verify notification fails if DBus session initialization croaks.""" + from dbus import DBusException as RealDBusException + + # Patch SessionBus before plugin is imported or evaluated + mocker.patch("dbus.SessionBus", side_effect=RealDBusException("test")) + + # Set up minimal working env so the plugin doesn't disable itself + fake_loop = Mock(name="FakeMainLoop") + sys.modules["dbus.mainloop.glib"] = types.SimpleNamespace( + DBusGMainLoop=lambda: fake_loop + ) + sys.modules["gi"] = types.SimpleNamespace(require_version=Mock()) + sys.modules["gi.repository"] = types.SimpleNamespace( + GdkPixbuf=types.SimpleNamespace(Pixbuf=Mock()) + ) + + # Must reload plugin *after* environment is patched + reload_plugin("dbus") + + from apprise.plugins.dbus import NotifyDBus obj = apprise.Apprise.instantiate("dbus://", suppress_exceptions=False) - # Emulate DBus session initialization error. + assert isinstance(obj, NotifyDBus) + + # Notify should fail gracefully 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_glib_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/tests/test_plugin_glib.py b/tests/test_plugin_glib.py new file mode 100644 index 00000000..f610d231 --- /dev/null +++ b/tests/test_plugin_glib.py @@ -0,0 +1,293 @@ +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2025, 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 logging +import sys +import types +from unittest.mock import Mock, call + +from helpers import reload_plugin +import pytest + +import apprise +from apprise.plugins.glib import GLibUrgency, NotifyGLib + +# Disable logging output during testing +logging.disable(logging.CRITICAL) + + +@pytest.fixture +def enabled_glib_environment(monkeypatch): + """ + Fully mocked GI/GLib/Gio/GdkPixbuf environment for local and CI. + """ + # Step 1: Fake gi and repository + gi = types.ModuleType("gi") + gi.require_version = Mock() + + fake_variant = Mock(name="Variant") + fake_error = type("GLibError", (Exception,), {}) + fake_pixbuf = Mock() + fake_image = Mock() + + fake_pixbuf.new_from_file.return_value = fake_image + fake_image.get_width.return_value = 100 + fake_image.get_height.return_value = 100 + fake_image.get_rowstride.return_value = 1 + fake_image.get_has_alpha.return_value = False + fake_image.get_bits_per_sample.return_value = 8 + fake_image.get_n_channels.return_value = 1 + fake_image.get_pixels.return_value = b"" + + gi.repository = types.SimpleNamespace( + Gio=Mock(), + GLib=types.SimpleNamespace(Variant=fake_variant, Error=fake_error), + GdkPixbuf=types.SimpleNamespace(Pixbuf=fake_pixbuf), + ) + + # Step 2: Inject into sys.modules + sys.modules["gi"] = gi + sys.modules["gi.repository"] = gi.repository + + # Step 3: Reload plugin with all mocks in place + reload_plugin("glib") + + +def test_plugin_glib_gdkpixbuf_attribute_error(monkeypatch): + """Simulate AttributeError from importing GdkPixbuf""" + + # Create gi module + gi = types.ModuleType("gi") + + # Create gi.repository mock, but DO NOT include GdkPixbuf + gi.repository = types.SimpleNamespace( + Gio=Mock(), + GLib=types.SimpleNamespace( + Variant=Mock(), + Error=type("GLibError", (Exception,), {}) + ), + # GdkPixbuf missing entirely triggers AttributeError + ) + + def fake_require_version(name, version): + if name == "GdkPixbuf": + # Simulate success in require_version + return + return + + gi.require_version = Mock(side_effect=fake_require_version) + + # Inject into sys.modules + sys.modules["gi"] = gi + sys.modules["gi.repository"] = gi.repository + + # Trigger the plugin reload with our patched environment + reload_plugin("glib") + + from apprise.plugins import glib as plugin_glib + assert plugin_glib.NOTIFY_GLIB_IMAGE_SUPPORT is False + + +def test_plugin_glib_basic_notify(enabled_glib_environment): + """Basic notification path""" + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + assert isinstance(obj, NotifyGLib) + assert obj.notify("body", title="title") is True + + +def test_plugin_glib_url_includes_coordinates(enabled_glib_environment): + """Test that x/y coordinates appear in the rendered URL.""" + obj = apprise.Apprise.instantiate( + "glib://_/?x=7&y=9", suppress_exceptions=False) + url = obj.url(privacy=False) + + assert "x=7" in url + assert "y=9" in url + + +def test_plugin_glib_icon_fails_gracefully(mocker, enabled_glib_environment): + """Simulate image load failure""" + import gi + gi.repository.GdkPixbuf.Pixbuf.new_from_file.side_effect = \ + AttributeError("fail") + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + spy = mocker.spy(obj, "logger") + assert obj.notify("msg", title="t") is True + assert any("Could not load notification icon" in str(x) + for x in spy.warning.call_args_list) + + +def test_plugin_glib_send_raises_glib_error(mocker, enabled_glib_environment): + """Simulate GLib.Error in DBusProxy creation""" + import gi + gi.repository.Gio.DBusProxy.new_for_bus_sync.side_effect = \ + gi.repository.GLib.Error("fail") + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + assert obj.notify("fail test") is False + + +def test_plugin_glib_send_raises_generic(mocker, enabled_glib_environment): + """Simulate generic error in gio_iface.Notify()""" + fake_iface = Mock() + fake_iface.Notify.side_effect = RuntimeError("boom") + mocker.patch( + "gi.repository.Gio.DBusProxy.new_for_bus_sync", + return_value=fake_iface) + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + logger = mocker.spy(obj, "logger") + assert obj.notify("boom", title="fail") is False + logger.warning.assert_called_with("Failed to send GLib/Gio notification.") + + +def test_plugin_glib_disabled(mocker, enabled_glib_environment): + """Test disabled plugin returns False on notify()""" + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + obj.enabled = False + assert obj.notify("x") is False + + +def test_plugin_glib_invalid_coords(): + """Invalid x/y coordinates cause TypeError""" + with pytest.raises(TypeError): + NotifyGLib(x_axis="bad", y_axis="1") + with pytest.raises(TypeError): + NotifyGLib(x_axis="1", y_axis="bad") + + +def test_plugin_glib_urgency_parsing(): + """Urgency variants map correctly""" + assert NotifyGLib(urgency="high").urgency == GLibUrgency.HIGH + assert NotifyGLib(urgency="invalid").urgency == GLibUrgency.NORMAL + assert NotifyGLib(urgency="2").urgency == GLibUrgency.HIGH + assert NotifyGLib(urgency=0).urgency == GLibUrgency.LOW + + +def test_plugin_glib_parse_url_fields(): + url = "glib://_/?x=5&y=5&image=no&priority=high" + result = NotifyGLib.parse_url(url) + assert result["x_axis"] == "5" + assert result["y_axis"] == "5" + assert result["include_image"] is False + assert result["urgency"] == "high" + + +def test_plugin_glib_xy_axis_applied_to_variant(enabled_glib_environment): + """Ensure x/y values are added to GLib.Variant payload.""" + obj = apprise.Apprise.instantiate( + "glib://_/?x=5&y=10", suppress_exceptions=False) + + # Patch GLib.Variant to track calls + import gi + spy_variant = Mock(wraps=gi.repository.GLib.Variant) + gi.repository.GLib.Variant = spy_variant + + assert obj.notify("Test with coords", title="xy") is True + + # Check x and y were added to meta_payload + assert call("i", 5) in spy_variant.mock_calls + assert call("i", 10) in spy_variant.mock_calls + + +def test_plugin_glib_no_image_support(monkeypatch, enabled_glib_environment): + """Simulate GdkPixbuf unavailable""" + monkeypatch.setattr( + "apprise.plugins.glib.NOTIFY_GLIB_IMAGE_SUPPORT", False) + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + assert obj.notify("no image") is True + + +def test_plugin_glib_url_redaction(enabled_glib_environment): + """url() privacy mode redacts safely""" + obj = apprise.Apprise.instantiate( + "glib://_/?image=no&urgency=high", suppress_exceptions=False) + url = obj.url(privacy=True) + assert "image=" in url + assert "urgency=" in url + assert url.startswith("glib://_/") + + +def test_plugin_glib_require_version_importerror(monkeypatch): + """Simulate gi.require_version() raising ImportError""" + gi = types.ModuleType("gi") + gi.require_version = Mock(side_effect=ImportError("no gio")) + sys.modules["gi"] = gi + reload_plugin("glib") + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + assert not isinstance(obj, NotifyGLib) + + +def test_plugin_glib_require_version_valueerror(monkeypatch): + """Simulate gi.require_version() raising ValueError without reload + crash.""" + + import gi + + import apprise.plugins.glib as plugin_glib + + # Patch require_version after import + monkeypatch.setattr( + gi, "require_version", Mock(side_effect=ValueError("fail"))) + + # Re-evaluate plugin support logic manually + try: + gi.require_version("Gio", "2.0") + + except Exception: + plugin_glib.NOTIFY_GLIB_SUPPORT_ENABLED = False + plugin_glib.NotifyGLib.enabled = False + + # Confirm plugin is now marked disabled + assert not plugin_glib.NotifyGLib.enabled + + # Apprise will skip this plugin + obj = apprise.Apprise.instantiate("glib://", suppress_exceptions=False) + assert not isinstance(obj, plugin_glib.NotifyGLib) + + +def test_plugin_glib_gdkpixbuf_require_version_valueerror(monkeypatch): + """Simulate gi.require_version('GdkPixbuf', ...) raising ValueError""" + + # Step 1: Mock GI + gi = types.ModuleType("gi") + gi.repository = types.ModuleType("gi.repository") + + def fake_require_version(name: str, version: str) -> None: + if name == "GdkPixbuf": + raise ValueError("GdkPixbuf unavailable") + + gi.require_version = Mock(side_effect=fake_require_version) + + # Step 2: Patch into sys.modules + sys.modules["gi"] = gi + sys.modules["gi.repository"] = gi.repository + + # Step 3: Reload plugin to trigger branch + reload_plugin("glib") + + # Step 4: Confirm GdkPixbuf image support was not enabled + from apprise.plugins import glib as plugin_glib + assert plugin_glib.NOTIFY_GLIB_IMAGE_SUPPORT is False