mirror of https://github.com/caronc/apprise
Dbus/Glib Python binding update + test refactoring (#1377)
Chris Caron
GitHub
parent
18255ebd4c
commit
b0fff4bc5b
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,393 @@
|
|||
# BSD 2-Clause License
|
||||
#
|
||||
# Apprise - Push Notification Library.
|
||||
# Copyright (c) 2025, Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# 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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -0,0 +1,293 @@
|
|||
# BSD 2-Clause License
|
||||
#
|
||||
# Apprise - Push Notification Library.
|
||||
# Copyright (c) 2025, Chris Caron <lead2gold@gmail.com>
|
||||
#
|
||||
# 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
|
||||
Loading…
Reference in New Issue