tests for hold macros

keymapper/dev/keycode_mapper.py

@@ -82,8 +82,7 @@ def handle_keycode(code_to_code, macros, event, uinput):
             # that the key is released and let it decide what to with that
             # information.
             macro = active_macros.get(input_keycode)
-            # TODO test
-            if macro is not None:
+            if macro is not None and macro.holding:
                 macro.release_key()
 
         if event.value != 1:
@@ -95,15 +94,11 @@ def handle_keycode(code_to_code, macros, event, uinput):
             # make sure that a duplicate key-down event won't make a
             # macro with a hold function run forever. there should always
             # be only one active.
-            # TODO test, throw in a ton of key-down events of various codes
-            #  and one key up event and check that no macro is writing stuff
-            existing_macro.release_key()
-
-        # TODO test holding down two macros
+            if existing_macro.running:
+                return
 
         macro = macros[input_keycode]
         active_macros[input_keycode] = macro
-        # TODO test that holding is true
         macro.press_key()
         logger.spam(
             'got code:%s value:%s, maps to macro %s',

keymapper/dev/macros.py

@@ -81,6 +81,8 @@ class _Macro:
         # supposed to be True between key event values 1 (down) and 0 (up)
         self.holding = False
 
+        self.running = False
+
         # all required capabilities, without those of child macros
         self.capabilities = set()
 
@@ -110,29 +112,29 @@ class _Macro:
 
     async def run(self):
         """Run the macro."""
+        self.running = True
         for task_type, task in self.tasks:
             coroutine = task()
             if asyncio.iscoroutine(coroutine):
                 await coroutine
 
+        # done
+        self.running = False
+
     def press_key(self):
         """Tell all child macros that the key was pressed down."""
-        # TODO test
         self.holding = True
         for macro in self.child_macros:
             macro.press_key()
 
     def release_key(self):
         """Tell all child macros that the key was released."""
-        # TODO test
         self.holding = False
         for macro in self.child_macros:
             macro.release_key()
 
     def hold(self, macro):
         """Loops the execution until key release."""
-        # TODO test. especially make sure that this doesn't loop forever
-        #  even with complicated macros and weird calls to press and release
         if not isinstance(macro, _Macro):
             raise ValueError(
                 'Expected the param for h (hold) to be '

tests/test.py

@@ -146,14 +146,14 @@ def push_event(device, event):
     ----------
     device : string
         For example 'device 1'
-    event : Event
+    event : InputEvent
     """
     if pending_events.get(device) is None:
         pending_events[device] = []
     pending_events[device].append(event)
 
 
-class Event:
+class InputEvent:
     """Event to put into the injector for tests.
 
     fakes evdev.InputEvent
@@ -283,7 +283,7 @@ def patch_evdev():
 
         def write(self, type, code, value):
             self.write_count += 1
-            event = Event(type, code, value)
+            event = InputEvent(type, code, value)
             uinput_write_history.append(event)
             uinput_write_history_pipe[1].send(event)
 

tests/testcases/test_daemon.py

@@ -32,7 +32,7 @@ from keymapper.state import custom_mapping, system_mapping
 from keymapper.config import config
 from keymapper.daemon import Daemon, get_dbus_interface, BUS_NAME
 
-from tests.test import uinput_write_history_pipe, Event, pending_events
+from tests.test import uinput_write_history_pipe, InputEvent, pending_events
 
 
 def gtk_iteration():
@@ -95,7 +95,7 @@ class TestDaemon(unittest.TestCase):
         config.set_autoload_preset('device 2', preset)
 
         pending_events['device 2'] = [
-            Event(evdev.events.EV_KEY, keycode_from_1, 0),
+            InputEvent(evdev.events.EV_KEY, keycode_from_1, 0),
         ]
 
         self.daemon = Daemon()
@@ -113,8 +113,8 @@ class TestDaemon(unittest.TestCase):
         self.assertFalse(self.daemon.is_injecting('device 2'))
 
         pending_events['device 2'] = [
-            Event(evdev.events.EV_KEY, keycode_from_2, 1),
-            Event(evdev.events.EV_KEY, keycode_from_2, 0),
+            InputEvent(evdev.events.EV_KEY, keycode_from_2, 1),
+            InputEvent(evdev.events.EV_KEY, keycode_from_2, 0),
         ]
 
         time.sleep(0.2)

tests/testcases/test_injector.py

@@ -19,7 +19,6 @@
 # along with key-mapper.  If not, see <https://www.gnu.org/licenses/>.
 
 
-import asyncio
 import unittest
 import time
 
@@ -28,13 +27,12 @@ from evdev.ecodes import EV_REL, EV_KEY, EV_ABS, ABS_HAT0X
 
 from keymapper.dev.injector import is_numlock_on, toggle_numlock, \
     ensure_numlock, KeycodeInjector
-from keymapper.dev.keycode_mapper import handle_keycode
 from keymapper.state import custom_mapping, system_mapping
 from keymapper.mapping import Mapping
 from keymapper.config import config
 from keymapper.dev.macros import parse
 
-from tests.test import Event, pending_events, fixtures, \
+from tests.test import InputEvent, pending_events, fixtures, \
     clear_write_history, EVENT_READ_TIMEOUT, uinput_write_history_pipe, \
     MAX_ABS
 
@@ -244,10 +242,10 @@ class TestInjector(unittest.TestCase):
         x = MAX_ABS / pointer_speed / divisor
         y = MAX_ABS / pointer_speed / divisor
         pending_events['gamepad'] = [
-            Event(EV_ABS, rel_x, x),
-            Event(EV_ABS, rel_y, y),
-            Event(EV_ABS, rel_x, -x),
-            Event(EV_ABS, rel_y, -y),
+            InputEvent(EV_ABS, rel_x, x),
+            InputEvent(EV_ABS, rel_y, y),
+            InputEvent(EV_ABS, rel_x, -x),
+            InputEvent(EV_ABS, rel_y, -y),
         ]
 
         self.injector = KeycodeInjector('gamepad', custom_mapping)
@@ -285,70 +283,6 @@ class TestInjector(unittest.TestCase):
         self.assertEqual(history[-2][1], rel_y)
         self.assertAlmostEqual(history[-2][2], -1)
 
-    def test_handle_keycode(self):
-        _code_to_code = {
-            1: 101,
-            2: 102
-        }
-
-        history = []
-
-        class UInput:
-            def write(self, type, code, value):
-                history.append((type, code, value))
-
-            def syn(self):
-                pass
-
-        uinput = UInput()
-
-        EV_KEY = evdev.ecodes.EV_KEY
-
-        handle_keycode(_code_to_code, {}, Event(EV_KEY, 1, 1), uinput)
-        handle_keycode(_code_to_code, {}, Event(EV_KEY, 3, 1), uinput)
-        handle_keycode(_code_to_code, {}, Event(EV_KEY, 2, 1), uinput)
-
-        self.assertEqual(len(history), 3)
-        self.assertEqual(history[0], (EV_KEY, 101, 1))
-        self.assertEqual(history[1], (EV_KEY, 3, 1))
-        self.assertEqual(history[2], (EV_KEY, 102, 1))
-
-    def test_handle_keycode_macro(self):
-        history = []
-
-        code_a = 100
-        code_b = 101
-        system_mapping.clear()
-        system_mapping._set('a', code_a)
-        system_mapping._set('b', code_b)
-
-        macro_mapping = {
-            1: parse('k(a)'),
-            2: parse('r(5, k(b))')
-        }
-
-        macro_mapping[1].set_handler(lambda *args: history.append(args))
-        macro_mapping[2].set_handler(lambda *args: history.append(args))
-
-        handle_keycode({}, macro_mapping, Event(EV_KEY, 1, 1), None)
-        handle_keycode({}, macro_mapping, Event(EV_KEY, 2, 1), None)
-
-        loop = asyncio.get_event_loop()
-
-        sleeptime = config.get('macros.keystroke_sleep_ms', 10) * 12
-
-        async def sleep():
-            await asyncio.sleep(sleeptime / 1000 + 0.1)
-
-        loop.run_until_complete(sleep())
-
-        # 6 keycodes written, with down and up events
-        self.assertEqual(len(history), 12)
-        self.assertIn((code_a, 1), history)
-        self.assertIn((code_a, 0), history)
-        self.assertIn((code_b, 1), history)
-        self.assertIn((code_b, 0), history)
-
     def test_injector(self):
         custom_mapping.change((EV_KEY, 8), 'k(KEY_Q).k(w)')
         custom_mapping.change((EV_ABS, ABS_HAT0X), 'a')
@@ -368,15 +302,15 @@ class TestInjector(unittest.TestCase):
         # keycode used in X and in the mappings
         pending_events['device 2'] = [
             # should execute a macro
-            Event(EV_KEY, 8, 1),
-            Event(EV_KEY, 8, 0),
+            InputEvent(EV_KEY, 8, 1),
+            InputEvent(EV_KEY, 8, 0),
             # normal keystrokes
-            Event(EV_ABS, ABS_HAT0X, 1),
-            Event(EV_ABS, ABS_HAT0X, 0),
+            InputEvent(EV_ABS, ABS_HAT0X, 1),
+            InputEvent(EV_ABS, ABS_HAT0X, 0),
             # just pass those over without modifying
-            Event(EV_KEY, 10, 1),
-            Event(EV_KEY, 10, 0),
-            Event(3124, 3564, 6542),
+            InputEvent(EV_KEY, 10, 1),
+            InputEvent(EV_KEY, 10, 0),
+            InputEvent(3124, 3564, 6542),
         ]
 
         self.injector = KeycodeInjector('device 2', custom_mapping)

tests/testcases/test_integration.py

@@ -42,7 +42,7 @@ from keymapper.config import config
 from keymapper.dev.reader import keycode_reader
 from keymapper.gtk.row import to_string
 
-from tests.test import tmp, pending_events, Event, uinput_write_history_pipe, \
+from tests.test import tmp, pending_events, InputEvent, uinput_write_history_pipe, \
     clear_write_history
 
 
@@ -433,8 +433,8 @@ class TestIntegration(unittest.TestCase):
         system_mapping._set('a', keycode_to)
 
         pending_events['device 2'] = [
-            Event(evdev.events.EV_KEY, keycode_from, 1),
-            Event(evdev.events.EV_KEY, keycode_from, 0)
+            InputEvent(evdev.events.EV_KEY, keycode_from, 1),
+            InputEvent(evdev.events.EV_KEY, keycode_from, 0)
         ]
 
         custom_mapping.save('device 2', 'foo preset')
@@ -471,7 +471,7 @@ class TestIntegration(unittest.TestCase):
 
         # not all of those events should be processed, since that takes some
         # time due to time.sleep in the fakes and the injection is stopped.
-        pending_events['device 2'] = [Event(1, keycode_from, 1)] * 100
+        pending_events['device 2'] = [InputEvent(1, keycode_from, 1)] * 100
 
         custom_mapping.save('device 2', 'foo preset')
 

tests/testcases/test_keycode_mapper.py

@@ -20,21 +20,453 @@
 
 
 import unittest
+import asyncio
 
 from evdev.ecodes import EV_KEY, EV_ABS, ABS_HAT0X, KEY_A, ABS_X, EV_REL, REL_X
 
-from keymapper.dev.keycode_mapper import should_map_event_as_btn
+from keymapper.dev.keycode_mapper import should_map_event_as_btn, \
+    active_macros, handle_keycode
+from keymapper.state import system_mapping
+from keymapper.dev.macros import parse
+from keymapper.config import config
+
+from tests.test import InputEvent
 
 
 class TestKeycodeMapper(unittest.TestCase):
+    def tearDown(self):
+        # make sure all macros are stopped by tests
+        for code in active_macros:
+            macro = active_macros[code]
+            if macro.holding:
+                macro.release_key()
+            self.assertFalse(macro.holding)
+            self.assertFalse(macro.running)
+
+        keys = list(active_macros.keys())
+        for key in keys:
+            del active_macros[key]
+
     def test_should_map_event_as_btn(self):
         self.assertTrue(should_map_event_as_btn(EV_ABS, ABS_HAT0X))
         self.assertTrue(should_map_event_as_btn(EV_KEY, KEY_A))
         self.assertFalse(should_map_event_as_btn(EV_ABS, ABS_X))
         self.assertFalse(should_map_event_as_btn(EV_REL, REL_X))
 
-    # TODO test for macro holding
+    def test_handle_keycode(self):
+        _code_to_code = {
+            1: 101,
+            2: 102
+        }
+
+        history = []
+
+        class UInput:
+            def write(self, type, code, value):
+                history.append((type, code, value))
+
+            def syn(self):
+                pass
+
+        uinput = UInput()
+
+        handle_keycode(_code_to_code, {}, InputEvent(EV_KEY, 1, 1), uinput)
+        handle_keycode(_code_to_code, {}, InputEvent(EV_KEY, 3, 1), uinput)
+        handle_keycode(_code_to_code, {}, InputEvent(EV_KEY, 2, 1), uinput)
+
+        self.assertEqual(len(history), 3)
+        self.assertEqual(history[0], (EV_KEY, 101, 1))
+        self.assertEqual(history[1], (EV_KEY, 3, 1))
+        self.assertEqual(history[2], (EV_KEY, 102, 1))
+
+    def test_handle_keycode_macro(self):
+        history = []
+
+        code_a = 100
+        code_b = 101
+        system_mapping.clear()
+        system_mapping._set('a', code_a)
+        system_mapping._set('b', code_b)
+
+        macro_mapping = {
+            1: parse('k(a)'),
+            2: parse('r(5, k(b))')
+        }
+
+        macro_mapping[1].set_handler(lambda *args: history.append(args))
+        macro_mapping[2].set_handler(lambda *args: history.append(args))
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 1), None)
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 1), None)
+
+        loop = asyncio.get_event_loop()
+
+        sleeptime = config.get('macros.keystroke_sleep_ms', 10) * 12
+
+        # let the mainloop run for some time so that the macro does its stuff
+        loop.run_until_complete(asyncio.sleep(sleeptime / 1000 + 0.1))
+
+        # 6 keycodes written, with down and up events
+        self.assertEqual(len(history), 12)
+        self.assertIn((code_a, 1), history)
+        self.assertIn((code_a, 0), history)
+        self.assertIn((code_b, 1), history)
+        self.assertIn((code_b, 0), history)
+
+    def calculate_event_number(self, holdtime, before, after):
+        """
+        Parameters
+        ----------
+        holdtime : int
+            in ms, how long was the key held down
+        before : int
+            how many extra k() calls are executed before h()
+        after : int
+            how many extra k() calls are executed after h()
+        """
+        keystroke_sleep = config.get('macros.keystroke_sleep_ms', 10)
+        # down and up: two sleeps per k
+        # one initial k(a):
+        events = before * 2
+        holdtime -= keystroke_sleep * 2
+        # hold events
+        events += (holdtime / (keystroke_sleep * 2)) * 2
+        # one trailing k(c)
+        events += after * 2
+        return events
+
+    def test_hold(self):
+        history = []
+
+        code_a = 100
+        code_b = 101
+        code_c = 102
+        system_mapping.clear()
+        system_mapping._set('a', code_a)
+        system_mapping._set('b', code_b)
+        system_mapping._set('c', code_c)
+
+        macro_mapping = {
+            1: parse('k(a).h(k(b)).k(c)')
+        }
+
+        def handler(*args):
+            history.append(args)
+
+        macro_mapping[1].set_handler(handler)
+
+        """start macro"""
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 1), None)
+
+        loop = asyncio.get_event_loop()
+
+        # let the mainloop run for some time so that the macro does its stuff
+        sleeptime = 500
+        keystroke_sleep = config.get('macros.keystroke_sleep_ms', 10)
+        loop.run_until_complete(asyncio.sleep(sleeptime / 1000))
+
+        self.assertTrue(active_macros[1].holding)
+        self.assertTrue(active_macros[1].running)
+
+        """stop macro"""
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 0), None)
+
+        loop.run_until_complete(asyncio.sleep(keystroke_sleep * 10 / 1000))
+
+        events = self.calculate_event_number(sleeptime, 1, 1)
+
+        self.assertGreater(len(history), events * 0.9)
+        self.assertLess(len(history), events * 1.1)
+
+        self.assertIn((code_a, 1), history)
+        self.assertIn((code_a, 0), history)
+        self.assertIn((code_b, 1), history)
+        self.assertIn((code_b, 0), history)
+        self.assertIn((code_c, 1), history)
+        self.assertIn((code_c, 0), history)
+        self.assertGreater(history.count((code_b, 1)), 1)
+        self.assertGreater(history.count((code_b, 0)), 1)
+
+        # it's stopped and won't write stuff anymore
+        count_before = len(history)
+        loop.run_until_complete(asyncio.sleep(0.2))
+        count_after = len(history)
+        self.assertEqual(count_before, count_after)
+
+        self.assertFalse(active_macros[1].holding)
+        self.assertFalse(active_macros[1].running)
+
+    def test_hold_2(self):
+        # test irregular input patterns
+        code_a = 100
+        code_b = 101
+        code_c = 102
+        code_d = 103
+        system_mapping.clear()
+        system_mapping._set('a', code_a)
+        system_mapping._set('b', code_b)
+        system_mapping._set('c', code_c)
+        system_mapping._set('d', code_d)
+
+        macro_mapping = {
+            1: parse('h(k(b))'),
+            2: parse('k(c).r(1, r(1, r(1, h(k(a))))).k(d)'),
+            3: parse('h(k(b))')
+        }
+
+        history = []
+
+        def handler(*args):
+            history.append(args)
+
+        macro_mapping[1].set_handler(handler)
+        macro_mapping[2].set_handler(handler)
+        macro_mapping[3].set_handler(handler)
+
+        """start macro 2"""
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 1), None)
+        loop = asyncio.get_event_loop()
+
+        loop.run_until_complete(asyncio.sleep(0.1))
+        # starting code_c written
+        self.assertEqual(history.count((code_c, 1)), 1)
+        self.assertEqual(history.count((code_c, 0)), 1)
+
+        # spam garbage events
+        for _ in range(5):
+            handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 1), None)
+            handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 3, 1), None)
+            loop.run_until_complete(asyncio.sleep(0.05))
+            self.assertTrue(active_macros[1].holding)
+            self.assertTrue(active_macros[1].running)
+            self.assertTrue(active_macros[2].holding)
+            self.assertTrue(active_macros[2].running)
+            self.assertTrue(active_macros[3].holding)
+            self.assertTrue(active_macros[3].running)
+
+        # there should only be one code_c in the events, because no key
+        # up event was ever done so the hold just continued
+        self.assertEqual(history.count((code_c, 1)), 1)
+        self.assertEqual(history.count((code_c, 0)), 1)
+        # without an key up event on 2, it won't write code_d
+        self.assertNotIn((code_d, 1), history)
+        self.assertNotIn((code_d, 0), history)
+
+        # stop macro 2
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 0), None)
+        loop.run_until_complete(asyncio.sleep(0.1))
+
+        # it stopped and didn't restart, so the count stays at 1
+        self.assertEqual(history.count((code_c, 1)), 1)
+        self.assertEqual(history.count((code_c, 0)), 1)
+        # and the trailing d was written
+        self.assertEqual(history.count((code_d, 1)), 1)
+        self.assertEqual(history.count((code_d, 0)), 1)
+
+        # it's stopped and won't write stuff anymore
+        count_before = history.count((code_a, 1))
+        self.assertGreater(count_before, 1)
+        loop.run_until_complete(asyncio.sleep(0.1))
+        count_after = history.count((code_a, 1))
+        self.assertEqual(count_before, count_after)
+
+        """restart macro 2"""
+
+        history = []
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 1), None)
+        loop.run_until_complete(asyncio.sleep(0.1))
+        self.assertEqual(history.count((code_c, 1)), 1)
+        self.assertEqual(history.count((code_c, 0)), 1)
+
+        # spam garbage events again, this time key-up events on all other
+        # macros
+        for _ in range(5):
+            handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 0), None)
+            handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 3, 0), None)
+            loop.run_until_complete(asyncio.sleep(0.05))
+            self.assertFalse(active_macros[1].holding)
+            self.assertFalse(active_macros[1].running)
+            self.assertTrue(active_macros[2].holding)
+            self.assertTrue(active_macros[2].running)
+            self.assertFalse(active_macros[3].holding)
+            self.assertFalse(active_macros[3].running)
+
+        # stop macro 2
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 0), None)
+        loop.run_until_complete(asyncio.sleep(0.1))
+        # was started only once
+        self.assertEqual(history.count((code_c, 1)), 1)
+        self.assertEqual(history.count((code_c, 0)), 1)
+        # and the trailing d was also written only once
+        self.assertEqual(history.count((code_d, 1)), 1)
+        self.assertEqual(history.count((code_d, 0)), 1)
+
+        # stop all macros
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 0), None)
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 3, 0), None)
+        loop.run_until_complete(asyncio.sleep(0.1))
+
+        # it's stopped and won't write stuff anymore
+        count_before = len(history)
+        loop.run_until_complete(asyncio.sleep(0.1))
+        count_after = len(history)
+        self.assertEqual(count_before, count_after)
+
+        self.assertFalse(active_macros[1].holding)
+        self.assertFalse(active_macros[1].running)
+        self.assertFalse(active_macros[2].holding)
+        self.assertFalse(active_macros[2].running)
+        self.assertFalse(active_macros[3].holding)
+        self.assertFalse(active_macros[3].running)
+
+    def test_hold_3(self):
+        # test irregular input patterns
+        code_a = 100
+        code_b = 101
+        code_c = 102
+        system_mapping.clear()
+        system_mapping._set('a', code_a)
+        system_mapping._set('b', code_b)
+        system_mapping._set('c', code_c)
+
+        macro_mapping = {
+            1: parse('k(a).h(k(b)).k(c)'),
+        }
+
+        history = []
+
+        def handler(*args):
+            history.append(args)
+
+        macro_mapping[1].set_handler(handler)
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 1), None)
+        loop = asyncio.get_event_loop()
+
+        loop.run_until_complete(asyncio.sleep(0.1))
+        for _ in range(5):
+            self.assertTrue(active_macros[1].holding)
+            self.assertTrue(active_macros[1].running)
+            handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 1), None)
+            loop.run_until_complete(asyncio.sleep(0.05))
+
+        # duplicate key down events don't do anything
+        self.assertEqual(history.count((code_a, 1)), 1)
+        self.assertEqual(history.count((code_a, 0)), 1)
+        self.assertEqual(history.count((code_c, 1)), 0)
+        self.assertEqual(history.count((code_c, 0)), 0)
+
+        # stop
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 0), None)
+        loop.run_until_complete(asyncio.sleep(0.1))
+        self.assertEqual(history.count((code_a, 1)), 1)
+        self.assertEqual(history.count((code_a, 0)), 1)
+        self.assertEqual(history.count((code_c, 1)), 1)
+        self.assertEqual(history.count((code_c, 0)), 1)
+        self.assertFalse(active_macros[1].holding)
+        self.assertFalse(active_macros[1].running)
+
+        # it's stopped and won't write stuff anymore
+        count_before = len(history)
+        loop.run_until_complete(asyncio.sleep(0.1))
+        count_after = len(history)
+        self.assertEqual(count_before, count_after)
+
+    def test_hold_two(self):
+        history = []
+
+        code_1 = 100
+        code_2 = 101
+        code_3 = 102
+        code_a = 103
+        code_b = 104
+        code_c = 105
+        system_mapping.clear()
+        system_mapping._set('1', code_1)
+        system_mapping._set('2', code_2)
+        system_mapping._set('3', code_3)
+        system_mapping._set('a', code_a)
+        system_mapping._set('b', code_b)
+        system_mapping._set('c', code_c)
+
+        macro_mapping = {
+            1: parse('k(1).h(k(2)).k(3)'),
+            2: parse('k(a).h(k(b)).k(c)')
+        }
+
+        def handler(*args):
+            history.append(args)
+
+        macro_mapping[1].set_handler(handler)
+        macro_mapping[2].set_handler(handler)
+
+        loop = asyncio.get_event_loop()
+
+        # key up won't do anything
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 0), None)
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 0), None)
+        loop.run_until_complete(asyncio.sleep(0.1))
+        self.assertEqual(len(active_macros), 0)
+
+        """start macros"""
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 1), None)
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 1), None)
+
+        # let the mainloop run for some time so that the macro does its stuff
+        sleeptime = 500
+        keystroke_sleep = config.get('macros.keystroke_sleep_ms', 10)
+        loop.run_until_complete(asyncio.sleep(sleeptime / 1000))
+
+        self.assertEqual(len(active_macros), 2)
+        self.assertTrue(active_macros[1].holding)
+        self.assertTrue(active_macros[1].running)
+        self.assertTrue(active_macros[2].holding)
+        self.assertTrue(active_macros[2].running)
+
+        """stop macros"""
+
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 1, 0), None)
+        handle_keycode({}, macro_mapping, InputEvent(EV_KEY, 2, 0), None)
+
+        loop.run_until_complete(asyncio.sleep(keystroke_sleep * 10 / 1000))
+
+        self.assertFalse(active_macros[1].holding)
+        self.assertFalse(active_macros[1].running)
+        self.assertFalse(active_macros[2].holding)
+        self.assertFalse(active_macros[2].running)
+
+        events = self.calculate_event_number(sleeptime, 1, 1) * 2
+
+        self.assertGreater(len(history), events * 0.9)
+        self.assertLess(len(history), events * 1.1)
+
+        self.assertIn((code_a, 1), history)
+        self.assertIn((code_a, 0), history)
+        self.assertIn((code_b, 1), history)
+        self.assertIn((code_b, 0), history)
+        self.assertIn((code_c, 1), history)
+        self.assertIn((code_c, 0), history)
+        self.assertIn((code_1, 1), history)
+        self.assertIn((code_1, 0), history)
+        self.assertIn((code_2, 1), history)
+        self.assertIn((code_2, 0), history)
+        self.assertIn((code_3, 1), history)
+        self.assertIn((code_3, 0), history)
+        self.assertGreater(history.count((code_b, 1)), 1)
+        self.assertGreater(history.count((code_b, 0)), 1)
+        self.assertGreater(history.count((code_2, 1)), 1)
+        self.assertGreater(history.count((code_2, 0)), 1)
 
