tests for KeycodeInjector

keymapper/injector.py

@@ -22,8 +22,9 @@
 """Device and evdev stuff that is independent from the display server."""
 
 
-import subprocess
-import time
+# By using processes instead of threads, the mappings are
+# automatically copied, so that they can be worked with in the ui
+# without breaking the device. And it's possible to terminate processes.
 import multiprocessing
 import asyncio
 
@@ -38,29 +39,13 @@ DEV_NAME = 'key-mapper'
 DEVICE_CREATED = 1
 
 
-def can_grab(path):
-    """Can input events from the device be read?
-
-    Parameters
-    ----------
-    path : string
-        Path in dev, for example '/dev/input/event7'
-    """
-    p = subprocess.run(['fuser', '-v', path])
-    return p.returncode == 1
-
-
 class KeycodeInjector:
     """Keeps injecting keycodes in the background based on the mapping."""
     def __init__(self, device):
+        """Start injecting keycodes based on custom_mapping."""
         self.device = device
         self.virtual_devices = []
         self.processes = []
-        self.start_injecting()
-
-    def start_injecting(self):
-        """Read keycodes and inject the mapped character forever."""
-        self.stop_injecting()
 
         paths = get_devices()[self.device]['paths']
 
@@ -75,7 +60,7 @@ class KeycodeInjector:
             pipe = multiprocessing.Pipe()
             worker = multiprocessing.Process(
                 target=self._start_injecting_worker,
-                args=(path, custom_mapping, pipe[1])
+                args=(path, pipe[1])
             )
             worker.start()
             # wait for the process to notify creation of the new injection
@@ -95,7 +80,7 @@ class KeycodeInjector:
                 process.terminate()
                 self.processes[i] = None
 
-    def _start_injecting_worker(self, path, mapping, pipe):
+    def _start_injecting_worker(self, path, pipe):
         """Inject keycodes for one of the virtual devices."""
         # TODO test
         loop = asyncio.new_event_loop()
@@ -153,7 +138,7 @@ class KeycodeInjector:
             # than the ones reported by xev and that X expects
             input_keycode = event.code + 8
 
-            character = mapping.get_character(input_keycode)
+            character = custom_mapping.get_character(input_keycode)
 
             if character is None:
                 # unknown keycode, forward it

keymapper/mapping.py

@@ -47,17 +47,13 @@ class Mapping:
     def __len__(self):
         return len(self._mapping)
 
-    def change(self, previous_keycode, new_keycode, character):
+    def change(self, new_keycode, character, previous_keycode=None):
         """Replace the mapping of a keycode with a different one.
 
         Return True on success.
 
         Parameters
         ----------
-        previous_keycode : int or None
-            If None, will not remove any previous mapping. If you recently
-            used 10 for new_keycode and want to overwrite that with 11,
-            provide 5 here.
         new_keycode : int
             The source keycode, what the mouse would report without any
             modification.
@@ -65,6 +61,10 @@ class Mapping:
             A single character known to xkb, Examples: KP_1, Shift_L, a, B.
             Can also be an array, which is used for reading the xkbmap output
             completely.
+        previous_keycode : int or None
+            If None, will not remove any previous mapping. If you recently
+            used 10 for new_keycode and want to overwrite that with 11,
+            provide 5 here.
         """
         try:
             new_keycode = int(new_keycode)

tests/test.py

@@ -23,30 +23,32 @@
 
 
 import sys
-import time
 import unittest
 from keymapper.logger import update_verbosity
 
 
 tmp = '/tmp/key-mapper-test'
 uinput_write_history = []
-fake_event_queue = []
+pending_events = {}
 
 
-input_device_events = {}
+def get_events():
+    """Get all events written by the injector."""
+    return uinput_write_history
 
 
 def push_event(device, event):
-    """Emit a fake event for the device.
+    """Emit a fake event for a device.
 
     Parameters
     ----------
     device : string
