DBus Support - kde://, qt://, dbus://, and glib:// added; refs #36

This commit is contained in:
Chris Caron 2019-02-03 01:39:14 -05:00
parent 4f2d917a2f
commit e95bfcf6fd
8 changed files with 577 additions and 10 deletions

View File

@ -1,5 +1,11 @@
language: python
dist: xenial
addons:
apt:
packages:
- libdbus-1-dev
matrix:
include:
@ -12,30 +18,27 @@ matrix:
- python: "3.6"
env: TOXENV=py36
- python: "3.7"
dist: xenial
env: TOXENV=py37
- python: "pypy2.7-5.8.0"
- python: "pypy2.7-6.0"
env: TOXENV=pypy
- python: "pypy3.5-5.8.0"
- python: "pypy3.5-6.0"
env: TOXENV=pypy3
install:
- pip install .
- pip install codecov tox
- pip install -r dev-requirements.txt
- pip install -r requirements.txt
- if [[ $TRAVIS_PYTHON_VERSION != 'pypy'* ]]; then travis_retry pip install dbus-python; fi
# run tests
script:
- tox
after_success:
- tox -e coverage-report
- codecov
notifications:
email: false

View File

@ -33,6 +33,7 @@ The table below identifies the services this tool supports and some example serv
| -------------------- | ---------- | ------------ | -------------- |
| [Boxcar](https://github.com/caronc/apprise/wiki/Notify_boxcar) | boxcar:// | (TCP) 443 | boxcar://hostname<br />boxcar://hostname/@tag<br/>boxcar://hostname/device_token<br />boxcar://hostname/device_token1/device_token2/device_tokenN<br />boxcar://hostname/@tag/@tag2/device_token
| [Discord](https://github.com/caronc/apprise/wiki/Notify_discord) | discord:// | (TCP) 443 | discord://webhook_id/webhook_token<br />discord://avatar@webhook_id/webhook_token
| [Dbus](https://github.com/caronc/apprise/wiki/Notify_dbus) | dbus://<br />qt://<br />glib://<br />kde:// | n/a | dbus://<br />qt://<br />glib://<br />kde://
| [Emby](https://github.com/caronc/apprise/wiki/Notify_emby) | emby:// or embys:// | (TCP) 8096 | emby://user@hostname/<br />emby://user:password@hostname
| [Faast](https://github.com/caronc/apprise/wiki/Notify_faast) | faast:// | (TCP) 443 | faast://authorizationtoken
| [Gnome](https://github.com/caronc/apprise/wiki/Notify_gnome) | gnome:// | n/a | gnome://

View File

@ -31,6 +31,7 @@ from .common import NotifyType
from .common import NotifyFormat
from .utils import parse_list
from .utils import compat_is_basestring
from .utils import GET_SCHEMA_RE
from .AppriseAsset import AppriseAsset
@ -43,9 +44,6 @@ logger = logging.getLogger(__name__)
# Build a list of supported plugins
SCHEMA_MAP = {}
# Used for attempting to acquire the schema if the URL can't be parsed.
GET_SCHEMA_RE = re.compile(r'\s*(?P<schema>[a-z0-9]{3,9})://.*$', re.I)
# Load our Lookup Matrix
def __load_matrix():

View File

@ -0,0 +1,322 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from __future__ import absolute_import
from __future__ import print_function
import re
from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
from ..utils import GET_SCHEMA_RE
# Default our global support flag
NOTIFY_DBUS_SUPPORT_ENABLED = False
# Image support is dependant on the GdkPixbuf library being available
NOTIFY_DBUS_IMAGE_SUPPORT = False
# The following are required to hook into the notifications:
NOTIFY_DBUS_INTERFACE = 'org.freedesktop.Notifications'
NOTIFY_DBUS_SETTING_LOCATION = '/org/freedesktop/Notifications'
# Initialize our mainloops
LOOP_GLIB = None
LOOP_QT = None
try:
# dbus essentials
from dbus import SessionBus
from dbus import Interface
from dbus import Byte
from dbus import ByteArray
#
# now we try to determine which mainloop(s) we can access
#
# glib
try:
from dbus.mainloop.glib import DBusGMainLoop
LOOP_GLIB = DBusGMainLoop()
except ImportError:
# No problem
pass
# qt
try:
from dbus.mainloop.qt 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)
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:
# No problem; this will get caught in outer try/catch
raise
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
# 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,
'glib': LOOP_GLIB,
'dbus': LOOP_QT if LOOP_QT else LOOP_GLIB,
}
# Urgencies
class DBusUrgency(object):
LOW = 0
NORMAL = 1
HIGH = 2
# Define our urgency levels
DBUS_URGENCIES = (
DBusUrgency.LOW,
DBusUrgency.NORMAL,
DBusUrgency.HIGH,
)
class NotifyDBus(NotifyBase):
"""
A wrapper for local DBus/Qt Notifications
"""
# The default descriptive name associated with the Notification
service_name = 'DBus Notification'
# The default protocols
# Python 3 keys() does not return a list object, it's it's 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
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'
# Allows the user to specify the NotifyImageSize object
image_size = NotifyImageSize.XY_128
# The number of seconds to keep the message present for
message_timeout_ms = 13000
# This entry is a bit hacky, but it allows us to unit-test this library
# in an environment that simply doesn't have the gnome packages
# available to us. It also allows us to handle situations where the
# packages actually are present but we need to test that they aren't.
# If anyone is seeing this had knows a better way of testing this
# outside of what is defined in test/test_glib_plugin.py, please
# let me know! :)
_enabled = NOTIFY_DBUS_SUPPORT_ENABLED
def __init__(self, urgency=None, x_axis=None, y_axis=None, **kwargs):
"""
Initialize DBus Object
"""
super(NotifyDBus, self).__init__(**kwargs)
# Track our notifications
self.registry = {}
# Store our schema; default to dbus
self.schema = kwargs.get('schema', 'dbus')
if self.schema not in MAINLOOP_MAP:
# Unsupported Schema
self.logger.warning(
'The schema specified ({}) is not supported.'
.format(self.schema))
raise TypeError(
'The schema specified ({}) is not supported.'
.format(self.schema))
# The urgency of the message
if urgency not in DBUS_URGENCIES:
self.urgency = DBusUrgency.NORMAL
else:
self.urgency = urgency
# Our x/y axis settings
self.x_axis = x_axis
self.y_axis = y_axis
def notify(self, title, body, notify_type, **kwargs):
"""
Perform DBus Notification
"""
if not self._enabled or MAINLOOP_MAP[self.schema] is None:
self.logger.warning(
"{} notifications could not be loaded.".format(self.schema))
return False
# Acquire our session
session = SessionBus(mainloop=MAINLOOP_MAP[self.schema])
# acquire our dbus object
dbus_obj = session.get_object(
NOTIFY_DBUS_INTERFACE,
NOTIFY_DBUS_SETTING_LOCATION,
)
# Acquire our dbus interface
dbus_iface = Interface(
dbus_obj,
dbus_interface=NOTIFY_DBUS_INTERFACE,
)
# image path
icon_path = self.image_path(notify_type, extension='.ico')
# Our meta payload
meta_payload = {
"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'] = self.x_axis
meta_payload['y'] = self.y_axis
if NOTIFY_DBUS_IMAGE_SUPPORT is True:
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())
)
except Exception as e:
self.logger.warning(
"Could not load Gnome notification icon ({}): {}"
.format(icon_path, e))
# Limit results to just the first 10 line otherwise
# there is just to much content to display
body = re.split('[\r\n]+', body)
if title:
# Place title on first line if it exists
body.insert(0, title)
body = '\r\n'.join(body[0:10])
try:
dbus_iface.Notify(
# Application Identifier
self.app_id,
# Message ID (0 = New Message)
0,
# Icon (str) - not used
'',
# Title
str(title),
# Body
str(body),
# Actions
list(),
# Meta
meta_payload,
# Message Timeout
self.message_timeout_ms,
)
self.logger.info('Sent DBus notification.')
except Exception as e:
self.logger.warning('Failed to send DBus notification.')
self.logger.exception('DBus Exception')
return False
return True
@staticmethod
def parse_url(url):
"""
There are no parameters nessisary for this protocol; simply having
gnome:// is all you need. This function just makes sure that
is in place.
"""
schema = GET_SCHEMA_RE.match(url)
if schema is None:
# Content is simply not parseable
return None
# return a very basic set of requirements
return {
'schema': schema.group('schema').lower(),
'user': None,
'password': None,
'port': None,
'host': 'localhost',
'fullpath': None,
'path': None,
'url': url,
'qsd': {},
# screen lat/lon (in pixels) where x=0 and y=0 if you want to put
# the notification in the top left hand side. Accept defaults if
# set to None
'x_axis': None,
'y_axis': None,
# Set the urgency to None so that we fall back to the default
# value.
'urgency': None,
}

View File

@ -28,6 +28,7 @@
from . import NotifyEmail as NotifyEmailBase
from .NotifyBoxcar import NotifyBoxcar
from .NotifyDBus import NotifyDBus
from .NotifyDiscord import NotifyDiscord
from .NotifyEmail import NotifyEmail
from .NotifyEmby import NotifyEmby
@ -63,7 +64,7 @@ from ..common import NOTIFY_TYPES
__all__ = [
# Notification Services
'NotifyBoxcar', 'NotifyEmail', 'NotifyEmby', 'NotifyDiscord',
'NotifyBoxcar', 'NotifyDBus', 'NotifyEmail', 'NotifyEmby', 'NotifyDiscord',
'NotifyFaast', 'NotifyGnome', 'NotifyGrowl', 'NotifyIFTTT', 'NotifyJoin',
'NotifyJSON', 'NotifyMatrix', 'NotifyMatterMost', 'NotifyProwl',
'NotifyPushed', 'NotifyPushBullet', 'NotifyPushjet',

View File

@ -91,6 +91,9 @@ TIDY_NUX_TRIM_RE = re.compile(
),
)
# Used for attempting to acquire the schema if the URL can't be parsed.
GET_SCHEMA_RE = re.compile(r'\s*(?P<schema>[a-z0-9]{2,9})://.*$', re.I)
def is_hostname(hostname):
"""

228
test/test_glib_plugin.py Normal file
View File

@ -0,0 +1,228 @@
# -*- coding: utf-8 -*-
#
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
# All rights reserved.
#
# This code is licensed under the MIT License.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files(the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions :
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import pytest
import mock
import sys
import types
import apprise
try:
# Python v3.4+
from importlib import reload
except ImportError:
try:
# Python v3.0-v3.3
from imp import reload
except ImportError:
# Python v2.7
pass
if 'dbus' not in sys.modules:
# Environment doesn't allow for dbus
pytest.skip("Skipping dbus-python based tests", allow_module_level=True)
@mock.patch('dbus.SessionBus')
@mock.patch('dbus.Interface')
@mock.patch('dbus.ByteArray')
@mock.patch('dbus.Byte')
@mock.patch('dbus.mainloop')
def test_dbus_plugin(mock_mainloop, mock_byte, mock_bytearray,
mock_interface, mock_sessionbus):
"""
API: NotifyDBus Plugin()
"""
# 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]
reload(sys.modules['apprise.plugins.NotifyDBus'])
# We need to fake our dbus environment for testing purposes since
# the gi library isn't available in Travis CI
gi = types.ModuleType(gi_name)
gi.repository = types.ModuleType(gi_name + '.repository')
mock_pixbuf = mock.Mock()
mock_image = mock.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.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
reload(sys.modules['apprise.plugins.NotifyDBus'])
mock_mainloop.qt.DBusQtMainLoop.side_effect = None
# Python v2.x
mock_mainloop.glib.DBusGMainLoop.return_value = True
mock_mainloop.glib.DBusGMainLoop.side_effect = ImportError()
# Python 3.x
mock_mainloop.glib.NativeMainLoop.return_value = True
mock_mainloop.glib.NativeMainLoop.side_effect = ImportError()
sys.modules['dbus.mainloop.glib'] = mock_mainloop.glib
reload(sys.modules['apprise.plugins.NotifyDBus'])
mock_mainloop.glib.DBusGMainLoop.side_effect = None
mock_mainloop.glib.NativeMainLoop.side_effect = None
# The following libraries need to be reloaded to prevent
# TypeError: super(type, obj): obj must be an instance or subtype of type
# This is better explained in this StackOverflow post:
# https://stackoverflow.com/questions/31363311/\
# any-way-to-manually-fix-operation-of-\
# super-after-ipython-reload-avoiding-ty
#
reload(sys.modules['apprise.plugins.NotifyDBus'])
reload(sys.modules['apprise.plugins'])
reload(sys.modules['apprise.Apprise'])
reload(sys.modules['apprise'])
# Create our instance (identify all supported types)
obj = apprise.Apprise.instantiate('dbus://', suppress_exceptions=False)
assert(isinstance(obj, apprise.plugins.NotifyDBus) is True)
obj = apprise.Apprise.instantiate('kde://', suppress_exceptions=False)
assert(isinstance(obj, apprise.plugins.NotifyDBus) is True)
obj = apprise.Apprise.instantiate('qt://', suppress_exceptions=False)
assert(isinstance(obj, apprise.plugins.NotifyDBus) is True)
obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False)
assert(isinstance(obj, apprise.plugins.NotifyDBus) is True)
obj.duration = 0
# Check that it found our mocked environments
assert(obj._enabled is True)
# Test our class loading using a series of arguments
try:
apprise.plugins.NotifyDBus(**{'schema': 'invalid'})
# We should not reach here as the invalid schema
# should force an exception
assert(False)
except TypeError:
# Expected behaviour
assert(True)
# Invalid URLs
assert apprise.plugins.NotifyDBus.parse_url('') is None
# Set our X and Y coordinate and try the notification
assert(
apprise.plugins.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)
# If our underlining object throws for whatever reason, we will
# gracefully fail
mock_notify = mock.Mock()
mock_interface.return_value = mock_notify
mock_notify.Notify.side_effect = AttributeError()
assert(obj.notify(title='', body='body',
notify_type=apprise.NotifyType.INFO) is False)
mock_notify.Notify.side_effect = None
# Test our loading of our icon exception; it will still allow the
# notification to be sent
mock_pixbuf.new_from_file.side_effect = AttributeError()
assert(obj.notify(title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True)
# Undo our change
mock_pixbuf.new_from_file.side_effect = None
# Test our exception handling during initialization
# Toggle our testing for when we can't send notifications because the
# package has been made unavailable to us
obj._enabled = False
assert(obj.notify(title='title', body='body',
notify_type=apprise.NotifyType.INFO) is False)
# Test the setting of a the urgency
apprise.plugins.NotifyDBus(urgency=0)
#
# We can still notify if the gi library is the only inaccessible
# compontent
#
# Emulate require_version function:
gi.require_version.side_effect = ImportError()
# The following libraries need to be reloaded to prevent
# TypeError: super(type, obj): obj must be an instance or subtype of type
# This is better explained in this StackOverflow post:
# https://stackoverflow.com/questions/31363311/\
# any-way-to-manually-fix-operation-of-\
# super-after-ipython-reload-avoiding-ty
#
reload(sys.modules['apprise.plugins.NotifyDBus'])
reload(sys.modules['apprise.plugins'])
reload(sys.modules['apprise.Apprise'])
reload(sys.modules['apprise'])
# Create our instance
obj = apprise.Apprise.instantiate('glib://', suppress_exceptions=False)
assert(isinstance(obj, apprise.plugins.NotifyDBus) is True)
obj.duration = 0
# Our notification succeeds even though the gi library was not loaded
assert(obj.notify(title='title', body='body',
notify_type=apprise.NotifyType.INFO) is True)

11
tox.ini
View File

@ -15,30 +15,35 @@ commands = python -m pytest {posargs}
[testenv:py27]
deps=
dbus-python
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:py34]
deps=
dbus-python
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:py35]
deps=
dbus-python
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:py36]
deps=
dbus-python
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:py37]
deps=
dbus-python
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands =
@ -51,6 +56,12 @@ deps=
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:pypy3]
deps=
-r{toxinidir}/requirements.txt
-r{toxinidir}/dev-requirements.txt
commands = coverage run --parallel -m pytest {posargs}
[testenv:coverage-report]
deps = coverage