+        # it's stopped and won't write stuff anymore
+        count_before = len(history)
+        loop.run_until_complete(asyncio.sleep(0.2))
+        count_after = len(history)
+        self.assertEqual(count_before, count_after)
 
 
 if __name__ == "__main__":

tests/testcases/test_permissions.py

@@ -149,6 +149,8 @@ class TestCheckGroup(unittest.TestCase):
 
         # groups command doesn't exist, so cannot check this suff
         self.assertIsNone(check_group('plugdev'))
+        # which doesn't affect the grp lib
+        self.assertIsNotNone(check_group('foobar'))
 
 
 if __name__ == "__main__":

tests/testcases/test_reader.py

@@ -26,7 +26,7 @@ import time
 
 from keymapper.dev.reader import keycode_reader
 
-from tests.test import Event, pending_events, EVENT_READ_TIMEOUT
+from tests.test import InputEvent, pending_events, EVENT_READ_TIMEOUT
 
 
 CODE_1 = 100
@@ -58,9 +58,9 @@ class TestReader(unittest.TestCase):
 
     def test_reading_1(self):
         pending_events['device 1'] = [
-            Event(EV_KEY, CODE_1, 1),
-            Event(EV_ABS, ABS_HAT0X, 1),
-            Event(EV_KEY, CODE_3, 1)
+            InputEvent(EV_KEY, CODE_1, 1),
+            InputEvent(EV_ABS, ABS_HAT0X, 1),
+            InputEvent(EV_KEY, CODE_3, 1)
         ]
         keycode_reader.start_reading('device 1')
 
