input-remapper/tests/unit/test_daemon.py
2022-08-28 12:38:40 +02:00

483 lines
17 KiB
Python

#!/usr/bin/python3
# -*- coding: utf-8 -*-
# input-remapper - GUI for device specific keyboard mappings
# Copyright (C) 2022 sezanzeb <proxima@sezanzeb.de>
#
# This file is part of input-remapper.
#
# input-remapper 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.
#
# input-remapper 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 input-remapper. If not, see <https://www.gnu.org/licenses/>.
from tests.test import (
cleanup,
uinput_write_history_pipe,
new_event,
push_events,
is_service_running,
fixtures,
tmp,
get_key_mapping,
)
import os
import unittest
import time
import subprocess
import json
import evdev
from evdev.ecodes import EV_KEY, EV_ABS, KEY_B, KEY_A, ABS_X, BTN_A, BTN_B
from pydbus import SystemBus
from inputremapper.configs.system_mapping import system_mapping
from inputremapper.configs.global_config import global_config
from inputremapper.groups import groups
from inputremapper.configs.paths import get_config_path, mkdir, get_preset_path
from inputremapper.event_combination import EventCombination
from inputremapper.configs.preset import Preset
from inputremapper.injection.injector import STARTING, RUNNING, STOPPED, UNKNOWN
from inputremapper.daemon import Daemon
from inputremapper.injection.global_uinputs import global_uinputs
check_output = subprocess.check_output
os_system = os.system
dbus_get = type(SystemBus()).get
class TestDaemon(unittest.TestCase):
new_fixture_path = "/dev/input/event9876"
def setUp(self):
self.grab = evdev.InputDevice.grab
self.daemon = None
mkdir(get_config_path())
global_config._save_config()
# the daemon should be able to create them on demand:
global_uinputs.devices = {}
global_uinputs.is_service = True
def tearDown(self):
# avoid race conditions with other tests, daemon may run processes
if self.daemon is not None:
self.daemon.stop_all()
self.daemon = None
evdev.InputDevice.grab = self.grab
subprocess.check_output = check_output
os.system = os_system
type(SystemBus()).get = dbus_get
cleanup()
def test_connect(self):
os_system_history = []
os.system = os_system_history.append
self.assertFalse(is_service_running())
# no daemon runs, should try to run it via pkexec instead.
# It fails due to the patch on os.system and therefore exits the process
self.assertRaises(SystemExit, Daemon.connect)
self.assertEqual(len(os_system_history), 1)
self.assertIsNone(Daemon.connect(False))
# make the connect command work this time by acting like a connection is
# available:
set_config_dir_callcount = 0
class FakeConnection:
def set_config_dir(self, *args, **kwargs):
nonlocal set_config_dir_callcount
set_config_dir_callcount += 1
type(SystemBus()).get = lambda *args, **kwargs: FakeConnection()
self.assertIsInstance(Daemon.connect(), FakeConnection)
self.assertEqual(set_config_dir_callcount, 1)
self.assertIsInstance(Daemon.connect(False), FakeConnection)
self.assertEqual(set_config_dir_callcount, 2)
def test_daemon(self):
# remove the existing system mapping to force our own into it
if os.path.exists(get_config_path("xmodmap.json")):
os.remove(get_config_path("xmodmap.json"))
preset_name = "foo"
ev_1 = (EV_KEY, BTN_A)
ev_2 = (EV_ABS, ABS_X)
group = groups.find(name="gamepad")
# unrelated group that shouldn't be affected at all
group2 = groups.find(name="Bar Device")
preset = Preset(group.get_preset_path(preset_name))
preset.add(get_key_mapping(EventCombination([*ev_1, 1]), "keyboard", "a"))
preset.add(get_key_mapping(EventCombination([*ev_2, -1]), "keyboard", "b"))
preset.save()
global_config.set_autoload_preset(group.key, preset_name)
"""Injection 1"""
# should forward the event unchanged
push_events(fixtures.gamepad, [new_event(EV_KEY, BTN_B, 1)])
self.daemon = Daemon()
self.assertFalse(uinput_write_history_pipe[0].poll())
# has been cleanedUp in setUp
self.assertNotIn("keyboard", global_uinputs.devices)
self.daemon.start_injecting(group.key, preset_name)
# created on demand
self.assertIn("keyboard", global_uinputs.devices)
self.assertNotIn("gamepad", global_uinputs.devices)
self.assertEqual(self.daemon.get_state(group.key), STARTING)
self.assertEqual(self.daemon.get_state(group2.key), UNKNOWN)
event = uinput_write_history_pipe[0].recv()
self.assertEqual(self.daemon.get_state(group.key), RUNNING)
self.assertEqual(event.type, EV_KEY)
self.assertEqual(event.code, BTN_B)
self.assertEqual(event.value, 1)
self.daemon.stop_injecting(group.key)
time.sleep(0.2)
self.assertEqual(self.daemon.get_state(group.key), STOPPED)
try:
self.assertFalse(uinput_write_history_pipe[0].poll())
except AssertionError:
print("Unexpected", uinput_write_history_pipe[0].recv())
# possibly a duplicate write!
raise
"""Injection 2"""
self.daemon.start_injecting(group.key, preset_name)
time.sleep(0.1)
# -1234 will be classified as -1 by the injector
push_events(fixtures.gamepad, [new_event(*ev_2, -1234)])
time.sleep(0.1)
self.assertTrue(uinput_write_history_pipe[0].poll())
# the written key is a key-down event, not the original
# event value of -1234
event = uinput_write_history_pipe[0].recv()
self.assertEqual(event.type, EV_KEY)
self.assertEqual(event.code, KEY_B)
self.assertEqual(event.value, 1)
def test_config_dir(self):
global_config.set("foo", "bar")
self.assertEqual(global_config.get("foo"), "bar")
# freshly loads the config and therefore removes the previosly added key.
# This is important so that if the service is started via sudo or pkexec
# it knows where to look for configuration files.
self.daemon = Daemon()
self.assertEqual(self.daemon.config_dir, get_config_path())
self.assertIsNone(global_config.get("foo"))
def test_refresh_on_start(self):
if os.path.exists(get_config_path("xmodmap.json")):
os.remove(get_config_path("xmodmap.json"))
preset_name = "foo"
ev = (EV_KEY, 9)
group_name = "9876 name"
# expected key of the group
group_key = group_name
group = groups.find(name=group_name)
# this test only makes sense if this device is unknown yet
self.assertIsNone(group)
system_mapping.clear()
system_mapping._set("a", KEY_A)
preset = Preset(get_preset_path(group_name, preset_name))
preset.add(get_key_mapping(EventCombination([*ev, 1]), "keyboard", "a"))
# make the daemon load the file instead
with open(get_config_path("xmodmap.json"), "w") as file:
json.dump(system_mapping._mapping, file, indent=4)
system_mapping.clear()
preset.save()
global_config.set_autoload_preset(group_key, preset_name)
self.daemon = Daemon()
# make sure the devices are populated
groups.refresh()
# the daemon is supposed to find this device by calling refresh
fixtures[self.new_fixture_path] = {
"capabilities": {evdev.ecodes.EV_KEY: [ev[1]]},
"phys": "9876 phys",
"info": evdev.device.DeviceInfo(4, 5, 6, 7),
"name": group_name,
}
push_events(fixtures[self.new_fixture_path], [new_event(*ev, 1)])
self.daemon.start_injecting(group_key, preset_name)
# test if the injector called groups.refresh successfully
group = groups.find(key=group_key)
self.assertEqual(group.name, group_name)
self.assertEqual(group.key, group_key)
time.sleep(0.1)
self.assertTrue(uinput_write_history_pipe[0].poll())
event = uinput_write_history_pipe[0].recv()
self.assertEqual(event.t, (EV_KEY, KEY_A, 1))
self.daemon.stop_injecting(group_key)
time.sleep(0.2)
self.assertEqual(self.daemon.get_state(group_key), STOPPED)
def test_refresh_for_unknown_key(self):
device = "9876 name"
# this test only makes sense if this device is unknown yet
self.assertIsNone(groups.find(name=device))
self.daemon = Daemon()
# make sure the devices are populated
groups.refresh()
self.daemon.refresh()
fixtures[self.new_fixture_path] = {
"capabilities": {evdev.ecodes.EV_KEY: [evdev.ecodes.KEY_A]},
"phys": "9876 phys",
"info": evdev.device.DeviceInfo(4, 5, 6, 7),
"name": device,
}
self.daemon._autoload("25v7j9q4vtj")
# this is unknown, so the daemon will scan the devices again
# test if the injector called groups.refresh successfully
self.assertIsNotNone(groups.find(name=device))
def test_xmodmap_file(self):
from_keycode = evdev.ecodes.KEY_A
target = "keyboard"
to_name = "q"
to_keycode = 100
event = (EV_KEY, from_keycode, 1)
name = "Bar Device"
preset_name = "foo"
group = groups.find(name=name)
config_dir = os.path.join(tmp, "foo")
path = os.path.join(config_dir, "presets", name, f"{preset_name}.json")
preset = Preset(path)
preset.add(get_key_mapping(EventCombination(event), target, to_name))
preset.save()
system_mapping.clear()
push_events(fixtures.bar_device, [new_event(*event)])
# an existing config file is needed otherwise set_config_dir refuses
# to use the directory
config_path = os.path.join(config_dir, "config.json")
global_config.path = config_path
global_config._save_config()
xmodmap_path = os.path.join(config_dir, "xmodmap.json")
with open(xmodmap_path, "w") as file:
file.write(f'{{"{to_name}":{to_keycode}}}')
self.daemon = Daemon()
self.daemon.set_config_dir(config_dir)
self.daemon.start_injecting(group.key, preset_name)
time.sleep(0.1)
self.assertTrue(uinput_write_history_pipe[0].poll())
event = uinput_write_history_pipe[0].recv()
self.assertEqual(event.type, EV_KEY)
self.assertEqual(event.code, to_keycode)
self.assertEqual(event.value, 1)
def test_start_stop(self):
group = groups.find(key="Foo Device 2")
preset_name = "preset8"
daemon = Daemon()
self.daemon = daemon
pereset = Preset(group.get_preset_path(preset_name))
pereset.add(get_key_mapping(EventCombination([3, 2, 1]), "keyboard", "a"))
pereset.save()
# start
daemon.start_injecting(group.key, preset_name)
# explicit start, not autoload, so the history stays empty
self.assertNotIn(group.key, daemon.autoload_history._autoload_history)
self.assertTrue(daemon.autoload_history.may_autoload(group.key, preset_name))
# path got translated to the device name
self.assertIn(group.key, daemon.injectors)
# start again
previous_injector = daemon.injectors[group.key]
self.assertNotEqual(previous_injector.get_state(), STOPPED)
daemon.start_injecting(group.key, preset_name)
self.assertNotIn(group.key, daemon.autoload_history._autoload_history)
self.assertTrue(daemon.autoload_history.may_autoload(group.key, preset_name))
self.assertIn(group.key, daemon.injectors)
time.sleep(0.2)
self.assertEqual(previous_injector.get_state(), STOPPED)
# a different injetor is now running
self.assertNotEqual(previous_injector, daemon.injectors[group.key])
self.assertNotEqual(daemon.injectors[group.key].get_state(), STOPPED)
# trying to inject a non existing preset keeps the previous inejction
# alive
injector = daemon.injectors[group.key]
daemon.start_injecting(group.key, "qux")
self.assertEqual(injector, daemon.injectors[group.key])
self.assertNotEqual(daemon.injectors[group.key].get_state(), STOPPED)
# trying to start injecting for an unknown device also just does
# nothing
daemon.start_injecting("quux", "qux")
self.assertNotEqual(daemon.injectors[group.key].get_state(), STOPPED)
# after all that stuff autoload_history is still unharmed
self.assertNotIn(group.key, daemon.autoload_history._autoload_history)
self.assertTrue(daemon.autoload_history.may_autoload(group.key, preset_name))
# stop
daemon.stop_injecting(group.key)
time.sleep(0.2)
self.assertNotIn(group.key, daemon.autoload_history._autoload_history)
self.assertEqual(daemon.injectors[group.key].get_state(), STOPPED)
self.assertTrue(daemon.autoload_history.may_autoload(group.key, preset_name))
def test_autoload(self):
preset_name = "preset7"
group = groups.find(key="Foo Device 2")
daemon = Daemon()
self.daemon = daemon
preset = Preset(group.get_preset_path(preset_name))
preset.add(get_key_mapping(EventCombination([3, 2, 1]), "keyboard", "a"))
preset.save()
# no autoloading is configured yet
self.daemon._autoload(group.key)
self.assertNotIn(group.key, daemon.autoload_history._autoload_history)
self.assertTrue(daemon.autoload_history.may_autoload(group.key, preset_name))
global_config.set_autoload_preset(group.key, preset_name)
len_before = len(self.daemon.autoload_history._autoload_history)
# now autoloading is configured, so it will autoload
self.daemon._autoload(group.key)
len_after = len(self.daemon.autoload_history._autoload_history)
self.assertEqual(
daemon.autoload_history._autoload_history[group.key][1], preset_name
)
self.assertFalse(daemon.autoload_history.may_autoload(group.key, preset_name))
injector = daemon.injectors[group.key]
self.assertEqual(len_before + 1, len_after)
# calling duplicate get_autoload does nothing
self.daemon._autoload(group.key)
self.assertEqual(
daemon.autoload_history._autoload_history[group.key][1], preset_name
)
self.assertEqual(injector, daemon.injectors[group.key])
self.assertFalse(daemon.autoload_history.may_autoload(group.key, preset_name))
# explicit start_injecting clears the autoload history
self.daemon.start_injecting(group.key, preset_name)
self.assertTrue(daemon.autoload_history.may_autoload(group.key, preset_name))
# calling autoload for (yet) unknown devices does nothing
len_before = len(self.daemon.autoload_history._autoload_history)
self.daemon._autoload("unknown-key-1234")
len_after = len(self.daemon.autoload_history._autoload_history)
self.assertEqual(len_before, len_after)
# autoloading input-remapper devices does nothing
len_before = len(self.daemon.autoload_history._autoload_history)
self.daemon.autoload_single("Bar Device")
len_after = len(self.daemon.autoload_history._autoload_history)
self.assertEqual(len_before, len_after)
def test_autoload_2(self):
self.daemon = Daemon()
history = self.daemon.autoload_history._autoload_history
# existing device
preset_name = "preset7"
group = groups.find(key="Foo Device 2")
preset = Preset(group.get_preset_path(preset_name))
preset.add(get_key_mapping(EventCombination([3, 2, 1]), "keyboard", "a"))
preset.save()
global_config.set_autoload_preset(group.key, preset_name)
# ignored, won't cause problems:
global_config.set_autoload_preset("non-existant-key", "foo")
self.daemon.autoload()
self.assertEqual(len(history), 1)
self.assertEqual(history[group.key][1], preset_name)
def test_autoload_3(self):
# based on a bug
preset_name = "preset7"
group = groups.find(key="Foo Device 2")
preset = Preset(group.get_preset_path(preset_name))
preset.add(get_key_mapping(EventCombination([3, 2, 1]), "keyboard", "a"))
preset.save()
global_config.set_autoload_preset(group.key, preset_name)
self.daemon = Daemon()
groups.set_groups([]) # caused the bug
self.assertIsNone(groups.find(key="Foo Device 2"))
self.daemon.autoload()
# it should try to refresh the groups because all the
# group_keys are unknown at the moment
history = self.daemon.autoload_history._autoload_history
self.assertEqual(history[group.key][1], preset_name)
self.assertEqual(self.daemon.get_state(group.key), STARTING)
self.assertIsNotNone(groups.find(key="Foo Device 2"))
if __name__ == "__main__":
unittest.main()