testing connecting to the dbus

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():

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():

tests/testcases/daemon.py

@@ -19,17 +19,58 @@
 # along with key-mapper.  If not, see <https://www.gnu.org/licenses/>.
 
 
+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

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']