From e6e310c7fc7abd4feae4de05ad71c3ef505561c7 Mon Sep 17 00:00:00 2001 From: sezanzeb Date: Sun, 29 Nov 2020 01:44:14 +0100 Subject: [PATCH] testing connecting to the dbus --- keymapper/daemon.py | 37 ++++------------------------- tests/test.py | 24 ++++++++++--------- tests/testcases/daemon.py | 43 +++++++++++++++++++++++++++++++++- tests/testcases/integration.py | 2 +- 4 files changed, 61 insertions(+), 45 deletions(-) diff --git a/keymapper/daemon.py b/keymapper/daemon.py index 924d6134..5f7da07c 100644 --- a/keymapper/daemon.py +++ b/keymapper/daemon.py @@ -22,8 +22,6 @@ """Starts injecting keycodes based on the configuration.""" -import subprocess - import dbus from dbus import service import dbus.mainloop.glib @@ -34,36 +32,19 @@ from keymapper.mapping import Mapping from keymapper.config import config -def is_service_running(): - """Check if the daemon is running.""" - try: - subprocess.check_output(['pgrep', '-f', 'key-mapper-service']) - except subprocess.CalledProcessError: - return False - return True - - def get_dbus_interface(): """Get an interface to start and stop injecting keystrokes.""" - if not is_service_running(): - logger.warning( - 'The daemon "key-mapper-service" is not running, mapping keys ' - 'only works as long as the window is open.' - ) - return Daemon(autoload=False) - try: - logger.debug('Found the daemon process') bus = dbus.SessionBus() remote_object = bus.get_object('keymapper.Control', '/') interface = dbus.Interface(remote_object, 'keymapper.Interface') logger.debug('Connected to dbus') except dbus.exceptions.DBusException as error: - logger.error( + logger.warning( 'Could not connect to the dbus of "key-mapper-service", mapping ' 'keys only works as long as the window is open.' ) - logger.error(error) + logger.debug(error) return Daemon() return interface @@ -94,10 +75,7 @@ class Daemon(service.Object): logger.error(error) super().__init__(*args, **kwargs) - @dbus.service.method( - 'keymapper.Interface', - in_signature='s' - ) + @dbus.service.method('keymapper.Interface', in_signature='s') def stop_injecting(self, device): """Stop injecting the mapping for a single device.""" if self.injectors.get(device) is None: @@ -111,10 +89,7 @@ class Daemon(service.Object): # TODO if ss is the correct signature for multiple parameters, add an # example to https://gitlab.freedesktop.org/dbus/dbus-python/-/blob/master/doc/tutorial.txt # noqa pylint: disable=line-too-long - @dbus.service.method( - 'keymapper.Interface', - in_signature='ss' - ) + @dbus.service.method('keymapper.Interface', in_signature='ss') def start_injecting(self, device, preset): """Start injecting the preset for the device. @@ -141,9 +116,7 @@ class Daemon(service.Object): return True - @dbus.service.method( - 'keymapper.Interface' - ) + @dbus.service.method('keymapper.Interface') def stop(self): """Stop all mapping injections.""" for injector in self.injectors.values(): diff --git a/tests/test.py b/tests/test.py index ae183f8c..1284a498 100644 --- a/tests/test.py +++ b/tests/test.py @@ -25,6 +25,7 @@ import sys import time import unittest +import subprocess import multiprocessing import asyncio @@ -207,16 +208,6 @@ def patch_unsaved(): unsaved.unsaved_changes_dialog = lambda: unsaved.CONTINUE -def patch_dbus(): - """Make sure that the dbus interface is just an instance of Daemon. - - Don't talk to an actual daemon if one is running. - """ - import dbus - from keymapper.daemon import Daemon - dbus.Interface = lambda *args: Daemon() - - def clear_write_history(): """Empty the history in preparation for the next test.""" while len(uinput_write_history) > 0: @@ -225,12 +216,23 @@ def clear_write_history(): uinput_write_history_pipe[0].recv() +def is_service_running(): + """Check if the daemon is running.""" + try: + subprocess.check_output(['pgrep', '-f', 'key-mapper-service']) + except subprocess.CalledProcessError: + return + # let tests control daemon existance + raise Exception('Expected the service not to be running already.') + + +is_service_running() + # quickly fake some stuff before any other file gets a chance to import # the original versions patch_paths() patch_evdev() patch_unsaved() -patch_dbus() def main(): diff --git a/tests/testcases/daemon.py b/tests/testcases/daemon.py index 4086e508..4b66deeb 100644 --- a/tests/testcases/daemon.py +++ b/tests/testcases/daemon.py @@ -19,17 +19,58 @@ # along with key-mapper. If not, see . +import os +import sys +import multiprocessing import unittest +import time +from unittest.mock import patch +from importlib.util import spec_from_loader, module_from_spec +from importlib.machinery import SourceFileLoader +import dbus import evdev +import gi +gi.require_version('Gtk', '3.0') +from gi.repository import Gtk from keymapper.state import custom_mapping, system_mapping from keymapper.config import config -from keymapper.daemon import Daemon +from keymapper.daemon import Daemon, get_dbus_interface from test import uinput_write_history_pipe, Event, pending_events +def gtk_iteration(): + """Iterate while events are pending.""" + while Gtk.events_pending(): + Gtk.main_iteration() + + +class TestDBusDaemon(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.process = multiprocessing.Process( + target=os.system, + args=('key-mapper-service',) + ) + cls.process.start() + time.sleep(0.5) + cls.interface = get_dbus_interface() + + @classmethod + def tearDownClass(cls): + cls.interface.stop() + time.sleep(0.1) + cls.process.terminate() + time.sleep(0.1) + os.system('pkill -f key-mapper-service') + time.sleep(0.1) + + def test_can_connect(self): + self.assertIsInstance(self.interface, dbus.Interface) + + class TestDaemon(unittest.TestCase): def setUp(self): self.grab = evdev.InputDevice.grab diff --git a/tests/testcases/integration.py b/tests/testcases/integration.py index 6d3ef00a..529da6b0 100644 --- a/tests/testcases/integration.py +++ b/tests/testcases/integration.py @@ -47,7 +47,7 @@ def gtk_iteration(): Gtk.main_iteration() -def launch(argv=None, bin_path='bin/key-mapper-gtk'): +def launch(argv=None, bin_path='/bin/key-mapper-gtk'): """Start key-mapper-gtk with the command line argument array argv.""" if not argv: argv = ['-d']