+        For example 'device 1'
     event : Event
     """
-    if input_device_events.get(device) is None:
-        input_device_events[device] = []
-    input_device_events[device].append(event)
+    if pending_events.get(device) is None:
+        pending_events[device] = []
+    pending_events[device].append(event)
 
 
 class Event:
@@ -134,18 +136,23 @@ def patch_evdev():
             pass
 
         def read_loop(self):
-            while True:
-                pending_events = input_device_events.get(self.name)
-                if pending_events is not None and len(pending_events) > 0:
-                    yield pending_events.pop(0)
-                time.sleep(0.01)
+            """Read all prepared events at once."""
+            if pending_events.get(self.name) is None:
+                return
 
-        def capabilities(self):
+            while len(pending_events[self.name]) > 0:
+                yield pending_events[self.name].pop(0)
+
+        def capabilities(self, absinfo=True):
             return fixtures[self.path]['capabilities']
 
     class UInput:
-        def write(self, *args):
-            uinput_write_history.append(args)
+        def __init__(self, *args, **kwargs):
+            self.fd = 0
+            pass
+
+        def write(self, type, code, value):
+            uinput_write_history.append(Event(type, code, value))
 
         def syn(self):
             pass

tests/testcases/getdevices.py

@@ -24,7 +24,7 @@ import unittest
 from keymapper.getdevices import _GetDevicesProcess
 
 
-class TestLinux(unittest.TestCase):
+class TestGetDevices(unittest.TestCase):
     def test_get_devices(self):
         class FakePipe:
             devices = None

tests/testcases/injector.py

@@ -0,0 +1,87 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# key-mapper - GUI for device specific keyboard mappings
+# Copyright (C) 2020 sezanzeb <proxima@hip70890b.de>
+#
+# This file is part of key-mapper.
+#
+# key-mapper is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# key-mapper is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with key-mapper.  If not, see <https://www.gnu.org/licenses/>.
+
+
+import unittest
+
+import evdev
+
+from keymapper.injector import KeycodeInjector
+from keymapper.getdevices import get_devices
+from keymapper.state import custom_mapping, system_mapping
+
+from test import uinput_write_history, Event, pending_events
+
+
+class TestInjector(unittest.TestCase):
+    def setUp(self):
+        self.injector = None
+
+    def tearDown(self):
+        if self.injector is not None:
+            self.injector.stop_injecting()
+
+    def test_injector(self):
+        device = get_devices()['device 2']
+
+        custom_mapping.change(9, 'a')
+        # one mapping that is unknown in the system_mapping on purpose
+        custom_mapping.change(10, 'b')
+
+        system_mapping.empty()
+        system_mapping.change(100, 'a')
+
+        # the second arg of those event objects is 8 lower than the
+        # keycode used in X and in the mappings
+        pending_events['device 2'] = [
+            Event(evdev.events.EV_KEY, 1, 0),
+            Event(evdev.events.EV_KEY, 1, 1),
+            # ignored because unknown to the system
+            Event(evdev.events.EV_KEY, 2, 0),
+            Event(evdev.events.EV_KEY, 2, 1),
+            # just pass those over without modifying
+            Event(3124, 3564, 6542),
+        ]
+
+        class FakePipe:
+            def send(self, message):
+                pass
+
+        self.injector = KeycodeInjector('device 2')
+        self.injector._start_injecting_worker(
+            path=device['paths'][0],
+            pipe=FakePipe()
+        )
+
+        self.assertEqual(uinput_write_history[0].type, evdev.events.EV_KEY)
+        self.assertEqual(uinput_write_history[0].code, 92)
+        self.assertEqual(uinput_write_history[0].value, 0)
+
+        self.assertEqual(uinput_write_history[1].type, evdev.events.EV_KEY)
+        self.assertEqual(uinput_write_history[1].code, 92)
+        self.assertEqual(uinput_write_history[1].value, 1)
+
+        self.assertEqual(uinput_write_history[2].type, 3124)
+        self.assertEqual(uinput_write_history[2].code, 3564)
+        self.assertEqual(uinput_write_history[2].value, 6542)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/testcases/integration.py

@@ -100,7 +100,7 @@ class Integration(unittest.TestCase):
         rows = len(self.window.get('key_list').get_children())
         self.assertEqual(rows, 1)
 
-        custom_mapping.change(None, 13, 'a')
+        custom_mapping.change(13, 'a', None)
         time.sleep(0.2)
         gtk_iteration()
 
@@ -180,12 +180,12 @@ class Integration(unittest.TestCase):
         self.assertTrue(custom_mapping.changed)
 
     def test_rename_and_save(self):
-        custom_mapping.change(None, 14, 'a')
+        custom_mapping.change(14, 'a', None)
         self.assertEqual(self.window.selected_preset, 'new preset')
         self.window.on_save_preset_clicked(None)
         self.assertEqual(custom_mapping.get_character(14), 'a')
 
-        custom_mapping.change(None, 14, 'b')
+        custom_mapping.change(14, 'b', None)
         self.window.get('preset_name_input').set_text('asdf')
         self.window.on_save_preset_clicked(None)
         self.assertEqual(self.window.selected_preset, 'asdf')
@@ -230,7 +230,7 @@ class Integration(unittest.TestCase):
         gtk_iteration()
         self.assertEqual(self.window.selected_preset, 'new preset')
         self.assertFalse(os.path.exists(f'{CONFIG}/device 1/abc 123'))
-        custom_mapping.change(None, 10, '1')
+        custom_mapping.change(10, '1', None)
         self.window.on_save_preset_clicked(None)
         gtk_iteration()
         self.assertEqual(self.window.selected_preset, 'abc 123')

tests/testcases/mapping.py

@@ -31,7 +31,7 @@ class TestMapping(unittest.TestCase):
 
     def test_change(self):
         # 1 is not assigned yet, ignore it
-        self.mapping.change(1, 2, 'a')
+        self.mapping.change(2, 'a', 1)
         self.assertTrue(self.mapping.changed)
         self.assertIsNone(self.mapping.get_character(1))
         self.assertEqual(self.mapping.get_character(2), 'a')
@@ -39,39 +39,39 @@ class TestMapping(unittest.TestCase):
         self.assertEqual(len(self.mapping), 1)
 
         # change 2 to 3 and change a to b
-        self.mapping.change(2, 3, 'b')
+        self.mapping.change(3, 'b', 2)
         self.assertIsNone(self.mapping.get_character(2))
         self.assertEqual(self.mapping.get_character(3), 'b')
         self.assertEqual(self.mapping.get_keycode('b'), 3)
         self.assertEqual(len(self.mapping), 1)
 
         # add 4
-        self.mapping.change(None, 4, 'c')
+        self.mapping.change(4, 'c', None)
         self.assertEqual(self.mapping.get_character(3), 'b')
         self.assertEqual(self.mapping.get_character(4), 'c')
         self.assertEqual(self.mapping.get_keycode('c'), 4)
         self.assertEqual(len(self.mapping), 2)
 
         # change the mapping of 4 to d
-        self.mapping.change(None, 4, 'd')
+        self.mapping.change(4, 'd', None)
         self.assertEqual(self.mapping.get_character(4), 'd')
         self.assertEqual(self.mapping.get_keycode('d'), 4)
         self.assertEqual(len(self.mapping), 2)
 
         # this also works in the same way
-        self.mapping.change(4, 4, 'e')
+        self.mapping.change(4, 'e', 4)
         self.assertEqual(self.mapping.get_character(4), 'e')
         self.assertEqual(self.mapping.get_keycode('e'), 4)
         self.assertEqual(len(self.mapping), 2)
 
         # and this
-        self.mapping.change('4', '4', 'f')
+        self.mapping.change('4', 'f', '4')
         self.assertEqual(self.mapping.get_character(4), 'f')
         self.assertEqual(self.mapping.get_keycode('f'), 4)
         self.assertEqual(len(self.mapping), 2)
 
         # non-int keycodes are ignored
-        self.mapping.change('a', 'b', 'c')
+        self.mapping.change('b', 'c', 'a')
         self.assertEqual(len(self.mapping), 2)
 
     def test_clear(self):
@@ -83,10 +83,10 @@ class TestMapping(unittest.TestCase):
         self.mapping.clear(40)
         self.assertTrue(self.mapping.changed)
 
-        self.mapping.change(None, 10, 'KP_1')
+        self.mapping.change(10, 'KP_1', None)
         self.assertTrue(self.mapping.changed)
-        self.mapping.change(None, 20, 'KP_2')
-        self.mapping.change(None, 30, 'KP_3')
+        self.mapping.change(20, 'KP_2', None)
+        self.mapping.change(30, 'KP_3', None)
         self.mapping.clear(20)
         self.assertEqual(self.mapping.get_character(10), 'KP_1')
         self.assertIsNone(self.mapping.get_character(20))