@@ -73,7 +73,7 @@ class TestReader(unittest.TestCase):
         self.assertEqual(keycode_reader.read(), (None, None))
 
     def test_reading_2(self):
-        pending_events['device 1'] = [Event(EV_ABS, ABS_HAT0X, 1)]
+        pending_events['device 1'] = [InputEvent(EV_ABS, ABS_HAT0X, 1)]
         keycode_reader.start_reading('device 1')
         wait(keycode_reader._pipe[0].poll, 0.5)
         self.assertEqual(keycode_reader.read(), (EV_ABS, ABS_HAT0X))
@@ -81,9 +81,9 @@ class TestReader(unittest.TestCase):
 
     def test_wrong_device(self):
         pending_events['device 1'] = [
-            Event(EV_KEY, CODE_1, 1),
-            Event(EV_KEY, CODE_2, 1),
-            Event(EV_KEY, CODE_3, 1)
+            InputEvent(EV_KEY, CODE_1, 1),
+            InputEvent(EV_KEY, CODE_2, 1),
+            InputEvent(EV_KEY, CODE_3, 1)
         ]
         keycode_reader.start_reading('device 2')
         time.sleep(EVENT_READ_TIMEOUT * 5)
@@ -95,9 +95,9 @@ class TestReader(unittest.TestCase):
         # intentionally programmed it won't even do that. But it was at some
         # point.
         pending_events['key-mapper device 2'] = [
-            Event(EV_KEY, CODE_1, 1),
-            Event(EV_KEY, CODE_2, 1),
-            Event(EV_KEY, CODE_3, 1)
+            InputEvent(EV_KEY, CODE_1, 1),
+            InputEvent(EV_KEY, CODE_2, 1),
+            InputEvent(EV_KEY, CODE_3, 1)
         ]
         keycode_reader.start_reading('device 2')
         time.sleep(EVENT_READ_TIMEOUT * 5)
@@ -105,9 +105,9 @@ class TestReader(unittest.TestCase):
 
     def test_clear(self):
         pending_events['device 1'] = [
-            Event(EV_KEY, CODE_1, 1),
-            Event(EV_KEY, CODE_2, 1),
-            Event(EV_KEY, CODE_3, 1)
+            InputEvent(EV_KEY, CODE_1, 1),
+            InputEvent(EV_KEY, CODE_2, 1),
+            InputEvent(EV_KEY, CODE_3, 1)
         ]
         keycode_reader.start_reading('device 1')
         time.sleep(EVENT_READ_TIMEOUT * 5)
@@ -115,8 +115,8 @@ class TestReader(unittest.TestCase):
         self.assertEqual(keycode_reader.read(), (None, None))
 
     def test_switch_device(self):
-        pending_events['device 2'] = [Event(EV_KEY, CODE_1, 1)]
-        pending_events['device 1'] = [Event(EV_KEY, CODE_3, 1)]
+        pending_events['device 2'] = [InputEvent(EV_KEY, CODE_1, 1)]
+        pending_events['device 1'] = [InputEvent(EV_KEY, CODE_3, 1)]
 
         keycode_reader.start_reading('device 2')
         time.sleep(EVENT_READ_TIMEOUT * 5)