Making ReaderService terminate automatically after some time (#519)

pull/505/head
Tobi 2 years ago committed by GitHub
parent 27e911be82
commit b5345ad4c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -43,7 +43,7 @@ HELLO = 'hello'
# internal stuff that the gui uses
START_DAEMON = 'start-daemon'
HELPER = 'helper'
START_READER_SERVICE = 'start-reader-service'
def run(cmd):
@ -56,7 +56,7 @@ def run(cmd):
COMMANDS = [AUTOLOAD, START, STOP, HELLO, STOP_ALL]
INTERNALS = [START_DAEMON, HELPER]
INTERNALS = [START_DAEMON, START_READER_SERVICE]
def utils(options):
@ -166,8 +166,8 @@ def internals(options):
"""
debug = ' -d' if options.debug else ''
if options.command == HELPER:
cmd = f'input-remapper-helper{debug}'
if options.command == START_READER_SERVICE:
cmd = f'input-remapper-reader-service{debug}'
elif options.command == START_DAEMON:
cmd = f'input-remapper-service --hide-info{debug}'
else:
@ -202,14 +202,16 @@ def _systemd_finished():
def boot_finished():
"""Check if booting is completed."""
# Get as much information as needed to really safely determine if booting up is complete.
# Get as much information as needed to really safely determine if booting up is
# complete.
# - `who` returns an empty list on some system for security purposes
# - something might be broken and might make systemd_analyze fail:
# Bootup is not yet finished (org.freedesktop.systemd1.Manager.FinishTimestampMonotonic=0).
# Bootup is not yet finished
# (org.freedesktop.systemd1.Manager.FinishTimestampMonotonic=0).
# Please try again later.
# Hint: Use 'systemctl list-jobs' to see active jobs
if _systemd_finished():
logger.debug('Booting finished')
logger.debug('System is booted')
return True
if _num_logged_in_users() > 0:

@ -21,43 +21,35 @@
"""Starts the user interface."""
from __future__ import annotations
import os
import sys
import atexit
import logging
from argparse import ArgumentParser
from inputremapper.gui.gettext import _, LOCALE_DIR
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('GLib', '2.0')
gi.require_version('GtkSource', '4')
from gi.repository import Gtk
# https://github.com/Nuitka/Nuitka/issues/607#issuecomment-650217096
Gtk.init()
from inputremapper.gui.gettext import _, LOCALE_DIR
from inputremapper.gui.reader_service import ReaderService
from inputremapper.daemon import DaemonProxy
from inputremapper.logger import logger, update_verbosity, log_info
def start_processes() -> DaemonProxy:
"""Start helper and daemon via pkexec to run in the background."""
"""Start reader-service and daemon via pkexec to run in the background."""
# this function is overwritten in tests
daemon = Daemon.connect()
debug = " -d" if logger.level <= logging.DEBUG else ""
cmd = f"pkexec input-remapper-control --command helper {debug}"
logger.debug("Running `%s`", cmd)
exit_code = os.system(cmd)
if exit_code != 0:
logger.error("Failed to pkexec the helper, code %d", exit_code)
try:
ReaderService.pkexec_reader_service()
except Exception as e:
logger.error(e)
sys.exit(11)
return daemon
return Daemon.connect()
if __name__ == '__main__':
@ -81,8 +73,8 @@ if __name__ == '__main__':
from inputremapper.gui.controller import Controller
from inputremapper.injection.global_uinputs import GlobalUInputs
from inputremapper.groups import _Groups
from inputremapper.gui.reader import Reader
from inputremapper.daemon import Daemon, DaemonProxy
from inputremapper.gui.reader_client import ReaderClient
from inputremapper.daemon import Daemon
from inputremapper.configs.global_config import GlobalConfig
from inputremapper.configs.migrations import migrate
@ -90,13 +82,13 @@ if __name__ == '__main__':
message_broker = MessageBroker()
# create the reader before we start the helper (start_processes) otherwise it
# can come to race conditions with the creation of pipes
reader = Reader(message_broker, _Groups())
# create the reader before we start the reader-service (start_processes) otherwise
# it can come to race conditions with the creation of pipes
reader_client = ReaderClient(message_broker, _Groups())
daemon = start_processes()
data_manager = DataManager(
message_broker, GlobalConfig(), reader, daemon, GlobalUInputs(), system_mapping
message_broker, GlobalConfig(), reader_client, daemon, GlobalUInputs(), system_mapping
)
controller = Controller(message_broker, data_manager)
user_interface = UserInterface(message_broker, controller)

@ -19,7 +19,7 @@
# along with input-remapper. If not, see <https://www.gnu.org/licenses/>.
"""Starts the root helper."""
"""Starts the root reader-service."""
import os
@ -44,7 +44,7 @@ if __name__ == '__main__':
update_verbosity(options.debug)
# import input-remapper stuff after setting the log verbosity
from inputremapper.gui.helper import RootHelper
from inputremapper.gui.reader_service import ReaderService
def on_exit():
"""Don't remain idle and alive when the GUI exits via ctrl+c."""
@ -54,6 +54,7 @@ if __name__ == '__main__':
os.kill(os.getpid(), signal.SIGKILL)
atexit.register(on_exit)
# TODO import `groups` instead?
groups = _Groups()
helper = RootHelper(groups)
helper.run()
reader_service = ReaderService(groups)
reader_service.run()

@ -7,8 +7,8 @@
<action id="inputremapper">
<description>Run Input Remapper as root</description>
<message>Authentication is required to discover and read devices.</message>
<message xml:lang="sk">Vyžaduje sa prihlásenie na objavenie a prístup k zariadeniam.</message>
<message xml:lang="ru">Требуется аутентификация для обнаружения и чтения устройств.</message>
<message xml:lang="sk">Vyžaduje sa prihlásenie na objavenie a prístup k zariadeniam.</message>
<message xml:lang="ru">Требуется аутентификация для обнаружения и чтения устройств.</message>
<defaults>
<allow_any>no</allow_any>
<allow_inactive>auth_admin_keep</allow_inactive>

@ -19,7 +19,11 @@
# along with input-remapper. If not, see <https://www.gnu.org/licenses/>.
"""Migration functions"""
"""Migration functions.
Only write changes to disk, if there actually are changes. Otherwise file-modification
dates are destroyed.
"""
import copy
import json
@ -106,7 +110,7 @@ def _config_suffix():
def _preset_path():
"""Migrate the folder structure from < 0.4.0.
Move existing presets into the new subfolder "presets"
Move existing presets into the new subfolder 'presets'
"""
new_preset_folder = os.path.join(CONFIG_PATH, "presets")
if os.path.exists(get_preset_path()) or not os.path.exists(CONFIG_PATH):
@ -128,18 +132,22 @@ def _preset_path():
def _mapping_keys():
"""Update all preset mappings.
Update all keys in preset to include value e.g.: "1,5"->"1,5,1"
Update all keys in preset to include value e.g.: '1,5'->'1,5,1'
"""
for preset, preset_dict in all_presets():
changes = 0
if "mapping" in preset_dict.keys():
mapping = copy.deepcopy(preset_dict["mapping"])
for key in mapping.keys():
if key.count(",") == 1:
preset_dict["mapping"][f"{key},1"] = preset_dict["mapping"].pop(key)
changes += 1
with open(preset, "w") as file:
json.dump(preset_dict, file, indent=4)
file.write("\n")
if changes:
with open(preset, "w") as file:
logger.info('Updating mapping keys of "%s"', preset)
json.dump(preset_dict, file, indent=4)
file.write("\n")
def _update_version():
@ -148,12 +156,12 @@ def _update_version():
if not os.path.exists(config_file):
return
logger.info("Updating version in config to %s", VERSION)
with open(config_file, "r") as file:
config = json.load(file)
config["version"] = VERSION
with open(config_file, "w") as file:
logger.info('Updating version in config to "%s"', VERSION)
json.dump(config, file, indent=4)
@ -183,7 +191,7 @@ def _find_target(symbol):
if capabilities[EV_KEY].issubset(uinput.capabilities()[EV_KEY]):
return name
logger.info("could not find a suitable target UInput for '%s'", symbol)
logger.info('could not find a suitable target UInput for "%s"', symbol)
return None
@ -217,6 +225,7 @@ def _add_target():
continue
with open(preset, "w") as file:
logger.info('Adding targets for "%s"', preset)
json.dump(preset_dict, file, indent=4)
file.write("\n")
@ -253,6 +262,7 @@ def _otherwise_to_else():
continue
with open(preset, "w") as file:
logger.info('Changing otherwise to else for "%s"', preset)
json.dump(preset_dict, file, indent=4)
file.write("\n")

@ -122,8 +122,6 @@ class SystemMapping:
if name.startswith("KEY") or name.startswith("BTN"):
self._set(name, ecode)
self._set(DISABLE_NAME, DISABLE_CODE)
def populate(self):
"""Get a mapping of all available names to their keycodes."""
logger.debug("Gathering available keycodes")
@ -136,6 +134,8 @@ class SystemMapping:
self._use_linux_evdev_symbols()
self._set(DISABLE_NAME, DISABLE_CODE)
def update(self, mapping: dict):
"""Update this with new keys.

@ -33,9 +33,10 @@ import time
from pathlib import PurePath
from typing import Protocol, Dict
import gi
from pydbus import SystemBus
import gi
gi.require_version("GLib", "2.0")
from gi.repository import GLib
@ -214,14 +215,13 @@ class Daemon:
macro_variables.start()
@classmethod
def connect(cls, fallback=True) -> DaemonProxy:
def connect(cls, fallback: bool = True) -> DaemonProxy:
"""Get an interface to start and stop injecting keystrokes.
Parameters
----------
fallback : bool
If true, returns an instance of the daemon instead if it cannot
connect
fallback
If true, starts the daemon via pkexec if it cannot connect.
"""
bus = SystemBus()
try:

@ -26,6 +26,12 @@ import re
from typing import Dict, Optional, List, Tuple
from evdev.ecodes import EV_KEY
import gi
gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0")
gi.require_version("GLib", "2.0")
from gi.repository import Gdk, Gtk, GLib, GObject
from inputremapper.configs.mapping import MappingData

@ -24,6 +24,9 @@
from __future__ import annotations
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from typing import Optional

@ -21,6 +21,9 @@
from __future__ import annotations
from typing import Optional
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from inputremapper.gui.components.common import FlowBoxEntry, FlowBoxWrapper
@ -65,7 +68,7 @@ class DeviceGroupEntry(FlowBoxEntry):
def _on_gtk_toggle(self, *_, **__):
logger.debug('Selecting device "%s"', self.group_key)
self._controller.load_group(self.group_key)
self.message_broker.send(DoStackSwitch(Stack.presets_page))
self.message_broker.publish(DoStackSwitch(Stack.presets_page))
class DeviceGroupSelection(FlowBoxWrapper):
@ -108,5 +111,8 @@ class DeviceGroupSelection(FlowBoxWrapper):
)
self._gui.insert(device_group_entry, -1)
if self._controller.data_manager.active_group:
self.show_active_entry(self._controller.data_manager.active_group.key)
def _on_group_changed(self, data: GroupData):
self.show_active_entry(data.group_key)

@ -29,6 +29,12 @@ from typing import List, Optional, Dict, Union, Callable, Literal
import cairo
from evdev.ecodes import EV_KEY, EV_ABS, EV_REL, bytype
import gi
gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0")
gi.require_version("GtkSource", "4")
from gi.repository import Gtk, GtkSource, Gdk
from inputremapper.configs.mapping import MappingData

@ -24,6 +24,9 @@
from __future__ import annotations
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from inputremapper.gui.controller import Controller

@ -24,6 +24,9 @@
from __future__ import annotations
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from inputremapper.gui.components.common import FlowBoxEntry, FlowBoxWrapper
@ -60,7 +63,7 @@ class PresetEntry(FlowBoxEntry):
def _on_gtk_toggle(self, *_, **__):
logger.debug('Selecting preset "%s"', self.preset_name)
self._controller.load_preset(self.preset_name)
self.message_broker.send(DoStackSwitch(Stack.editor_page))
self.message_broker.publish(DoStackSwitch(Stack.editor_page))
class PresetSelection(FlowBoxWrapper):

@ -33,6 +33,10 @@ from typing import (
)
from evdev.ecodes import EV_KEY, EV_REL, EV_ABS
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from inputremapper.configs.mapping import MappingData, UIMapping
@ -42,7 +46,6 @@ from inputremapper.event_combination import EventCombination
from inputremapper.exceptions import DataManagementError
from inputremapper.gui.data_manager import DataManager, DEFAULT_PRESET_NAME
from inputremapper.gui.gettext import _
from inputremapper.gui.helper import is_helper_running
from inputremapper.gui.messages.message_broker import (
MessageBroker,
MessageType,
@ -90,28 +93,36 @@ class Controller:
self.message_broker.subscribe(MessageType.preset, self._on_preset_changed)
self.message_broker.subscribe(MessageType.init, self._on_init)
self.message_broker.subscribe(
MessageType.preset, self._send_mapping_errors_as_status_msg
MessageType.preset, self._publish_mapping_errors_as_status_msg
)
self.message_broker.subscribe(
MessageType.mapping, self._send_mapping_errors_as_status_msg
MessageType.mapping, self._publish_mapping_errors_as_status_msg
)
def _on_init(self, __):
"""initialize the gui and the data_manager"""
# make sure we get a groups_changed event when everything is ready
# this might not be necessary if the helper takes longer to provide the
# this might not be necessary if the reader-service takes longer to provide the
# initial groups
self.data_manager.send_groups()
self.data_manager.send_uinputs()
if not is_helper_running():
self.show_status(CTX_ERROR, _("The helper did not start"))
self.data_manager.publish_groups()
self.data_manager.publish_uinputs()
def _on_groups_changed(self, _):
"""load the newest group as soon as everyone got notified
about the updated groups"""
if self.data_manager.active_group is not None:
# don't jump to a different group and preset suddenly, if the user
# is already looking at one
logger.debug("A group is already active")
return
group_key = self.get_a_group()
if group_key:
self.load_group(self.get_a_group())
if group_key is None:
logger.debug("Could not find a group")
return
self.load_group(group_key)
def _on_preset_changed(self, data: PresetData):
"""load a mapping as soon as everyone got notified about the new preset"""
@ -127,17 +138,17 @@ class Controller:
self.load_event(combination[0])
else:
# send an empty mapping to make sure the ui is reset to default values
self.message_broker.send(MappingData(**MAPPING_DEFAULTS))
self.message_broker.publish(MappingData(**MAPPING_DEFAULTS))
def _on_combination_recorded(self, data: CombinationRecorded):
self.update_combination(data.combination)
def _send_mapping_errors_as_status_msg(self, *__):
def _publish_mapping_errors_as_status_msg(self, *__):
"""send mapping ValidationErrors to the MessageBroker."""
if not self.data_manager.active_preset:
return
if self.data_manager.active_preset.is_valid():
self.message_broker.send(StatusData(CTX_MAPPING))
self.message_broker.publish(StatusData(CTX_MAPPING))
return
for mapping in self.data_manager.active_preset:
@ -231,7 +242,7 @@ class Controller:
self.data_manager.copy_preset(
self.data_manager.get_available_preset_name(f"{name} copy")
)
self.message_broker.send(DoStackSwitch(1))
self.message_broker.publish(DoStackSwitch(1))
def update_combination(self, combination: EventCombination):
"""update the event_combination of the active mapping"""
@ -306,8 +317,8 @@ class Controller:
self.data_manager.update_event(new_event)
except KeyError:
# we need to synchronize the gui
self.data_manager.send_mapping()
self.data_manager.send_event()
self.data_manager.publish_mapping()
self.data_manager.publish_event()
def remove_event(self):
"""remove the active InputEvent from the active mapping event combination"""
@ -324,8 +335,8 @@ class Controller:
self.save()
except (KeyError, ValueError):
# we need to synchronize the gui
self.data_manager.send_mapping()
self.data_manager.send_event()
self.data_manager.publish_mapping()
self.data_manager.publish_event()
def set_event_as_analog(self, analog: bool):
"""use the active event as an analog input"""
@ -354,8 +365,8 @@ class Controller:
# didn't update successfully
# we need to synchronize the gui
self.data_manager.send_mapping()
self.data_manager.send_event()
self.data_manager.publish_mapping()
self.data_manager.publish_event()
def load_groups(self):
"""refresh the groups"""
@ -400,7 +411,7 @@ class Controller:
if answer:
self.data_manager.delete_preset()
self.data_manager.load_preset(self.get_a_preset())
self.message_broker.send(DoStackSwitch(1))
self.message_broker.publish(DoStackSwitch(1))
if not self.data_manager.active_preset:
return
@ -408,7 +419,7 @@ class Controller:
_('Are you sure you want to delete the preset "%s"?')
% self.data_manager.active_preset.name
)
self.message_broker.send(UserConfirmRequest(msg, f))
self.message_broker.publish(UserConfirmRequest(msg, f))
def load_mapping(self, event_combination: EventCombination):
"""load the mapping with the given event_combination form the active_preset"""
@ -420,8 +431,8 @@ class Controller:
if "mapping_type" in kwargs.keys():
if not (kwargs := self._change_mapping_type(kwargs)):
# we need to synchronize the gui
self.data_manager.send_mapping()
self.data_manager.send_event()
self.data_manager.publish_mapping()
self.data_manager.publish_event()
return
self.data_manager.update_mapping(**kwargs)
@ -448,7 +459,7 @@ class Controller:
if not self.data_manager.active_mapping:
return
self.message_broker.send(
self.message_broker.publish(
UserConfirmRequest(_("Are you sure you want to delete this mapping?"), f)
)
@ -469,31 +480,33 @@ class Controller:
Updates the active_mapping.event_combination with the recorded events.
"""
self.message_broker.signal(MessageType.recording_started) # TODO test
state = self.data_manager.get_state()
if state == InjectorState.RUNNING or state == InjectorState.STARTING:
self.data_manager.stop_combination_recording()
self.message_broker.signal(MessageType.recording_finished)
self.show_status(CTX_ERROR, _('Use "Stop" to stop before editing'))
return
logger.debug("Recording Keys")
def f(_):
self.message_broker.unsubscribe(f)
def on_recording_finished(_):
self.message_broker.unsubscribe(on_recording_finished)
self.message_broker.unsubscribe(self._on_combination_recorded)
self.gui.connect_shortcuts()
self.gui.disconnect_shortcuts()
self.message_broker.subscribe(
MessageType.combination_recorded, self._on_combination_recorded
MessageType.combination_recorded,
self._on_combination_recorded,
)
self.message_broker.subscribe(
MessageType.recording_finished, on_recording_finished
)
self.message_broker.subscribe(MessageType.recording_finished, f)
self.data_manager.start_combination_recording()
def stop_key_recording(self):
"""stop recording the input"""
logger.debug("Stopping Key recording")
logger.debug("Stopping Recording Keys")
self.data_manager.stop_combination_recording()
def start_injecting(self):
@ -600,7 +613,7 @@ class Controller:
self, ctx_id: int, msg: Optional[str] = None, tooltip: Optional[str] = None
):
"""send a status message to the ui to show it in the status-bar"""
self.message_broker.send(StatusData(ctx_id, msg, tooltip))
self.message_broker.publish(StatusData(ctx_id, msg, tooltip))
def is_empty_mapping(self) -> bool:
"""check if the active_mapping is empty"""
@ -670,7 +683,7 @@ class Controller:
nonlocal answer
answer = a
self.message_broker.send(UserConfirmRequest(msg, f))
self.message_broker.publish(UserConfirmRequest(msg, f))
if answer:
kwargs["output_symbol"] = None
return kwargs
@ -691,7 +704,7 @@ class Controller:
nonlocal answer
answer = a
self.message_broker.send(
self.message_broker.publish(
UserConfirmRequest(
f"You are about to change the mapping to a Key or Macro mapping!\n"
f"Go to the advanced input configuration and set a "

@ -23,6 +23,9 @@ import re
import time
from typing import Optional, List, Tuple, Set
import gi
gi.require_version("GLib", "2.0")
from gi.repository import GLib
from inputremapper.configs.global_config import GlobalConfig
@ -43,7 +46,7 @@ from inputremapper.gui.messages.message_data import (
PresetData,
CombinationUpdate,
)
from inputremapper.gui.reader import Reader
from inputremapper.gui.reader_client import ReaderClient
from inputremapper.injection.global_uinputs import GlobalUInputs
from inputremapper.injection.injector import (
InjectorState,
@ -71,13 +74,13 @@ class DataManager:
self,
message_broker: MessageBroker,
config: GlobalConfig,
reader: Reader,
reader_client: ReaderClient,
daemon: DaemonProxy,
uinputs: GlobalUInputs,
system_mapping: SystemMapping,
):
self.message_broker = message_broker
self._reader = reader
self._reader_client = reader_client
self._daemon = daemon
self._uinputs = uinputs
self._system_mapping = system_mapping
@ -90,38 +93,38 @@ class DataManager:
self._active_mapping: Optional[UIMapping] = None
self._active_event: Optional[InputEvent] = None
def send_group(self):
def publish_group(self):
"""send active group to the MessageBroker.
This is internally called whenever the group changes.
It is usually not necessary to call this explicitly from
outside DataManager"""
self.message_broker.send(
self.message_broker.publish(
GroupData(self.active_group.key, self.get_preset_names())
)
def send_preset(self):
def publish_preset(self):
"""send active preset to the MessageBroker.
This is internally called whenever the preset changes.
It is usually not necessary to call this explicitly from
outside DataManager"""
self.message_broker.send(
self.message_broker.publish(
PresetData(
self.active_preset.name, self.get_mappings(), self.get_autoload()
)
)
def send_mapping(self):
def publish_mapping(self):
"""send active mapping to the MessageBroker
This is internally called whenever the mapping changes.
It is usually not necessary to call this explicitly from
outside DataManager"""
if self.active_mapping:
self.message_broker.send(self.active_mapping.get_bus_message())
self.message_broker.publish(self.active_mapping.get_bus_message())
def send_event(self):
def publish_event(self):
"""send active event to the MessageBroker.
This is internally called whenever the event changes.
@ -129,11 +132,11 @@ class DataManager:
outside DataManager"""
if self.active_event:
assert self.active_event in self.active_mapping.event_combination
self.message_broker.send(self.active_event)
self.message_broker.publish(self.active_event)
def send_uinputs(self):
def publish_uinputs(self):
"""send the "uinputs" message on the MessageBroker"""
self.message_broker.send(
self.message_broker.publish(
UInputsData(
{
name: uinput.capabilities()
@ -142,21 +145,21 @@ class DataManager:
)
)
def send_groups(self):
def publish_groups(self):
"""send the "groups" message on the MessageBroker"""
self._reader.send_groups()
self._reader_client.publish_groups()
def send_injector_state(self):
def publish_injector_state(self):
"""send the "injector_state" message with the state of the injector
for the active_group"""
if not self.active_group:
return
self.message_broker.send(InjectorStateMessage(self.get_state()))
self.message_broker.publish(InjectorStateMessage(self.get_state()))
@property
def active_group(self) -> Optional[_Group]:
"""the currently loaded group"""
return self._reader.group
return self._reader_client.group
@property
def active_preset(self) -> Optional[Preset[UIMapping]]:
@ -175,7 +178,7 @@ class DataManager:
def get_group_keys(self) -> Tuple[GroupKey, ...]:
"""Get all group keys (plugged devices)"""
return tuple(group.key for group in self._reader.groups.filter())
return tuple(group.key for group in self._reader_client.groups.filter())
def get_preset_names(self) -> Tuple[Name, ...]:
"""Get all preset names for active_group and current user,
@ -223,13 +226,13 @@ class DataManager:
elif self.get_autoload:
self._config.set_autoload_preset(self.active_group.key, None)
self.send_preset()
self.publish_preset()
def get_newest_group_key(self) -> GroupKey:
"""group_key of the group with the most recently modified preset"""
paths = []
for path in glob.glob(os.path.join(get_preset_path(), "*/*.json")):
if self._reader.groups.find(key=split_all(path)[-2]):
if self._reader_client.groups.find(key=split_all(path)[-2]):
paths.append((path, os.path.getmtime(path)))
if not paths:
@ -291,13 +294,15 @@ class DataManager:
if group_key not in self.get_group_keys():
raise DataManagementError("Unable to load non existing group")
logger.info('Loading group "%s"', group_key)
self._active_event = None
self._active_mapping = None
self._active_preset = None
group = self._reader.groups.find(key=group_key)
self._reader.set_group(group)
self.send_group()
self.send_injector_state()
group = self._reader_client.groups.find(key=group_key)
self._reader_client.set_group(group)
self.publish_group()
self.publish_injector_state()
def load_preset(self, name: str):
"""Load a preset. Will send "preset" message on the MessageBroker
@ -307,13 +312,15 @@ class DataManager:
if not self.active_group:
raise DataManagementError("Unable to load preset. Group is not set")
logger.info('Loading preset "%s"', name)
preset_path = get_preset_path(self.active_group.name, name)
preset = Preset(preset_path, mapping_factory=UIMapping)
preset.load()
self._active_event = None
self._active_mapping = None
self._active_preset = preset
self.send_preset()
self.publish_preset()
def load_mapping(self, combination: EventCombination):
"""Load a mapping. Will send "mapping" message on the MessageBroker"""
@ -328,7 +335,7 @@ class DataManager:
)
self._active_event = None
self._active_mapping = mapping
self.send_mapping()
self.publish_mapping()
def load_event(self, event: InputEvent):
"""Load a InputEvent from the combination in the active mapping.
@ -342,7 +349,7 @@ class DataManager:
f"{self.active_mapping.event_combination}"
)
self._active_event = event
self.send_event()
self.publish_event()
def rename_preset(self, new_name: str):
"""rename the current preset and move the correct file
@ -373,8 +380,8 @@ class DataManager:
self._config.set_autoload_preset(self.active_group.key, new_name)
self.active_preset.path = get_preset_path(self.active_group.name, new_name)
self.send_group()
self.send_preset()
self.publish_group()
self.publish_preset()
def copy_preset(self, name: str):
"""copy the current preset to the given name.
@ -394,8 +401,8 @@ class DataManager:
logger.info('Copy "%s" to "%s"', self.active_preset.path, new_path)
self.active_preset.path = new_path
self.save()
self.send_group()
self.send_preset()
self.publish_group()
self.publish_preset()
def create_preset(self, name: str):
"""create empty preset in the active_group.
@ -409,7 +416,7 @@ class DataManager:
raise DataManagementError("Unable to add preset. Preset exists")
Preset(path).save()
self.send_group()
self.publish_group()
def delete_preset(self):
"""delete the active preset
@ -421,7 +428,7 @@ class DataManager:
os.remove(preset_path)
self._active_mapping = None
self._active_preset = None
self.send_group()
self.publish_group()
def update_mapping(self, **kwargs):
"""update the active mapping with the given keywords and values.
@ -444,14 +451,14 @@ class DataManager:
and combination != self.active_mapping.event_combination
):
self._active_event = None
self.message_broker.send(
self.message_broker.publish(
CombinationUpdate(combination, self._active_mapping.event_combination)
)
if "mapping_type" in kwargs:
# mapping_type must be the last update because it is automatically updated
# by a validation function
self._active_mapping.mapping_type = kwargs["mapping_type"]
self.send_mapping()
self.publish_mapping()
def update_event(self, new_event: InputEvent):
"""update the active event.
@ -466,7 +473,7 @@ class DataManager:
combination[combination.index(self.active_event)] = new_event
self.update_mapping(event_combination=EventCombination(combination))
self._active_event = new_event
self.send_event()
self.publish_event()
def create_mapping(self):
"""create empty mapping in the active preset.
@ -475,7 +482,7 @@ class DataManager:
if not self._active_preset:
raise DataManagementError("cannot create mapping: preset is not set")
self._active_preset.add(UIMapping())
self.send_preset()
self.publish_preset()
def delete_mapping(self):
"""delete the active mapping
@ -488,7 +495,7 @@ class DataManager:
self._active_preset.remove(self._active_mapping.event_combination)
self._active_mapping = None
self.send_preset()
self.publish_preset()
def save(self):
"""save the active preset"""
@ -500,7 +507,7 @@ class DataManager:
Should send "groups" message to MessageBroker this will not happen immediately
because the system might take a bit until the groups are available
"""
self._reader.refresh_groups()
self._reader_client.refresh_groups()
def start_combination_recording(self):
"""Record user input.
@ -508,14 +515,14 @@ class DataManager:
Will send "combination_recorded" messages as new input arrives.
Will eventually send a "recording_finished" message.
"""
self._reader.start_recorder()
self._reader_client.start_recorder()
def stop_combination_recording(self):
"""Stop recording user input.
Will send RecordingFinished message if a recording is running.
"""
self._reader.stop_recorder()
self._reader_client.stop_recorder()
def stop_injecting(self) -> None:
"""stop injecting for the active group
@ -524,7 +531,9 @@ class DataManager:
if not self.active_group:
raise DataManagementError("cannot stop injection: group is not set")
self._daemon.stop_injecting(self.active_group.key)
self.do_when_injector_state({InjectorState.STOPPED}, self.send_injector_state)
self.do_when_injector_state(
{InjectorState.STOPPED}, self.publish_injector_state
)
def start_injecting(self) -> bool:
"""start injecting the active preset for the active group.
@ -545,7 +554,7 @@ class DataManager:
InjectorState.NO_GRAB,
InjectorState.UPGRADE_EVDEV,
},
self.send_injector_state,
self.publish_injector_state,
)
return True
return False

@ -56,35 +56,35 @@ class MessageBroker:
def __init__(self):
self._listeners: Dict[MessageType, Set[MessageListener]] = defaultdict(set)
self._messages: Deque[Tuple[Message, str, int]] = deque()
self._sending = False
self._publishing = False
def send(self, data: Message):
def publish(self, data: Message):
"""schedule a massage to be sent.
The message will be sent after all currently pending messages are sent"""
self._messages.append((data, *self.get_caller()))
self._send_all()
self._publish_all()
def signal(self, signal: MessageType):
"""send a signal without any data payload"""
self.send(Signal(signal))
self.publish(Signal(signal))
def _send(self, data: Message, file: str, line: int):
def _publish(self, data: Message, file: str, line: int):
logger.debug(f"from {file}:{line}: Signal={data.message_type.name}: {data}")
for listener in self._listeners[data.message_type].copy():
listener(data)
def _send_all(self):
def _publish_all(self):
"""send all scheduled messages in order"""
if self._sending:
if self._publishing:
# don't run this twice, so we not mess up the order
return
self._sending = True
self._publishing = True
try:
while self._messages:
self._send(*self._messages.popleft())
self._publish(*self._messages.popleft())
finally:
self._sending = False
self._publishing = False
def subscribe(self, massage_type: MessageType, listener: MessageListener):
"""attach a listener to an event"""

@ -42,8 +42,11 @@ class MessageType(Enum):
mapping = "mapping"
selected_event = "selected_event"
combination_recorded = "combination_recorded"
# only the reader_client should send those messages:
recording_started = "recording_started"
recording_finished = "recording_finished"
combination_update = "combination_update"
status_msg = "status_msg"
injector_state = "injector_state"

@ -19,119 +19,191 @@
# along with input-remapper. If not, see <https://www.gnu.org/licenses/>.
"""Talking to the GUI helper that has root permissions.
"""Talking to the ReaderService that has root permissions.
see gui.helper.helper
see gui.reader_service.ReaderService
"""
from typing import Optional, List, Generator, Dict, Tuple, Set
import time
import evdev
import gi
gi.require_version("GLib", "2.0")
from gi.repository import GLib
from inputremapper.event_combination import EventCombination
from inputremapper.groups import _Groups, _Group
from inputremapper.gui.helper import (
from inputremapper.gui.reader_service import (
MSG_EVENT,
MSG_GROUPS,
CMD_TERMINATE,
CMD_REFRESH_GROUPS,
CMD_STOP_READING,
get_pipe_paths,
ReaderService,
)
from inputremapper.gui.messages.message_types import MessageType
from inputremapper.gui.messages.message_broker import MessageBroker
from inputremapper.gui.messages.message_data import GroupsData, CombinationRecorded
from inputremapper.gui.messages.message_data import (
GroupsData,
CombinationRecorded,
StatusData,
)
from inputremapper.gui.utils import CTX_ERROR
from inputremapper.gui.gettext import _
from inputremapper.input_event import InputEvent
from inputremapper.ipc.pipe import Pipe
from inputremapper.logger import logger
from inputremapper.user import USER
BLACKLISTED_EVENTS = [(1, evdev.ecodes.BTN_TOOL_DOUBLETAP)]
RecordingGenerator = Generator[None, InputEvent, None]
class Reader:
"""Processes events from the helper for the GUI to use.
class ReaderClient:
"""Processes events from the reader-service for the GUI to use.
Does not serve any purpose for the injection service.
When a button was pressed, the newest keycode can be obtained from this
object. GTK has get_key for keyboard keys, but Reader also
has knowledge of buttons like the middle-mouse button.
When a button was pressed, the newest keycode can be obtained from this object.
GTK has get_key for keyboard keys, but Reader also has knowledge of buttons like
the middle-mouse button.
"""
# how long to wait for the reader-service at most
_timeout: int = 5
def __init__(self, message_broker: MessageBroker, groups: _Groups):
self.groups = groups
self.message_broker = message_broker
self.group: Optional[_Group] = None
self.read_timeout: Optional[int] = None
self._recording_generator: Optional[RecordingGenerator] = None
self._results = None
self._commands = None
self._results_pipe = None
self._commands_pipe = None
self.connect()
self.attach_to_events()
self._read_continuously()
self._read_timeout = GLib.timeout_add(30, self._read)
def ensure_reader_service_running(self):
if ReaderService.is_running():
return
logger.info("ReaderService not running anymore, restarting")
ReaderService.pkexec_reader_service()
# wait until the ReaderService is up
# wait no more than:
polling_period = 0.01
# this will make the gui non-responsive for 0.4s or something. The pkexec
# password prompt will appear, so the user understands that the lag has to
# be connected to the authentication. I would actually prefer the frozen gui
# over a reactive one here, because the short lag shows that stuff is going on
# behind the scenes.
for __ in range(int(self._timeout / polling_period)):
if self._results_pipe.poll():
logger.info("ReaderService started")
break
time.sleep(polling_period)
else:
msg = "The reader-service did not start"
logger.error(msg)
self.message_broker.publish(StatusData(CTX_ERROR, _(msg)))
def _send_command(self, command):
"""Send a command to the ReaderService."""
if command not in [CMD_TERMINATE, CMD_STOP_READING]:
self.ensure_reader_service_running()
logger.debug('Sending "%s" to ReaderService', command)
self._commands_pipe.send(command)
def connect(self):
"""Connect to the helper."""
self._results = Pipe(f"/tmp/input-remapper-{USER}/results")
self._commands = Pipe(f"/tmp/input-remapper-{USER}/commands")
"""Connect to the reader-service."""
self._results_pipe = Pipe(get_pipe_paths()[0])
self._commands_pipe = Pipe(get_pipe_paths()[1])
def attach_to_events(self):
"""Connect listeners to event_reader."""
self.message_broker.subscribe(MessageType.terminate, lambda _: self.terminate())
def _read_continuously(self):
"""Poll the result pipe in regular intervals."""
self.read_timeout = GLib.timeout_add(30, self._read)
self.message_broker.subscribe(
MessageType.terminate,
lambda _: self.terminate(),
)
def _read(self):
"""Read the messages from the helper and handle them."""
while self._results.poll():
message = self._results.recv()
"""Read the messages from the reader-service and handle them."""
while self._results_pipe.poll():
message = self._results_pipe.recv()
logger.debug("Reader received %s", message)
logger.debug("received %s", message)
message_type = message["type"]
message_body = message["message"]
if message_type == MSG_GROUPS:
self._update_groups(message_body)
continue
if message_type == MSG_EVENT:
if not self._recording_generator:
continue
# update the generator
try:
self._recording_generator.send(InputEvent(*message_body))
if self._recording_generator is not None:
self._recording_generator.send(InputEvent(*message_body))
else:
# the ReaderService should only send events while the gui
# is recording, so this is unexpected.
logger.error("Got event, but recorder is not running.")
except StopIteration:
self.message_broker.signal(MessageType.recording_finished)
self._recording_generator = None
# the _recording_generator returned
logger.debug("Recorder finished.")
self.stop_recorder()
break
return True
def start_recorder(self) -> None:
"""Record user input."""
if self.group is None:
logger.error("No group set")
return
logger.debug("Starting recorder.")
self._send_command(self.group.key)
self._recording_generator = self._recorder()
next(self._recording_generator)
self.message_broker.signal(MessageType.recording_started) # TODO test
def stop_recorder(self) -> None:
"""Stop recording the input.
Will send RecordingFinished message.
"""
logger.debug("Stopping recorder.")
self._send_command(CMD_STOP_READING)
if self._recording_generator:
self._recording_generator.close()
self._recording_generator = None
self.message_broker.signal(MessageType.recording_finished)
else:
# this would be unexpected. but this is not critical enough to
# show to the user without debug logs
logger.debug("No recording generator existed")
self.message_broker.signal(MessageType.recording_finished)
def _recorder(self) -> RecordingGenerator:
"""Generator which receives InputEvents.
it accumulates them into EventCombinations and sends those on the message_broker.
it will stop once all keys or inputs are released.
It accumulates them into EventCombinations and sends those on the
message_broker. It will stop once all keys or inputs are released.
"""
active: Set[Tuple[int, int]] = set()
accumulator: List[InputEvent] = []
@ -160,45 +232,45 @@ class Reader:
# update the event
i = accu_type_code.index(event.type_and_code)
accumulator[i] = event
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination(accumulator))
)
if event not in accumulator:
accumulator.append(event)
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination(accumulator))
)
def set_group(self, group):
"""Start reading keycodes for a device."""
logger.debug('Sending start msg to helper for "%s"', group.key)
if self._recording_generator:
self._recording_generator.close()
self._recording_generator = None
self._commands.send(group.key)
"""Set the group for which input events should be read later."""
# TODO load the active_group from the controller instead?
self.group = group
def terminate(self):
"""Stop reading keycodes for good."""
logger.debug("Sending close msg to helper")
self._commands.send(CMD_TERMINATE)
if self.read_timeout:
GLib.source_remove(self.read_timeout)
while self._results.poll():
self._results.recv()
self._send_command(CMD_TERMINATE)
self.stop_recorder()
if self._read_timeout is not None:
GLib.source_remove(self._read_timeout)
self._read_timeout = None
while self._results_pipe.poll():
self._results_pipe.recv()
def refresh_groups(self):
"""Ask the helper for new device groups."""
self._commands.send(CMD_REFRESH_GROUPS)
"""Ask the ReaderService for new device groups."""
self._send_command(CMD_REFRESH_GROUPS)
def send_groups(self):
"""announce all known groups"""
def publish_groups(self):
"""Announce all known groups."""
groups: Dict[str, List[str]] = {
group.key: group.types or []
for group in self.groups.filter(include_inputremapper=False)
}
self.message_broker.send(GroupsData(groups))
self.message_broker.publish(GroupsData(groups))
def _update_groups(self, dump):
if dump != self.groups.dumps():
@ -208,4 +280,4 @@ class Reader:
# send this even if the groups did not change, as the user expects the ui
# to respond in some form
self.send_groups()
self.publish_groups()

@ -27,15 +27,21 @@ GUIs should not run as root
https://wiki.archlinux.org/index.php/Running_GUI_applications_as_root
The service shouldn't do that even though it has root rights, because that
would provide a key-logger that can be accessed by any user at all times,
whereas for the helper to start a password is needed and it stops when the ui
closes.
would enable key-loggers to just ask input-remapper for all user-input.
Instead, the ReaderService is used, which will be stopped when the gui closes.
Whereas for the reader-service to start a password is needed and it stops whe
the ui closes.
This uses the backend injection.event_reader and mapping_handlers to process all the
different input-events into simple on/off events and sends them to the gui.
"""
from __future__ import annotations
import time
import logging
import os
import asyncio
import multiprocessing
import subprocess
@ -58,28 +64,29 @@ from inputremapper.ipc.pipe import Pipe
from inputremapper.logger import logger
from inputremapper.user import USER
# received by the helper
# received by the reader-service
CMD_TERMINATE = "terminate"
CMD_STOP_READING = "stop-reading"
CMD_REFRESH_GROUPS = "refresh_groups"
# sent by the helper to the reader
# sent by the reader-service to the reader
MSG_GROUPS = "groups"
MSG_EVENT = "event"
MSG_STATUS = "status"
def is_helper_running():
"""Check if the helper is running."""
try:
subprocess.check_output(["pgrep", "-f", "input-remapper-helper"])
except subprocess.CalledProcessError:
return False
return True
def get_pipe_paths():
"""Get the path where the pipe can be found."""
return (
f"/tmp/input-remapper-{USER}/reader-results",
f"/tmp/input-remapper-{USER}/reader-commands",
)
class RootHelper:
"""Client that runs as root and works for the GUI.
class ReaderService:
"""Service that only reads events and is supposed to run as root.
Sends device information and keycodes to the GUIs socket.
Sends device information and keycodes to the GUI.
Commands are either numbers for generic commands,
or strings to start listening on a specific device.
@ -93,16 +100,49 @@ class RootHelper:
rel_xy_speed[REL_WHEEL] = 1
rel_xy_speed[REL_HWHEEL] = 1
# Polkit won't ask for another password if the pid stays the same or something, and
# if the previous request was no more than 5 minutes ago. see
# https://unix.stackexchange.com/a/458260.
# If the user does something after 6 minutes they will get a prompt already if the
# reader timed out already, which sounds annoying. Instead, I'd rather have the
# password prompt appear at most every 15 minutes.
_maximum_lifetime: int = 60 * 15
_timeout_tolerance: int = 60
def __init__(self, groups: _Groups):
"""Construct the helper and initialize its sockets."""
"""Construct the reader-service and initialize its communication pipes."""
self._start_time = time.time()
self.groups = groups
self._results = Pipe(f"/tmp/input-remapper-{USER}/results")
self._commands = Pipe(f"/tmp/input-remapper-{USER}/commands")
self._results_pipe = Pipe(get_pipe_paths()[0])
self._commands_pipe = Pipe(get_pipe_paths()[1])
self._pipe = multiprocessing.Pipe()
self._tasks: Set[asyncio.Task] = set()
self._stop_event = asyncio.Event()
self._results_pipe.send({"type": MSG_STATUS, "message": "ready"})
@staticmethod
def is_running():
"""Check if the reader-service is running."""
try:
subprocess.check_output(["pgrep", "-f", "input-remapper-reader-service"])
except subprocess.CalledProcessError:
return False
return True
@staticmethod
def pkexec_reader_service():
"""Start reader-service via pkexec to run in the background."""
debug = " -d" if logger.level <= logging.DEBUG else ""
cmd = f"pkexec input-remapper-control --command start-reader-service{debug}"
logger.debug("Running `%s`", cmd)
exit_code = os.system(cmd)
if exit_code != 0:
raise Exception(f"Failed to pkexec the reader-service, code {exit_code}")
def run(self):
"""Start doing stuff. Blocks."""
# the reader will check for new commands later, once it is running
@ -111,32 +151,65 @@ class RootHelper:
logger.debug("Discovering initial groups")
self.groups.refresh()
self._send_groups()
logger.debug("Waiting commands")
loop.run_until_complete(self._read_commands())
logger.debug("Helper terminates")
sys.exit(0)
loop.run_until_complete(
asyncio.gather(
self._read_commands(),
self._timeout(),
)
)
def _send_groups(self):
"""Send the groups to the gui."""
logger.debug("Sending groups")
self._results.send({"type": MSG_GROUPS, "message": self.groups.dumps()})
self._results_pipe.send({"type": MSG_GROUPS, "message": self.groups.dumps()})
async def _timeout(self):
"""Stop automatically after some time."""
# Prevents a permanent hole for key-loggers to exist, in case the gui crashes.
# If the ReaderService stops even though the gui needs it, it needs to restart
# it. This makes it also more comfortable to have debug mode running during
# development, because it won't keep writing inputs containing passwords and
# such to the terminal forever.
await asyncio.sleep(self._maximum_lifetime)
# if it is currently reading, wait a bit longer for the gui to complete
# what it is doing.
if self._is_reading():
logger.debug("Waiting a bit longer for the gui to finish reading")
for _ in range(self._timeout_tolerance):
if not self._is_reading():
# once reading completes, it should terminate right away
break
await asyncio.sleep(1)
logger.debug("Maximum life-span reached, terminating")
sys.exit(1)
async def _read_commands(self):
"""Handle all unread commands.
this will run until it receives CMD_TERMINATE
"""
async for cmd in self._commands:
logger.debug("Waiting for commands")
async for cmd in self._commands_pipe:
logger.debug('Received command "%s"', cmd)
if cmd == CMD_TERMINATE:
await self._stop_reading()
return
logger.debug("Terminating")
sys.exit(0)
if cmd == CMD_REFRESH_GROUPS:
self.groups.refresh()
self._send_groups()
continue
if cmd == CMD_STOP_READING:
await self._stop_reading()
continue
group = self.groups.find(key=cmd)
if group is None:
# this will block for a bit maybe we want to do this async?
@ -150,6 +223,10 @@ class RootHelper:
logger.error('Received unknown command "%s"', cmd)
def _is_reading(self) -> bool:
"""Check if the ReaderService is currently sending events to the GUI."""
return len(self._tasks) > 0
def _start_reading(self, group: _Group):
"""find all devices of that group, filter interesting ones and send the events
to the gui"""
@ -194,7 +271,7 @@ class RootHelper:
for ev_code in capabilities.get(EV_KEY) or ():
context.notify_callbacks[(EV_KEY, ev_code)].append(
ForwardToUIHandler(self._results).notify
ForwardToUIHandler(self._results_pipe).notify
)
for ev_code in capabilities.get(EV_ABS) or ():
@ -206,7 +283,7 @@ class RootHelper:
handler: MappingHandler = AbsToBtnHandler(
EventCombination((EV_ABS, ev_code, 30)), mapping
)
handler.set_sub_handler(ForwardToUIHandler(self._results))
handler.set_sub_handler(ForwardToUIHandler(self._results_pipe))
context.notify_callbacks[(EV_ABS, ev_code)].append(handler.notify)
# negative direction
@ -217,7 +294,7 @@ class RootHelper:
handler = AbsToBtnHandler(
EventCombination((EV_ABS, ev_code, -30)), mapping
)
handler.set_sub_handler(ForwardToUIHandler(self._results))
handler.set_sub_handler(ForwardToUIHandler(self._results_pipe))
context.notify_callbacks[(EV_ABS, ev_code)].append(handler.notify)
for ev_code in capabilities.get(EV_REL) or ():
@ -234,7 +311,7 @@ class RootHelper:
EventCombination((EV_REL, ev_code, self.rel_xy_speed[ev_code])),
mapping,
)
handler.set_sub_handler(ForwardToUIHandler(self._results))
handler.set_sub_handler(ForwardToUIHandler(self._results_pipe))
context.notify_callbacks[(EV_REL, ev_code)].append(handler.notify)
# negative direction
@ -250,7 +327,7 @@ class RootHelper:
EventCombination((EV_REL, ev_code, -self.rel_xy_speed[ev_code])),
mapping,
)
handler.set_sub_handler(ForwardToUIHandler(self._results))
handler.set_sub_handler(ForwardToUIHandler(self._results_pipe))
context.notify_callbacks[(EV_REL, ev_code)].append(handler.notify)
return context

@ -22,6 +22,11 @@
"""User Interface."""
from typing import Dict, Callable
import gi
gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0")
gi.require_version("GtkSource", "4")
from gi.repository import Gtk, GtkSource, Gdk, GObject
from inputremapper.configs.data import get_data_path

@ -22,6 +22,11 @@ from __future__ import annotations
import time
from typing import List
import gi
gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0")
gi.require_version("GLib", "2.0")
from gi.repository import Gtk, GLib, Gdk
from inputremapper.logger import logger

@ -20,7 +20,9 @@
"""Because multiple calls to async_read_loop won't work."""
import asyncio
import os
from typing import AsyncIterator, Protocol, Set, Dict, Tuple, List
import evdev
@ -44,9 +46,9 @@ class Context(Protocol):
class EventReader:
"""Reads input events from a single device and distributes them.
There is one EventReader object for each source, which tells multiple mapping_handlers
that a new event is ready so that they can inject all sorts of funny
things.
There is one EventReader object for each source, which tells multiple
mapping_handlers that a new event is ready so that they can inject all sorts of
funny things.
Other devnodes may be present for the hardware device, in which case this
needs to be created multiple times.
@ -75,17 +77,30 @@ class EventReader:
self.context = context
self.stop_event = stop_event
def stop(self):
"""Stop the reader."""
self.stop_event.set()
async def read_loop(self) -> AsyncIterator[evdev.InputEvent]:
stop_task = asyncio.Task(self.stop_event.wait())
loop = asyncio.get_running_loop()
events_ready = asyncio.Event()
loop.add_reader(self._source.fileno(), events_ready.set)
while True:
_, pending = await asyncio.wait(
{stop_task, events_ready.wait()},
return_when=asyncio.FIRST_COMPLETED,
)
if stop_task.done():
fd_broken = os.stat(self._source.fileno()).st_nlink == 0
if fd_broken:
# happens when the device is unplugged while reading, causing 100% cpu
# usage because events_ready.set is called repeatedly forever,
# while read_loop will hang at self._source.read_one().
logger.error("fd broke, was the device unplugged?")
if stop_task.done() or fd_broken:
for task in pending:
task.cancel()
loop.remove_reader(self._source.fileno())
@ -176,7 +191,6 @@ class EventReader:
self._source.path,
self._source.fd,
)
async for event in self.read_loop():
await self.handle(InputEvent.from_event(event))

@ -35,6 +35,7 @@
Beware that pipes read any available messages,
even those written by themselves.
"""
import asyncio
import json
import os
@ -46,7 +47,12 @@ from inputremapper.logger import logger
class Pipe:
"""Pipe object."""
"""Pipe object.
This is not for secure communication. If pipes already exist, they will be used,
but existing pipes might have open permissions! Only use this for stuff that
non-privileged users would be allowed to read.
"""
def __init__(self, path):
"""Create a pipe, or open it if it already exists."""
@ -91,7 +97,7 @@ class Pipe:
self._handles = (open(self._fds[0], "r"), open(self._fds[1], "w"))
# clear the pipe of any contents, to avoid leftover messages from breaking
# the helper
# the reader-client or reader-service
while self.poll():
leftover = self.recv()
logger.debug('Cleared leftover message "%s"', leftover)
@ -107,7 +113,7 @@ class Pipe:
"""Read an object from the pipe or None if nothing available.
Doesn't transmit pickles, to avoid injection attacks on the
privileged helper. Only messages that can be converted to json
privileged reader-service. Only messages that can be converted to json
are allowed.
"""
if len(self._unread) > 0:
@ -124,7 +130,7 @@ class Pipe:
if parsed[0] < self._created_at and os.environ.get("UNITTEST"):
# important to avoid race conditions between multiple unittests,
# for example old terminate messages reaching a new instance of
# the helper.
# the reader-service.
logger.debug("Ignoring old message %s", parsed)
return None

@ -43,7 +43,8 @@ are much easier to handle.
# Issues:
# - Tests don't pass with Server (reader) and Client (helper) instead of Pipe
# - Tests don't pass with Server and Client instead of Pipe for reader-client
# and service communication or something
# - Had one case of a test that was blocking forever, seems very rare.
# - Hard to debug, generally very problematic compared to Pipes
# The tool works fine, it's just the tests. BrokenPipe errors reported
@ -120,7 +121,7 @@ class Base:
if len(chunk) == 0:
# select keeps telling me the socket has messages
# ready to be received, and I keep getting empty
# buffers. Happened during a test that ran two helper
# buffers. Happened during a test that ran two reader-service
# processes without stopping the first one.
attempts += 1
if attempts == 2 or not self.reconnect():
@ -136,7 +137,7 @@ class Base:
if parsed[0] < self._created_at:
# important to avoid race conditions between multiple
# unittests, for example old terminate messages reaching
# a new instance of the helper.
# a new instance of the reader-service.
logger.debug("Ignoring old message %s", parsed)
continue
@ -146,7 +147,7 @@ class Base:
"""Get the next message or None if nothing to read.
Doesn't transmit pickles, to avoid injection attacks on the
privileged helper. Only messages that can be converted to json
privileged reader-service. Only messages that can be converted to json
are allowed.
"""
self._receive_new_messages()

@ -27,7 +27,7 @@ import os
import sys
import time
from datetime import datetime
from typing import cast
from typing import cast, Tuple
import pkg_resources
@ -80,12 +80,15 @@ class Logger(logging.Logger):
msg = indent * line[1] + line[0]
self._log(logging.DEBUG, msg, args=None)
def debug_key(self, key, msg, *args):
"""Log a spam message custom tailored to keycode_mapper.
def debug_key(self, key: Tuple[int, int, int], msg, *args):
"""Log a key-event message.
Example:
... DEBUG event_reader.py:143: forwarding ···················· (1, 71, 1)
Parameters
----------
key : tuple of int
key
anything that can be string formatted, but usually a tuple of
(type, code, value) tuples
"""
@ -188,13 +191,16 @@ class ColorfulFormatter(logging.Formatter):
def _get_process_name(self):
"""Generate a beaitiful to read name for this process."""
name = sys.argv[0].split("/")[-1].split("-")[-1]
return {
"gtk": "GUI",
"helper": "GUI-Helper",
"service": "Service",
"control": "Control",
}.get(name, name)
process_path = sys.argv[0]
process_name = process_path.split("/")[-1]
if "input-remapper-" in process_name:
process_name = process_name.replace("input-remapper-", "")
if process_name == "gtk":
process_name = "GUI"
return process_name
def _get_format(self, record):
"""Generate a message format string."""

Binary file not shown.

Before

Width:  |  Height:  |  Size: 244 KiB

After

Width:  |  Height:  |  Size: 251 KiB

@ -7,6 +7,8 @@ Syntax errors are shown in the UI on save. Each `key` function adds a short dela
between key-down, key-up and at the end. See [usage.md](usage.md#configuration-files)
for more info.
Macros are written into the same text field, that would usually contain the output symbol.
Bear in mind that anti-cheat software might detect macros in games.
### key

@ -16,6 +16,8 @@ as shown in the screenshots.
In the text input field, type the key to which you would like to map this key.
More information about the possible mappings can be found in [examples.md](./examples.md) and [below](#key-names).
You can also write your macro into the text input field. If you hit enter, it will switch to a multiline-editor with
line-numbers.
Changes are saved automatically.
Press the "Apply" button to activate (inject) the mapping you created.
@ -37,8 +39,8 @@ No injection should be running anymore.
## Combinations
You can use combinations of different inputs to trigger a mapping: While you recorde
the input (`Recorde Input` - Button) press multiple keys and/or move axis at once.
You can use combinations of different inputs to trigger a mapping: While you record
the input (`Record` - Button) press multiple keys and/or move axis at once.
The mapping will be triggered as soon as all the recorded inputs are pressed.
If you use an axis an input you can modify the threshold at which the mapping is
@ -108,7 +110,7 @@ ultimately decide which character to write.
## Analog Axis
It is possible to map analog inputs to analog outputs. E.g. use a gamepad as a mouse.
For this you need to create a mapping and recorde the input axis. Then click on
For this you need to create a mapping and record the input axis. Then click on
`Advanced` and select `Use as Analog`. Make sure to select a target
which supports analog axis and switch to the `Analog Axis` tab.
There you can select an output axis and use the different sliders to configure the

@ -125,7 +125,7 @@ setup(
("/usr/bin/", ["bin/input-remapper-gtk"]),
("/usr/bin/", ["bin/input-remapper-service"]),
("/usr/bin/", ["bin/input-remapper-control"]),
("/usr/bin/", ["bin/input-remapper-helper"]),
("/usr/bin/", ["bin/input-remapper-reader-service"]),
# those will be deleted at some point:
("/usr/bin/", ["bin/key-mapper-gtk"]),
("/usr/bin/", ["bin/key-mapper-service"]),

@ -25,6 +25,7 @@ import time
import evdev
from evdev.ecodes import KEY_A, KEY_B, KEY_C
import gi
gi.require_version("Gdk", "3.0")
@ -172,7 +173,7 @@ class TestDeviceGroupSelection(ComponentBaseTest):
self.controller_mock,
self.gui,
)
self.message_broker.send(
self.message_broker.publish(
GroupsData(
{
"foo": [DeviceType.GAMEPAD, DeviceType.KEYBOARD],
@ -199,7 +200,7 @@ class TestDeviceGroupSelection(ComponentBaseTest):
self.assertEqual(group_keys, ["foo", "bar", "baz"])
self.assertEqual(icons, ["input-gaming", None, "input-tablet"])
self.message_broker.send(
self.message_broker.publish(
GroupsData(
{
"kuu": [DeviceType.KEYBOARD],
@ -213,9 +214,9 @@ class TestDeviceGroupSelection(ComponentBaseTest):
self.assertEqual(icons, ["input-keyboard", "input-gaming"])
def test_selects_correct_device(self):
self.message_broker.send(GroupData("bar", ()))
self.message_broker.publish(GroupData("bar", ()))
self.assertEqual(FlowBoxTestUtils.get_active_entry(self.gui).group_key, "bar")
self.message_broker.send(GroupData("baz", ()))
self.message_broker.publish(GroupData("baz", ()))
self.assertEqual(FlowBoxTestUtils.get_active_entry(self.gui).group_key, "baz")
def test_loads_group(self):
@ -223,7 +224,7 @@ class TestDeviceGroupSelection(ComponentBaseTest):
self.controller_mock.load_group.assert_called_once_with("bar")
def test_avoids_infinite_recursion(self):
self.message_broker.send(GroupData("bar", ()))
self.message_broker.publish(GroupData("bar", ()))
self.controller_mock.load_group.assert_not_called()
@ -234,7 +235,7 @@ class TestTargetSelection(ComponentBaseTest):
self.selection = TargetSelection(
self.message_broker, self.controller_mock, self.gui
)
self.message_broker.send(
self.message_broker.publish(
UInputsData(
{
"foo": {},
@ -248,7 +249,7 @@ class TestTargetSelection(ComponentBaseTest):
names = [row[0] for row in self.gui.get_model()]
self.assertEqual(names, ["foo", "bar", "baz"])
self.message_broker.send(
self.message_broker.publish(
UInputsData(
{
"kuu": {},
@ -264,13 +265,13 @@ class TestTargetSelection(ComponentBaseTest):
self.controller_mock.update_mapping.called_once_with(target_uinput="baz")
def test_selects_correct_target(self):
self.message_broker.send(MappingData(target_uinput="baz"))
self.message_broker.publish(MappingData(target_uinput="baz"))
self.assertEqual(self.gui.get_active_id(), "baz")
self.message_broker.send(MappingData(target_uinput="bar"))
self.message_broker.publish(MappingData(target_uinput="bar"))
self.assertEqual(self.gui.get_active_id(), "bar")
def test_avoids_infinite_recursion(self):
self.message_broker.send(MappingData(target_uinput="baz"))
self.message_broker.publish(MappingData(target_uinput="baz"))
self.controller_mock.update_mapping.assert_not_called()
@ -281,18 +282,18 @@ class TestPresetSelection(ComponentBaseTest):
self.selection = PresetSelection(
self.message_broker, self.controller_mock, self.gui
)
self.message_broker.send(GroupData("foo", ("preset1", "preset2")))
self.message_broker.publish(GroupData("foo", ("preset1", "preset2")))
def test_populates_presets(self):
names = FlowBoxTestUtils.get_child_names(self.gui)
self.assertEqual(names, ["preset1", "preset2"])
self.message_broker.send(GroupData("foo", ("preset3", "preset4")))
self.message_broker.publish(GroupData("foo", ("preset3", "preset4")))
names = FlowBoxTestUtils.get_child_names(self.gui)
self.assertEqual(names, ["preset3", "preset4"])
def test_selects_preset(self):
self.message_broker.send(
self.message_broker.publish(
PresetData(
"preset2",
(
@ -304,7 +305,7 @@ class TestPresetSelection(ComponentBaseTest):
)
self.assertEqual(FlowBoxTestUtils.get_active_entry(self.gui).name, "preset2")
self.message_broker.send(
self.message_broker.publish(
PresetData(
"preset1",
(
@ -317,7 +318,7 @@ class TestPresetSelection(ComponentBaseTest):
self.assertEqual(FlowBoxTestUtils.get_active_entry(self.gui).name, "preset1")
def test_avoids_infinite_recursion(self):
self.message_broker.send(
self.message_broker.publish(
PresetData(
"preset2",
(
@ -342,7 +343,7 @@ class TestMappingListbox(ComponentBaseTest):
self.message_broker, self.controller_mock, self.gui
)
self.message_broker.send(
self.message_broker.publish(
PresetData(
"preset1",
(
@ -392,7 +393,7 @@ class TestMappingListbox(ComponentBaseTest):
self.assertEqual(labels, ["a + b", "mapping1", "mapping2"])
def test_activates_correct_row(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
name="mapping1", event_combination=EventCombination((1, KEY_C, 1))
)
@ -408,7 +409,7 @@ class TestMappingListbox(ComponentBaseTest):
)
def test_avoids_infinite_recursion(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
name="mapping1", event_combination=EventCombination((1, KEY_C, 1))
)
@ -416,7 +417,7 @@ class TestMappingListbox(ComponentBaseTest):
self.controller_mock.load_mapping.assert_not_called()
def test_sorts_empty_mapping_to_bottom(self):
self.message_broker.send(
self.message_broker.publish(
PresetData(
"preset1",
(
@ -437,7 +438,7 @@ class TestMappingListbox(ComponentBaseTest):
)
bottom_row: MappingSelectionLabel = self.gui.get_row_at_index(2)
self.assertEqual(bottom_row.combination, EventCombination.empty_combination())
self.message_broker.send(
self.message_broker.publish(
PresetData(
"preset1",
(
@ -498,7 +499,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.mapping_selection_label.combination,
EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
self.message_broker.send(
self.message_broker.publish(
CombinationUpdate(
EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
EventCombination((1, KEY_A, 1)),
@ -513,7 +514,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.mapping_selection_label.combination,
EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
self.message_broker.send(
self.message_broker.publish(
CombinationUpdate(
EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
EventCombination((1, KEY_A, 1)),
@ -525,7 +526,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
)
def test_updates_name_when_mapping_changed_and_combination_matches(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
name="foo",
@ -534,7 +535,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.assertEqual(self.mapping_selection_label.label.get_label(), "foo")
def test_ignores_mapping_when_combination_does_not_match(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_C, 1)]),
name="foo",
@ -547,7 +548,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.assertFalse(self.mapping_selection_label.edit_btn.get_visible())
# load the mapping associated with the ListBoxRow
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
@ -555,7 +556,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.assertTrue(self.mapping_selection_label.edit_btn.get_visible())
# load a different row
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_C, 1)]),
)
@ -563,7 +564,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.assertFalse(self.mapping_selection_label.edit_btn.get_visible())
def test_enter_edit_mode_focuses_name_input(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
@ -574,7 +575,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
)
def test_enter_edit_mode_updates_visibility(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
@ -586,7 +587,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.assert_selected()
def test_leaves_edit_mode_on_esc(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
@ -605,7 +606,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.controller_mock.update_mapping.assert_not_called()
def test_update_name(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
@ -617,7 +618,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.controller_mock.update_mapping.assert_called_once_with(name="foo")
def test_name_input_contains_combination_when_name_not_set(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
)
@ -626,7 +627,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.assertEqual(self.mapping_selection_label.name_input.get_text(), "a + b")
def test_name_input_contains_name(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
name="foo",
@ -636,7 +637,7 @@ class TestMappingSelectionLabel(ComponentBaseTest):
self.assertEqual(self.mapping_selection_label.name_input.get_text(), "foo")
def test_removes_name_when_name_matches_combination(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)]),
name="foo",
@ -660,47 +661,47 @@ class TestCodeEditor(ComponentBaseTest):
return buffer.get_text(buffer.get_start_iter(), buffer.get_end_iter(), True)
def test_shows_output_symbol(self):
self.message_broker.send(MappingData(output_symbol="foo"))
self.message_broker.publish(MappingData(output_symbol="foo"))
self.assertEqual(self.get_text(), "foo")
def test_shows_record_input_first_message_when_mapping_is_empty(self):
self.controller_mock.is_empty_mapping.return_value = True
self.message_broker.send(MappingData(output_symbol="foo"))
self.message_broker.publish(MappingData(output_symbol="foo"))
self.assertEqual(self.get_text(), "Record the input first")
def test_active_when_mapping_is_not_empty(self):
self.message_broker.send(MappingData(output_symbol="foo"))
self.message_broker.publish(MappingData(output_symbol="foo"))
self.assertTrue(self.gui.get_sensitive())
self.assertEqual(self.gui.get_opacity(), 1)
def test_expands_to_multiline(self):
self.message_broker.send(MappingData(output_symbol="foo\nbar"))
self.message_broker.publish(MappingData(output_symbol="foo\nbar"))
self.assertIn("multiline", self.gui.get_style_context().list_classes())
def test_shows_line_numbers_when_multiline(self):
self.message_broker.send(MappingData(output_symbol="foo\nbar"))
self.message_broker.publish(MappingData(output_symbol="foo\nbar"))
self.assertTrue(self.gui.get_show_line_numbers())
def test_no_multiline_when_macro_not_multiline(self):
self.message_broker.send(MappingData(output_symbol="foo"))
self.message_broker.publish(MappingData(output_symbol="foo"))
self.assertNotIn("multiline", self.gui.get_style_context().list_classes())
def test_no_line_numbers_macro_not_multiline(self):
self.message_broker.send(MappingData(output_symbol="foo"))
self.message_broker.publish(MappingData(output_symbol="foo"))
self.assertFalse(self.gui.get_show_line_numbers())
def test_is_empty_when_mapping_has_no_output_symbol(self):
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
self.assertEqual(self.get_text(), "")
def test_updates_mapping(self):
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
buffer = self.gui.get_buffer()
buffer.set_text("foo")
self.controller_mock.update_mapping.assert_called_once_with(output_symbol="foo")
def test_avoids_infinite_recursion_when_loading_mapping(self):
self.message_broker.send(MappingData(output_symbol="foo"))
self.message_broker.publish(MappingData(output_symbol="foo"))
self.controller_mock.update_mapping.assert_not_called()
def test_gets_focus_when_input_recording_finises(self):
@ -793,47 +794,47 @@ class TestStatusBar(ComponentBaseTest):
self.assert_empty()
def test_shows_error_status(self):
self.message_broker.send(StatusData(CTX_ERROR, "msg", "tooltip"))
self.message_broker.publish(StatusData(CTX_ERROR, "msg", "tooltip"))
self.assertEqual(self.get_text(), "msg")
self.assertEqual(self.get_tooltip(), "tooltip")
self.assert_error_status()
def test_shows_warning_status(self):
self.message_broker.send(StatusData(CTX_WARNING, "msg", "tooltip"))
self.message_broker.publish(StatusData(CTX_WARNING, "msg", "tooltip"))
self.assertEqual(self.get_text(), "msg")
self.assertEqual(self.get_tooltip(), "tooltip")
self.assert_warning_status()
def test_shows_newest_message(self):
self.message_broker.send(StatusData(CTX_ERROR, "msg", "tooltip"))
self.message_broker.send(StatusData(CTX_WARNING, "msg2", "tooltip2"))
self.message_broker.publish(StatusData(CTX_ERROR, "msg", "tooltip"))
self.message_broker.publish(StatusData(CTX_WARNING, "msg2", "tooltip2"))
self.assertEqual(self.get_text(), "msg2")
self.assertEqual(self.get_tooltip(), "tooltip2")
self.assert_warning_status()
def test_data_without_message_removes_messages(self):
self.message_broker.send(StatusData(CTX_WARNING, "msg", "tooltip"))
self.message_broker.send(StatusData(CTX_WARNING, "msg2", "tooltip2"))
self.message_broker.send(StatusData(CTX_WARNING))
self.message_broker.publish(StatusData(CTX_WARNING, "msg", "tooltip"))
self.message_broker.publish(StatusData(CTX_WARNING, "msg2", "tooltip2"))
self.message_broker.publish(StatusData(CTX_WARNING))
self.assert_empty()
def test_restores_message_from_not_removed_ctx_id(self):
self.message_broker.send(StatusData(CTX_ERROR, "msg", "tooltip"))
self.message_broker.send(StatusData(CTX_WARNING, "msg2", "tooltip2"))
self.message_broker.send(StatusData(CTX_WARNING))
self.message_broker.publish(StatusData(CTX_ERROR, "msg", "tooltip"))
self.message_broker.publish(StatusData(CTX_WARNING, "msg2", "tooltip2"))
self.message_broker.publish(StatusData(CTX_WARNING))
self.assertEqual(self.get_text(), "msg")
self.assert_error_status()
# works also the other way round
self.message_broker.send(StatusData(CTX_ERROR))
self.message_broker.send(StatusData(CTX_WARNING, "msg", "tooltip"))
self.message_broker.send(StatusData(CTX_ERROR, "msg2", "tooltip2"))
self.message_broker.send(StatusData(CTX_ERROR))
self.message_broker.publish(StatusData(CTX_ERROR))
self.message_broker.publish(StatusData(CTX_WARNING, "msg", "tooltip"))
self.message_broker.publish(StatusData(CTX_ERROR, "msg2", "tooltip2"))
self.message_broker.publish(StatusData(CTX_ERROR))
self.assertEqual(self.get_text(), "msg")
self.assert_warning_status()
def test_sets_msg_as_tooltip_if_tooltip_is_none(self):
self.message_broker.send(StatusData(CTX_ERROR, "msg"))
self.message_broker.publish(StatusData(CTX_ERROR, "msg"))
self.assertEqual(self.get_tooltip(), "msg")
@ -853,14 +854,14 @@ class TestAutoloadSwitch(ComponentBaseTest):
self.controller_mock.set_autoload.assert_called_once_with(False)
def test_updates_state(self):
self.message_broker.send(PresetData(None, None, autoload=True))
self.message_broker.publish(PresetData(None, None, autoload=True))
self.assertTrue(self.gui.get_active())
self.message_broker.send(PresetData(None, None, autoload=False))
self.message_broker.publish(PresetData(None, None, autoload=False))
self.assertFalse(self.gui.get_active())
def test_avoids_infinite_recursion(self):
self.message_broker.send(PresetData(None, None, autoload=True))
self.message_broker.send(PresetData(None, None, autoload=False))
self.message_broker.publish(PresetData(None, None, autoload=True))
self.message_broker.publish(PresetData(None, None, autoload=False))
self.controller_mock.set_autoload.assert_not_called()
@ -884,14 +885,14 @@ class TestReleaseCombinationSwitch(ComponentBaseTest):
)
def test_updates_state(self):
self.message_broker.send(MappingData(release_combination_keys=True))
self.message_broker.publish(MappingData(release_combination_keys=True))
self.assertTrue(self.gui.get_active())
self.message_broker.send(MappingData(release_combination_keys=False))
self.message_broker.publish(MappingData(release_combination_keys=False))
self.assertFalse(self.gui.get_active())
def test_avoids_infinite_recursion(self):
self.message_broker.send(MappingData(release_combination_keys=True))
self.message_broker.send(MappingData(release_combination_keys=False))
self.message_broker.publish(MappingData(release_combination_keys=True))
self.message_broker.publish(MappingData(release_combination_keys=False))
self.controller_mock.update_mapping.assert_not_called()
@ -921,7 +922,7 @@ class TestCombinationListbox(ComponentBaseTest):
self.message_broker, self.controller_mock, self.gui
)
self.controller_mock.is_empty_mapping.return_value = False
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination="1,1,1+3,0,1+1,2,1", target_uinput="keyboard")
)
@ -952,17 +953,17 @@ class TestCombinationListbox(ComponentBaseTest):
def test_does_not_create_rows_when_mapping_is_empty(self):
self.controller_mock.is_empty_mapping.return_value = True
self.message_broker.send(MappingData(event_combination="1,1,1+3,0,1"))
self.message_broker.publish(MappingData(event_combination="1,1,1+3,0,1"))
self.assertEqual(len(self.gui.get_children()), 0)
def test_selects_row_when_selected_event_message_arrives(self):
self.message_broker.send(InputEvent.from_string("3,0,1"))
self.message_broker.publish(InputEvent.from_string("3,0,1"))
self.assertEqual(
self.get_selected_row().input_event, InputEvent.from_string("3,0,1")
)
def test_avoids_infinite_recursion(self):
self.message_broker.send(InputEvent.from_string("3,0,1"))
self.message_broker.publish(InputEvent.from_string("3,0,1"))
self.controller_mock.load_event.assert_not_called()
@ -982,29 +983,29 @@ class TestAnalogInputSwitch(ComponentBaseTest):
self.controller_mock.set_event_as_analog.assert_called_once_with(False)
def test_updates_state(self):
self.message_broker.send(InputEvent.from_string("3,0,0"))
self.message_broker.publish(InputEvent.from_string("3,0,0"))
self.assertTrue(self.gui.get_active())
self.message_broker.send(InputEvent.from_string("3,0,10"))
self.message_broker.publish(InputEvent.from_string("3,0,10"))
self.assertFalse(self.gui.get_active())
def test_avoids_infinite_recursion(self):
self.message_broker.send(InputEvent.from_string("3,0,0"))
self.message_broker.send(InputEvent.from_string("3,0,-10"))
self.message_broker.publish(InputEvent.from_string("3,0,0"))
self.message_broker.publish(InputEvent.from_string("3,0,-10"))
self.controller_mock.set_event_as_analog.assert_not_called()
def test_disables_switch_when_key_event(self):
self.message_broker.send(InputEvent.from_string("1,1,1"))
self.message_broker.publish(InputEvent.from_string("1,1,1"))
self.assertLess(self.gui.get_opacity(), 0.6)
self.assertFalse(self.gui.get_sensitive())
def test_enables_switch_when_axis_event(self):
self.message_broker.send(InputEvent.from_string("1,1,1"))
self.message_broker.send(InputEvent.from_string("3,0,10"))
self.message_broker.publish(InputEvent.from_string("1,1,1"))
self.message_broker.publish(InputEvent.from_string("3,0,10"))
self.assertEqual(self.gui.get_opacity(), 1)
self.assertTrue(self.gui.get_sensitive())
self.message_broker.send(InputEvent.from_string("1,1,1"))
self.message_broker.send(InputEvent.from_string("2,0,10"))
self.message_broker.publish(InputEvent.from_string("1,1,1"))
self.message_broker.publish(InputEvent.from_string("2,0,10"))
self.assertEqual(self.gui.get_opacity(), 1)
self.assertTrue(self.gui.get_sensitive())
@ -1016,7 +1017,7 @@ class TestTriggerThresholdInput(ComponentBaseTest):
self.input = TriggerThresholdInput(
self.message_broker, self.controller_mock, self.gui
)
self.message_broker.send(InputEvent.from_string("3,0,-10"))
self.message_broker.publish(InputEvent.from_string("3,0,-10"))
def assert_abs_event_config(self):
self.assertEqual(self.gui.get_range(), (-99, 99))
@ -1039,18 +1040,18 @@ class TestTriggerThresholdInput(ComponentBaseTest):
)
def test_sets_value_on_selected_event_message(self):
self.message_broker.send(InputEvent.from_string("3,0,10"))
self.message_broker.publish(InputEvent.from_string("3,0,10"))
self.assertEqual(self.gui.get_value(), 10)
def test_avoids_infinite_recursion(self):
self.message_broker.send(InputEvent.from_string("3,0,10"))
self.message_broker.publish(InputEvent.from_string("3,0,10"))
self.controller_mock.update_event.assert_not_called()
def test_updates_configuration_according_to_selected_event(self):
self.assert_abs_event_config()
self.message_broker.send(InputEvent.from_string("2,0,-10"))
self.message_broker.publish(InputEvent.from_string("2,0,-10"))
self.assert_rel_event_config()
self.message_broker.send(InputEvent.from_string("1,1,1"))
self.message_broker.publish(InputEvent.from_string("1,1,1"))
self.assert_key_event_config()
@ -1061,14 +1062,14 @@ class TestReleaseTimeoutInput(ComponentBaseTest):
self.input = ReleaseTimeoutInput(
self.message_broker, self.controller_mock, self.gui
)
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination("2,0,1"), target_uinput="keyboard"
)
)
def test_updates_timeout_on_mapping_message(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination=EventCombination("2,0,1"), release_timeout=1)
)
self.assertEqual(self.gui.get_value(), 1)
@ -1078,28 +1079,28 @@ class TestReleaseTimeoutInput(ComponentBaseTest):
self.controller_mock.update_mapping.assert_called_once_with(release_timeout=0.5)
def test_avoids_infinite_recursion(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination=EventCombination("2,0,1"), release_timeout=1)
)
self.controller_mock.update_mapping.assert_not_called()
def test_disables_input_based_on_input_combination(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination=EventCombination.from_string("2,0,1+1,1,1"))
)
self.assertTrue(self.gui.get_sensitive())
self.assertEqual(self.gui.get_opacity(), 1)
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination=EventCombination.from_string("1,1,1+1,2,1"))
)
self.assertFalse(self.gui.get_sensitive())
self.assertLess(self.gui.get_opacity(), 0.6)
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination=EventCombination.from_string("2,0,1+1,1,1"))
)
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination=EventCombination.from_string("3,0,1+1,2,1"))
)
self.assertFalse(self.gui.get_sensitive())
@ -1114,7 +1115,7 @@ class TestOutputAxisSelector(ComponentBaseTest):
self.message_broker, self.controller_mock, self.gui
)
absinfo = evdev.AbsInfo(0, -10, 10, 0, 0, 0)
self.message_broker.send(
self.message_broker.publish(
UInputsData(
{
"mouse": {1: [1, 2, 3, 4], 2: [0, 1, 2, 3]},
@ -1126,7 +1127,7 @@ class TestOutputAxisSelector(ComponentBaseTest):
}
)
)
self.message_broker.send(
self.message_broker.publish(
MappingData(target_uinput="mouse", event_combination="1,1,1")
)
@ -1152,22 +1153,22 @@ class TestOutputAxisSelector(ComponentBaseTest):
def test_selects_correct_entry(self):
self.assertEqual(self.gui.get_active_id(), "None, None")
self.message_broker.send(
self.message_broker.publish(
MappingData(target_uinput="mouse", output_type=2, output_code=3)
)
self.assertEqual(self.get_active_selection(), (2, 3))
def test_avoids_infinite_recursion(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(target_uinput="mouse", output_type=2, output_code=3)
)
self.controller_mock.update_mapping.assert_not_called()
def test_updates_dropdown_model(self):
self.assertEqual(len(self.gui.get_model()), 5)
self.message_broker.send(MappingData(target_uinput="keyboard"))
self.message_broker.publish(MappingData(target_uinput="keyboard"))
self.assertEqual(len(self.gui.get_model()), 1)
self.message_broker.send(MappingData(target_uinput="gamepad"))
self.message_broker.publish(MappingData(target_uinput="gamepad"))
self.assertEqual(len(self.gui.get_model()), 9)
@ -1207,12 +1208,12 @@ class TestKeyAxisStackSwitcher(ComponentBaseTest):
self.assertTrue(self.analog_toggle.get_active())
def test_switches_to_axis(self):
self.message_broker.send(MappingData(mapping_type="analog"))
self.message_broker.publish(MappingData(mapping_type="analog"))
self.assert_analog_active()
def test_switches_to_key_macro(self):
self.message_broker.send(MappingData(mapping_type="analog"))
self.message_broker.send(MappingData(mapping_type="key_macro"))
self.message_broker.publish(MappingData(mapping_type="analog"))
self.message_broker.publish(MappingData(mapping_type="key_macro"))
self.assert_key_macro_active()
def test_updates_mapping_type(self):
@ -1228,8 +1229,8 @@ class TestKeyAxisStackSwitcher(ComponentBaseTest):
)
def test_avoids_infinite_recursion(self):
self.message_broker.send(MappingData(mapping_type="analog"))
self.message_broker.send(MappingData(mapping_type="key_macro"))
self.message_broker.publish(MappingData(mapping_type="analog"))
self.message_broker.publish(MappingData(mapping_type="key_macro"))
self.controller_mock.update_mapping.assert_not_called()
@ -1257,7 +1258,7 @@ class TestTransformationDrawArea(ComponentBaseTest):
def test_updates_transform_when_mapping_updates(self):
old_tf = self.transform_draw_area._transformation
self.message_broker.send(MappingData(gain=2))
self.message_broker.publish(MappingData(gain=2))
self.assertIsNot(old_tf, self.transform_draw_area._transformation)
def test_redraws_when_mapping_updates(self):
@ -1265,7 +1266,7 @@ class TestTransformationDrawArea(ComponentBaseTest):
gtk_iteration(20)
mock = MagicMock()
self.draw_area.connect("draw", mock)
self.message_broker.send(MappingData(gain=2))
self.message_broker.publish(MappingData(gain=2))
gtk_iteration(20)
mock.assert_called()
@ -1290,7 +1291,7 @@ class TestSliders(ComponentBaseTest):
self.deadzone,
self.expo,
)
self.message_broker.send(
self.message_broker.publish(
MappingData(event_combination="3,0,0", target_uinput="mouse")
)
@ -1311,7 +1312,7 @@ class TestSliders(ComponentBaseTest):
self.assertEqual(self.get_range(self.expo), (-1, 1))
def test_updates_value(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
gain=0.5,
deadzone=0.6,
@ -1335,11 +1336,11 @@ class TestSliders(ComponentBaseTest):
self.controller_mock.update_mapping.assert_called_once_with(deadzone=0.5)
def test_avoids_recursion(self):
self.message_broker.send(MappingData(gain=0.5))
self.message_broker.publish(MappingData(gain=0.5))
self.controller_mock.update_mapping.assert_not_called()
self.message_broker.send(MappingData(expo=0.5))
self.message_broker.publish(MappingData(expo=0.5))
self.controller_mock.update_mapping.assert_not_called()
self.message_broker.send(MappingData(deadzone=0.5))
self.message_broker.publish(MappingData(deadzone=0.5))
self.controller_mock.update_mapping.assert_not_called()
@ -1350,7 +1351,7 @@ class TestRelativeInputCutoffInput(ComponentBaseTest):
self.input = RelativeInputCutoffInput(
self.message_broker, self.controller_mock, self.gui
)
self.message_broker.send(
self.message_broker.publish(
MappingData(
target_uinput="mouse",
event_combination="2,0,0",
@ -1369,7 +1370,7 @@ class TestRelativeInputCutoffInput(ComponentBaseTest):
self.assertLess(self.gui.get_opacity(), 0.6)
def test_avoids_infinite_recursion(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
target_uinput="mouse",
event_combination="2,0,0",
@ -1381,7 +1382,7 @@ class TestRelativeInputCutoffInput(ComponentBaseTest):
self.controller_mock.update_mapping.assert_not_called()
def test_updates_value(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
target_uinput="mouse",
event_combination="2,0,0",
@ -1398,7 +1399,7 @@ class TestRelativeInputCutoffInput(ComponentBaseTest):
def test_disables_input_when_no_rel_axis_input(self):
self.assert_active()
self.message_broker.send(
self.message_broker.publish(
MappingData(
target_uinput="mouse",
event_combination="3,0,0",
@ -1410,7 +1411,7 @@ class TestRelativeInputCutoffInput(ComponentBaseTest):
def test_disables_input_when_no_abs_axis_output(self):
self.assert_active()
self.message_broker.send(
self.message_broker.publish(
MappingData(
target_uinput="mouse",
event_combination="2,0,0",
@ -1422,7 +1423,7 @@ class TestRelativeInputCutoffInput(ComponentBaseTest):
self.assert_inactive()
def test_enables_input(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
target_uinput="mouse",
event_combination="3,0,0",
@ -1431,7 +1432,7 @@ class TestRelativeInputCutoffInput(ComponentBaseTest):
)
)
self.assert_inactive()
self.message_broker.send(
self.message_broker.publish(
MappingData(
target_uinput="mouse",
event_combination="2,0,0",
@ -1453,21 +1454,21 @@ class TestRequireActiveMapping(ComponentBaseTest):
)
combination = EventCombination([(1, KEY_A, 1)])
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
self.assert_inactive(self.box)
self.message_broker.send(PresetData(name="preset", mappings=()))
self.message_broker.publish(PresetData(name="preset", mappings=()))
self.assert_inactive(self.box)
# a mapping is available, that is all the widget needs to be activated. one
# mapping is always selected, so there is no need to check the mapping message
self.message_broker.send(PresetData(name="preset", mappings=(combination,)))
self.message_broker.publish(PresetData(name="preset", mappings=(combination,)))
self.assert_active(self.box)
self.message_broker.send(MappingData(event_combination=combination))
self.message_broker.publish(MappingData(event_combination=combination))
self.assert_active(self.box)
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
self.assert_active(self.box)
def test_recorded_input_required(self):
@ -1479,21 +1480,21 @@ class TestRequireActiveMapping(ComponentBaseTest):
)
combination = EventCombination([(1, KEY_A, 1)])
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
self.assert_inactive(self.box)
self.message_broker.send(PresetData(name="preset", mappings=()))
self.message_broker.publish(PresetData(name="preset", mappings=()))
self.assert_inactive(self.box)
self.message_broker.send(PresetData(name="preset", mappings=(combination,)))
self.message_broker.publish(PresetData(name="preset", mappings=(combination,)))
self.assert_inactive(self.box)
# the widget will be enabled once a mapping with recorded input is selected
self.message_broker.send(MappingData(event_combination=combination))
self.message_broker.publish(MappingData(event_combination=combination))
self.assert_active(self.box)
# this mapping doesn't have input recorded, so the box is disabled
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
self.assert_inactive(self.box)
def assert_inactive(self, widget: Gtk.Widget):
@ -1515,13 +1516,13 @@ class TestStack(ComponentBaseTest):
self.stack.show_all()
stack_wrapper = Stack(self.message_broker, self.controller_mock, self.stack)
self.message_broker.send(DoStackSwitch(Stack.devices_page))
self.message_broker.publish(DoStackSwitch(Stack.devices_page))
self.assertEqual(self.stack.get_visible_child_name(), "Devices")
self.message_broker.send(DoStackSwitch(Stack.presets_page))
self.message_broker.publish(DoStackSwitch(Stack.presets_page))
self.assertEqual(self.stack.get_visible_child_name(), "Presets")
self.message_broker.send(DoStackSwitch(Stack.editor_page))
self.message_broker.publish(DoStackSwitch(Stack.editor_page))
self.assertEqual(self.stack.get_visible_child_name(), "Editor")
@ -1575,7 +1576,7 @@ class TestBreadcrumbs(ComponentBaseTest):
self.assertEqual(self.label_4.get_text(), "? / ? / ?")
self.assertEqual(self.label_5.get_text(), "?")
self.message_broker.send(PresetData("preset", None))
self.message_broker.publish(PresetData("preset", None))
self.assertEqual(self.label_1.get_text(), "")
self.assertEqual(self.label_2.get_text(), "?")
@ -1583,7 +1584,7 @@ class TestBreadcrumbs(ComponentBaseTest):
self.assertEqual(self.label_4.get_text(), "? / preset / ?")
self.assertEqual(self.label_5.get_text(), "?")
self.message_broker.send(GroupData("group", ()))
self.message_broker.publish(GroupData("group", ()))
self.assertEqual(self.label_1.get_text(), "")
self.assertEqual(self.label_2.get_text(), "group")
@ -1591,7 +1592,7 @@ class TestBreadcrumbs(ComponentBaseTest):
self.assertEqual(self.label_4.get_text(), "group / preset / ?")
self.assertEqual(self.label_5.get_text(), "?")
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
self.assertEqual(self.label_1.get_text(), "")
self.assertEqual(self.label_2.get_text(), "group")
@ -1599,16 +1600,18 @@ class TestBreadcrumbs(ComponentBaseTest):
self.assertEqual(self.label_4.get_text(), "group / preset / Empty Mapping")
self.assertEqual(self.label_5.get_text(), "Empty Mapping")
self.message_broker.send(MappingData(name="mapping"))
self.message_broker.publish(MappingData(name="mapping"))
self.assertEqual(self.label_4.get_text(), "group / preset / mapping")
self.assertEqual(self.label_5.get_text(), "mapping")
combination = EventCombination([(1, KEY_A, 1), (1, KEY_B, 1)])
self.message_broker.send(MappingData(event_combination=combination))
self.message_broker.publish(MappingData(event_combination=combination))
self.assertEqual(self.label_4.get_text(), "group / preset / a + b")
self.assertEqual(self.label_5.get_text(), "a + b")
combination = EventCombination([(1, KEY_A, 1)])
self.message_broker.send(MappingData(name="qux", event_combination=combination))
self.message_broker.publish(
MappingData(name="qux", event_combination=combination)
)
self.assertEqual(self.label_4.get_text(), "group / preset / qux")
self.assertEqual(self.label_5.get_text(), "qux")

@ -26,6 +26,9 @@ import multiprocessing
import unittest
import time
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
from inputremapper.daemon import Daemon, BUS_NAME

@ -58,13 +58,14 @@ from unittest.mock import patch, MagicMock, call
from importlib.util import spec_from_loader, module_from_spec
from importlib.machinery import SourceFileLoader
import gi
from inputremapper.input_event import InputEvent
import gi
gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0")
gi.require_version("GLib", "2.0")
gi.require_version("GtkSource", "4")
gi.require_version("GLib", "2.0")
from gi.repository import Gtk, GLib, Gdk, GtkSource
from inputremapper.configs.system_mapping import system_mapping
@ -81,7 +82,7 @@ from inputremapper.gui.messages.message_data import StatusData, CombinationRecor
from inputremapper.gui.components.editor import MappingSelectionLabel, SET_KEY_FIRST
from inputremapper.gui.components.device_groups import DeviceGroupEntry
from inputremapper.gui.controller import Controller
from inputremapper.gui.helper import RootHelper
from inputremapper.gui.reader_service import ReaderService
from inputremapper.gui.utils import gtk_iteration, Colors
from inputremapper.gui.user_interface import UserInterface
from inputremapper.injection.injector import InjectorState
@ -131,16 +132,16 @@ def launch(
@contextmanager
def patch_launch():
"""patch the launch function such that we don't connect to
the dbus and don't use pkexec to start the helper"""
the dbus and don't use pkexec to start the reader-service"""
original_connect = Daemon.connect
original_os_system = os.system
Daemon.connect = Daemon
def os_system(cmd):
# instead of running pkexec, fork instead. This will make
# the helper aware of all the test patches
if "pkexec input-remapper-control --command helper" in cmd:
multiprocessing.Process(target=RootHelper(_Groups()).run).start()
# the reader-service aware of all the test patches
if "pkexec input-remapper-control --command start-reader-service" in cmd:
multiprocessing.Process(target=ReaderService(_Groups()).run).start()
return 0
return original_os_system(cmd)
@ -173,7 +174,7 @@ class GtkKeyEvent:
return True, self.keyval
class TestGroupsFromHelper(unittest.TestCase):
class TestGroupsFromReaderService(unittest.TestCase):
def setUp(self):
# don't try to connect, return an object instance of it instead
self.original_connect = Daemon.connect
@ -183,13 +184,13 @@ class TestGroupsFromHelper(unittest.TestCase):
# because we want to discover the groups as early a possible, to reduce startup
# time for the application
self.original_os_system = os.system
self.helper_started = MagicMock()
self.reader_service_started = MagicMock()
def os_system(cmd):
# instead of running pkexec, fork instead. This will make
# the helper aware of all the test patches
if "pkexec input-remapper-control --command helper" in cmd:
self.helper_started() # don't start the helper just log that it was.
# the reader-service aware of all the test patches
if "pkexec input-remapper-control --command start-reader-service" in cmd:
self.reader_service_started() # don't start the reader-service just log that it was.
return 0
return self.original_os_system(cmd)
@ -210,14 +211,14 @@ class TestGroupsFromHelper(unittest.TestCase):
def test_knows_devices(self):
# verify that it is working as expected. The gui doesn't have knowledge
# of groups until the root-helper provides them
self.data_manager._reader.groups.set_groups([])
# of groups until the root-reader-service provides them
self.data_manager._reader_client.groups.set_groups([])
gtk_iteration()
self.helper_started.assert_called()
self.reader_service_started.assert_called()
self.assertEqual(len(self.data_manager.get_group_keys()), 0)
# start the helper delayed
multiprocessing.Process(target=RootHelper(_Groups()).run).start()
# start the reader-service delayed
multiprocessing.Process(target=ReaderService(_Groups()).run).start()
# perform some iterations so that the reader ends up reading from the pipes
# which will make it receive devices.
for _ in range(10):
@ -276,6 +277,8 @@ class GuiTestBase(unittest.TestCase):
self.daemon,
) = launch()
self._test_initial_state()
get = self.user_interface.get
self.device_selection: Gtk.FlowBox = get("device_selection")
self.preset_selection: Gtk.ComboBoxText = get("preset_selection")
@ -316,8 +319,25 @@ class GuiTestBase(unittest.TestCase):
def tearDown(self):
clean_up_integration(self)
# this is important, otherwise it keeps breaking things in the background
self.assertIsNone(self.data_manager._reader_client._read_timeout)
self.throttle()
def _test_initial_state(self):
# make sure each test deals with the same initial state
self.assertEqual(self.controller.data_manager, self.data_manager)
self.assertEqual(self.data_manager.active_group.key, "Foo Device")
# if the modification-date from `prepare_presets` is not destroyed, preset3
# should be selected as the newest one
self.assertEqual(self.data_manager.active_preset.name, "preset3")
self.assertEqual(self.data_manager.active_mapping.target_uinput, "keyboard")
self.assertEqual(
self.data_manager.active_mapping.event_combination,
EventCombination((1, 5, 1)),
)
self.assertEqual(self.data_manager.active_event, InputEvent(0, 0, 1, 5, 1))
def _callTestMethod(self, method):
"""Retry all tests if they fail.
@ -616,7 +636,7 @@ class TestGui(GuiTestBase):
self.assertFalse(self.recording_status.get_visible())
self.assertFalse(self.recording_toggle.get_active())
def test_events_from_helper_arrive(self):
def test_events_from_reader_service_arrive(self):
# load a device with more capabilities
self.controller.load_group("Foo Device 2")
gtk_iteration()
@ -686,7 +706,7 @@ class TestGui(GuiTestBase):
EventCombination.empty_combination(),
)
# try to recorde the same combination
# try to record the same combination
self.controller.start_key_recording()
push_events(
fixtures.foo_device_2_keyboard,
@ -699,7 +719,7 @@ class TestGui(GuiTestBase):
EventCombination.empty_combination(),
)
# try to recorde a different combination
# try to record a different combination
self.controller.start_key_recording()
push_events(fixtures.foo_device_2_keyboard, [InputEvent.from_string("1,30,1")])
self.throttle(20)
@ -736,7 +756,7 @@ class TestGui(GuiTestBase):
self.throttle(20)
# sending a combination update now should not do anything
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,35,1"))
)
gtk_iteration()
@ -768,7 +788,7 @@ class TestGui(GuiTestBase):
self.assertEqual(len(self.selection_label_listbox.get_children()), 2)
self.assertEqual(len(self.data_manager.active_preset), 2)
# 2. recorde a combination for that mapping
# 2. record a combination for that mapping
self.recording_toggle.set_active(True)
gtk_iteration()
push_events(fixtures.foo_device_2_keyboard, [InputEvent.from_string("1,30,1")])
@ -828,12 +848,12 @@ class TestGui(GuiTestBase):
)
def test_show_status(self):
self.message_broker.send(StatusData(0, "a" * 500))
self.message_broker.publish(StatusData(0, "a" * 500))
gtk_iteration()
text = self.get_status_text()
self.assertIn("...", text)
self.message_broker.send(StatusData(0, "b"))
self.message_broker.publish(StatusData(0, "b"))
gtk_iteration()
text = self.get_status_text()
self.assertNotIn("...", text)
@ -894,7 +914,7 @@ class TestGui(GuiTestBase):
self.controller.load_group("Foo Device 2")
gtk_iteration()
# it should be possible to write a combination combination
# it should be possible to write a combination
ev_1 = InputEvent.from_tuple((EV_KEY, evdev.ecodes.KEY_A, 1))
ev_2 = InputEvent.from_tuple((EV_ABS, evdev.ecodes.ABS_HAT0X, 1))
ev_3 = InputEvent.from_tuple((EV_KEY, evdev.ecodes.KEY_C, 1))
@ -1078,7 +1098,7 @@ class TestGui(GuiTestBase):
self.recording_toggle.set_active(True)
gtk_iteration()
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination((EV_KEY, KEY_Q, 1)))
)
gtk_iteration()
@ -1098,7 +1118,7 @@ class TestGui(GuiTestBase):
gtk_iteration()
self.recording_toggle.set_active(True)
gtk_iteration()
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination((EV_KEY, KEY_Q, 1)))
)
gtk_iteration()
@ -1577,9 +1597,9 @@ class TestGui(GuiTestBase):
self.assertEqual(self.data_manager.get_state(), InjectorState.RUNNING)
# this is a stupid workaround for the bad test fixtures
# by switching the group we make sure that the helper no longer listens for
# events on "Foo Device 2" otherwise we would have two processes
# (helper and injector) reading the same pipe which can block this test
# by switching the group we make sure that the reader-service no longer
# listens for events on "Foo Device 2" otherwise we would have two processes
# (reader-service and injector) reading the same pipe which can block this test
# indefinitely
self.controller.load_group("Foo Device")
gtk_iteration()
@ -1703,9 +1723,11 @@ class TestGui(GuiTestBase):
self.controller.refresh_groups()
gtk_iteration()
self.throttle(100)
# the newest preset should be selected
# the gui should not jump to a different preset suddenly
self.assertEqual(self.data_manager.active_preset.name, "preset1")
# just to verify that the mtime still tells us that preset3 is the newest one
self.assertEqual(self.controller.get_a_preset(), "preset3")
self.assertEqual(self.data_manager.active_preset.name, "preset3")
# the list contains correct entries
# and the non-existing entry should be removed
@ -1725,8 +1747,8 @@ class TestGui(GuiTestBase):
# it won't crash due to "list index out of range"
# when `types` is an empty list. Won't show an icon
self.data_manager._reader.groups.find(key="Foo Device 2").types = []
self.data_manager._reader.send_groups()
self.data_manager._reader_client.groups.find(key="Foo Device 2").types = []
self.data_manager._reader_client.publish_groups()
gtk_iteration()
self.assertIn(
"Foo Device 2",

@ -90,7 +90,7 @@ class TestUserInterface(unittest.TestCase):
mock.assert_called_once()
def test_combination_label_shows_combination(self):
self.message_broker.send(
self.message_broker.publish(
MappingData(
event_combination=EventCombination((EV_KEY, KEY_A, 1)), name="foo"
)
@ -101,7 +101,7 @@ class TestUserInterface(unittest.TestCase):
self.assertEqual(label.get_opacity(), 1)
def test_combination_label_shows_text_when_empty_mapping(self):
self.message_broker.send(MappingData())
self.message_broker.publish(MappingData())
gtk_iteration()
label: Gtk.Label = self.user_interface.get("combination-label")
self.assertEqual(label.get_text(), "no input configured")

@ -145,7 +145,7 @@ if is_service_running():
EVENT_READ_TIMEOUT = 0.01
# based on experience how much time passes at most until
# the helper starts receiving previously pushed events after a
# the reader-service starts receiving previously pushed events after a
# call to start_reading
START_READING_DELAY = 0.05
@ -449,14 +449,14 @@ fixtures = _Fixtures()
def setup_pipe(fixture: Fixture):
"""Create a pipe that can be used to send events to the helper,
which in turn will be sent to the reader
"""Create a pipe that can be used to send events to the reader-service,
which in turn will be sent to the reader-client
"""
if pending_events.get(fixture) is None:
pending_events[fixture] = multiprocessing.Pipe()
# make sure those pipes exist before any process (the helper) gets forked,
# make sure those pipes exist before any process (the reader-service) gets forked,
# so that events can be pushed after the fork.
for _fixture in fixtures:
setup_pipe(_fixture)
@ -791,7 +791,8 @@ from inputremapper.configs.mapping import Mapping, UIMapping
from inputremapper.groups import groups, _Groups
from inputremapper.configs.system_mapping import system_mapping
from inputremapper.gui.messages.message_broker import MessageBroker
from inputremapper.gui.reader import Reader
from inputremapper.gui.reader_client import ReaderClient
from inputremapper.gui.reader_service import ReaderService
from inputremapper.configs.paths import get_config_path, get_preset_path
from inputremapper.configs.preset import Preset
@ -804,6 +805,14 @@ Injector.regrab_timeout = 0.05
environ_copy = copy.deepcopy(os.environ)
def is_running_patch():
logger.info("is_running is patched to always return True")
return True
setattr(ReaderService, "is_running", is_running_patch)
def convert_to_internal_events(events):
"""Convert an iterable of InputEvent to a list of inputremapper.InputEvent."""
return [InternalInputEvent.from_event(event) for event in events]

@ -299,9 +299,13 @@ class TestControl(unittest.TestCase):
def test_internals(self):
with mock.patch("os.system") as os_system_patch:
internals(options("helper", None, None, None, False, False, False))
internals(
options("start-reader-service", None, None, None, False, False, False)
)
os_system_patch.assert_called_once()
self.assertIn("input-remapper-helper", os_system_patch.call_args.args[0])
self.assertIn(
"input-remapper-reader-service", os_system_patch.call_args.args[0]
)
self.assertNotIn("-d", os_system_patch.call_args.args[0])
with mock.patch("os.system") as os_system_patch:

@ -28,13 +28,9 @@ from inputremapper.configs.system_mapping import system_mapping
from inputremapper.injection.injector import InjectorState
from inputremapper.input_event import InputEvent
gi.require_version("Gdk", "3.0")
gi.require_version("Gtk", "3.0")
gi.require_version("GLib", "2.0")
gi.require_version("GtkSource", "4")
from gi.repository import Gtk
# from inputremapper.gui.helper import is_helper_running
from inputremapper.event_combination import EventCombination
from inputremapper.groups import _Groups
from inputremapper.gui.messages.message_broker import (
@ -50,7 +46,7 @@ from inputremapper.gui.messages.message_data import (
CombinationUpdate,
UserConfirmRequest,
)
from inputremapper.gui.reader import Reader
from inputremapper.gui.reader_client import ReaderClient
from inputremapper.gui.utils import CTX_ERROR, CTX_APPLY, gtk_iteration
from inputremapper.gui.gettext import _
from inputremapper.injection.global_uinputs import GlobalUInputs
@ -78,7 +74,7 @@ class TestController(unittest.TestCase):
self.data_manager = DataManager(
self.message_broker,
GlobalConfig(),
Reader(self.message_broker, _Groups()),
ReaderClient(self.message_broker, _Groups()),
FakeDaemonProxy(),
uinputs,
system_mapping,
@ -192,29 +188,6 @@ class TestController(unittest.TestCase):
for m in calls:
self.assertEqual(m, UIMapping(**MAPPING_DEFAULTS))
def test_on_init_should_provide_status_if_helper_is_not_running(self):
calls: List[StatusData] = []
def f(data):
calls.append(data)
self.message_broker.subscribe(MessageType.status_msg, f)
with patch("inputremapper.gui.controller.is_helper_running", lambda: False):
self.message_broker.signal(MessageType.init)
self.assertIn(StatusData(CTX_ERROR, _("The helper did not start")), calls)
def test_on_init_should_not_provide_status_if_helper_is_running(self):
calls: List[StatusData] = []
def f(data):
calls.append(data)
self.message_broker.subscribe(MessageType.status_msg, f)
with patch("inputremapper.gui.controller.is_helper_running", lambda: True):
self.message_broker.signal(MessageType.init)
self.assertNotIn(StatusData(CTX_ERROR, _("The helper did not start")), calls)
def test_on_load_group_should_provide_preset(self):
with patch.object(self.data_manager, "load_preset") as mock:
self.controller.load_group("Foo Device")
@ -663,7 +636,7 @@ class TestController(unittest.TestCase):
self.message_broker.subscribe(MessageType.combination_update, f)
self.controller.start_key_recording()
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,10,1"))
)
self.assertEqual(
@ -673,7 +646,7 @@ class TestController(unittest.TestCase):
EventCombination.from_string("1,10,1"),
),
)
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,10,1+1,3,1"))
)
self.assertEqual(
@ -697,7 +670,7 @@ class TestController(unittest.TestCase):
self.message_broker.subscribe(MessageType.combination_update, f)
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,10,1"))
)
self.assertEqual(len(calls), 0)
@ -716,11 +689,11 @@ class TestController(unittest.TestCase):
self.message_broker.subscribe(MessageType.combination_update, f)
self.controller.start_key_recording()
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,10,1"))
)
self.message_broker.signal(MessageType.recording_finished)
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,10,1+1,3,1"))
)
@ -740,11 +713,11 @@ class TestController(unittest.TestCase):
self.message_broker.subscribe(MessageType.combination_update, f)
self.controller.start_key_recording()
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,10,1"))
)
self.controller.stop_key_recording()
self.message_broker.send(
self.message_broker.publish(
CombinationRecorded(EventCombination.from_string("1,10,1+1,3,1"))
)

@ -40,7 +40,7 @@ from inputremapper.gui.messages.message_data import (
PresetData,
CombinationUpdate,
)
from inputremapper.gui.reader import Reader
from inputremapper.gui.reader_client import ReaderClient
from inputremapper.injection.global_uinputs import GlobalUInputs
from inputremapper.input_event import InputEvent
from tests.test import get_key_mapping, quick_cleanup, FakeDaemonProxy, prepare_presets
@ -61,7 +61,7 @@ class Listener:
class TestDataManager(unittest.TestCase):
def setUp(self) -> None:
self.message_broker = MessageBroker()
self.reader = Reader(self.message_broker, _Groups())
self.reader = ReaderClient(self.message_broker, _Groups())
self.uinputs = GlobalUInputs()
self.uinputs.prepare_all()
self.data_manager = DataManager(
@ -857,11 +857,11 @@ class TestDataManager(unittest.TestCase):
self.data_manager.load_group("Foo Device")
self.assertEqual(self.data_manager.get_available_preset_name("foo 1"), "foo 3")
def test_should_send_groups(self):
def test_should_publish_groups(self):
listener = Listener()
self.message_broker.subscribe(MessageType.groups, listener)
self.data_manager.send_groups()
self.data_manager.publish_groups()
data = listener.calls[0]
# we expect a list of tuples with the group key and their device types
@ -901,7 +901,7 @@ class TestDataManager(unittest.TestCase):
listener = Listener()
self.message_broker.subscribe(MessageType.uinputs, listener)
self.data_manager.send_uinputs()
self.data_manager.publish_uinputs()
data = listener.calls[0]
# we expect a list of tuples with the group key and their device types

@ -18,11 +18,8 @@
# 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 inputremapper.configs.mapping import Mapping
from tests.test import new_event, quick_cleanup, get_key_mapping
import unittest
import asyncio
import unittest
import evdev
from evdev.ecodes import (
@ -39,14 +36,14 @@ from evdev.ecodes import (
REL_WHEEL_HI_RES,
)
from inputremapper.configs.global_config import BUTTONS, MOUSE, WHEEL
from inputremapper.injection.context import Context
from inputremapper.configs.mapping import Mapping
from inputremapper.configs.preset import Preset
from inputremapper.configs.system_mapping import system_mapping
from inputremapper.event_combination import EventCombination
from inputremapper.injection.context import Context
from inputremapper.injection.event_reader import EventReader
from inputremapper.configs.system_mapping import system_mapping
from inputremapper.injection.global_uinputs import global_uinputs
from tests.test import new_event, quick_cleanup, get_key_mapping
class TestEventReader(unittest.IsolatedAsyncioTestCase):

@ -24,17 +24,17 @@ class TestMessageBroker(unittest.TestCase):
message_broker = MessageBroker()
listener = Listener()
message_broker.subscribe(MessageType.test1, listener)
message_broker.send(Message(MessageType.test1, "foo"))
message_broker.send(Message(MessageType.test2, "bar"))
message_broker.publish(Message(MessageType.test1, "foo"))
message_broker.publish(Message(MessageType.test2, "bar"))
self.assertEqual(listener.calls[0], Message(MessageType.test1, "foo"))
def test_unsubscribe(self):
message_broker = MessageBroker()
listener = Listener()
message_broker.subscribe(MessageType.test1, listener)
message_broker.send(Message(MessageType.test1, "a"))
message_broker.publish(Message(MessageType.test1, "a"))
message_broker.unsubscribe(listener)
message_broker.send(Message(MessageType.test1, "b"))
message_broker.publish(Message(MessageType.test1, "b"))
self.assertEqual(len(listener.calls), 1)
self.assertEqual(listener.calls[0], Message(MessageType.test1, "a"))
@ -45,7 +45,7 @@ class TestMessageBroker(unittest.TestCase):
listener2 = Listener()
message_broker.subscribe(MessageType.test1, listener1)
message_broker.unsubscribe(listener2)
message_broker.send(Message(MessageType.test1, "a"))
message_broker.publish(Message(MessageType.test1, "a"))
self.assertEqual(listener1.calls[0], Message(MessageType.test1, "a"))
def test_preserves_order(self):
@ -53,15 +53,15 @@ class TestMessageBroker(unittest.TestCase):
calls = []
def listener1(_):
message_broker.send(Message(MessageType.test2, "f"))
message_broker.publish(Message(MessageType.test2, "f"))
calls.append(1)
def listener2(_):
message_broker.send(Message(MessageType.test2, "f"))
message_broker.publish(Message(MessageType.test2, "f"))
calls.append(2)
def listener3(_):
message_broker.send(Message(MessageType.test2, "f"))
message_broker.publish(Message(MessageType.test2, "f"))
calls.append(3)
def listener4(_):
@ -71,7 +71,7 @@ class TestMessageBroker(unittest.TestCase):
message_broker.subscribe(MessageType.test1, listener2)
message_broker.subscribe(MessageType.test1, listener3)
message_broker.subscribe(MessageType.test2, listener4)
message_broker.send(Message(MessageType.test1, ""))
message_broker.publish(Message(MessageType.test1, ""))
first = calls[:3]
first.sort()

@ -17,30 +17,14 @@
#
# You should have received a copy of the GNU General Public License
# along with input-remapper. If not, see <https://www.gnu.org/licenses/>.
import json
from typing import List
from inputremapper.gui.messages.message_types import MessageType
from inputremapper.gui.messages.message_broker import (
MessageBroker,
Signal,
)
from inputremapper.gui.messages.message_data import CombinationRecorded
from tests.test import (
new_event,
push_events,
EVENT_READ_TIMEOUT,
START_READING_DELAY,
quick_cleanup,
MAX_ABS,
MIN_ABS,
fixtures,
push_event,
)
import unittest
import time
import os
import json
import multiprocessing
import time
import unittest
from typing import List, Optional
from unittest.mock import patch, MagicMock
from evdev.ecodes import (
EV_KEY,
@ -48,21 +32,35 @@ from evdev.ecodes import (
ABS_HAT0X,
KEY_COMMA,
BTN_TOOL_DOUBLETAP,
ABS_Z,
ABS_Y,
KEY_A,
EV_REL,
REL_WHEEL,
REL_X,
ABS_X,
ABS_RZ,
REL_HWHEEL,
)
from inputremapper.gui.reader import Reader
from inputremapper.event_combination import EventCombination
from inputremapper.gui.helper import RootHelper
from inputremapper.groups import _Groups, DeviceType
from inputremapper.gui.messages.message_broker import (
MessageBroker,
Signal,
)
from inputremapper.gui.messages.message_data import CombinationRecorded
from inputremapper.gui.messages.message_types import MessageType
from inputremapper.gui.reader_client import ReaderClient
from inputremapper.gui.reader_service import ReaderService
from tests.test import (
new_event,
push_events,
EVENT_READ_TIMEOUT,
START_READING_DELAY,
quick_cleanup,
MAX_ABS,
MIN_ABS,
fixtures,
push_event,
)
CODE_1 = 100
CODE_2 = 101
@ -90,33 +88,33 @@ def wait(func, timeout=1.0):
class TestReader(unittest.TestCase):
def setUp(self):
self.helper = None
self.reader_service = None
self.groups = _Groups()
self.message_broker = MessageBroker()
self.reader = Reader(self.message_broker, self.groups)
self.reader_client = ReaderClient(self.message_broker, self.groups)
def tearDown(self):
quick_cleanup()
try:
self.reader.terminate()
self.reader_client.terminate()
except (BrokenPipeError, OSError):
pass
if self.helper is not None:
self.helper.join()
if self.reader_service is not None:
self.reader_service.join()
def create_helper(self, groups: _Groups = None):
# this will cause pending events to be copied over to the helper
def create_reader_service(self, groups: Optional[_Groups] = None):
# this will cause pending events to be copied over to the reader-service
# process
if not groups:
groups = self.groups
def start_helper():
helper = RootHelper(groups)
helper.run()
def start_reader_service():
reader_service = ReaderService(groups)
reader_service.run()
self.helper = multiprocessing.Process(target=start_helper)
self.helper.start()
self.reader_service = multiprocessing.Process(target=start_reader_service)
self.reader_service.start()
time.sleep(0.1)
def test_reading(self):
@ -124,9 +122,9 @@ class TestReader(unittest.TestCase):
l2 = Listener()
self.message_broker.subscribe(MessageType.combination_recorded, l1)
self.message_broker.subscribe(MessageType.recording_finished, l2)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
push_events(fixtures.foo_device_2_gamepad, [new_event(EV_ABS, ABS_HAT0X, 1)])
# we need to sleep because we have two different fixtures,
@ -138,7 +136,7 @@ class TestReader(unittest.TestCase):
time.sleep(0.1)
# read all pending events. Having a glib mainloop would be better,
# as it would call read automatically periodically
self.reader._read()
self.reader_client._read()
self.assertEqual(
[
CombinationRecorded(EventCombination.from_string("3,16,1")),
@ -151,7 +149,7 @@ class TestReader(unittest.TestCase):
# as both the hat and relative axis are released by now
push_events(fixtures.foo_device_2_gamepad, [new_event(EV_ABS, ABS_HAT0X, 0)])
time.sleep(0.3)
self.reader._read()
self.reader_client._read()
self.assertEqual([Signal(MessageType.recording_finished)], l2.calls)
def test_should_release_relative_axis(self):
@ -160,13 +158,13 @@ class TestReader(unittest.TestCase):
l2 = Listener()
self.message_broker.subscribe(MessageType.combination_recorded, l1)
self.message_broker.subscribe(MessageType.recording_finished, l2)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
push_events(fixtures.foo_device_2_mouse, [new_event(EV_REL, REL_X, -5)])
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
[CombinationRecorded(EventCombination.from_string("2,0,-1"))],
@ -175,34 +173,34 @@ class TestReader(unittest.TestCase):
self.assertEqual([], l2.calls) # no stop recording yet
time.sleep(0.3)
self.reader._read()
self.reader_client._read()
self.assertEqual([Signal(MessageType.recording_finished)], l2.calls)
def test_should_not_trigger_at_low_speed_for_rel_axis(self):
l1 = Listener()
self.message_broker.subscribe(MessageType.combination_recorded, l1)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
push_events(fixtures.foo_device_2_mouse, [new_event(EV_REL, REL_X, -1)])
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(0, len(l1.calls))
def test_should_trigger_wheel_at_low_speed(self):
l1 = Listener()
self.message_broker.subscribe(MessageType.combination_recorded, l1)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
push_events(
fixtures.foo_device_2_mouse,
[new_event(EV_REL, REL_WHEEL, -1), new_event(EV_REL, REL_HWHEEL, 1)],
)
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
[
@ -215,17 +213,17 @@ class TestReader(unittest.TestCase):
def test_wont_emit_the_same_combination_twice(self):
l1 = Listener()
self.message_broker.subscribe(MessageType.combination_recorded, l1)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
push_events(fixtures.foo_device_2_keyboard, [new_event(EV_KEY, KEY_A, 1)])
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
# the duplicate event should be ignored
push_events(fixtures.foo_device_2_keyboard, [new_event(EV_KEY, KEY_A, 1)])
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
[CombinationRecorded(EventCombination.from_string("1,30,1"))],
@ -237,9 +235,9 @@ class TestReader(unittest.TestCase):
l2 = Listener()
self.message_broker.subscribe(MessageType.combination_recorded, l1)
self.message_broker.subscribe(MessageType.recording_finished, l2)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
# over 30% should trigger
push_events(
@ -247,7 +245,7 @@ class TestReader(unittest.TestCase):
[new_event(EV_ABS, ABS_X, int(MAX_ABS * 0.4))],
)
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
[CombinationRecorded(EventCombination.from_string("3,0,1"))],
l1.calls,
@ -260,7 +258,7 @@ class TestReader(unittest.TestCase):
[new_event(EV_ABS, ABS_X, int(MAX_ABS * 0.2))],
)
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
[CombinationRecorded(EventCombination.from_string("3,0,1"))],
l1.calls,
@ -270,9 +268,9 @@ class TestReader(unittest.TestCase):
def test_should_change_direction(self):
l1 = Listener()
self.message_broker.subscribe(MessageType.combination_recorded, l1)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
push_event(fixtures.foo_device_2_keyboard, new_event(EV_KEY, KEY_A, 1))
time.sleep(0.1)
@ -290,7 +288,7 @@ class TestReader(unittest.TestCase):
],
)
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
[
CombinationRecorded(EventCombination.from_string("1,30,1")),
@ -326,25 +324,25 @@ class TestReader(unittest.TestCase):
* 3,
)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(l1.calls[0].combination, EventCombination((EV_KEY, 1, 1)))
self.reader.set_group(self.groups.find(name="Bar Device"))
self.reader_client.set_group(self.groups.find(name="Bar Device"))
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
# we did not get the event from the "Bar Device" because the group change
# stopped the recording
self.assertEqual(len(l1.calls), 1)
self.reader.start_recorder()
self.reader_client.start_recorder()
push_events(fixtures.bar_device, [new_event(EV_KEY, 2, 1)])
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(l1.calls[1].combination, EventCombination((EV_KEY, 2, 1)))
def test_reading_2(self):
@ -362,19 +360,19 @@ class TestReader(unittest.TestCase):
pipe = multiprocessing.Pipe()
def refresh():
# from within the helper process notify this test that
# from within the reader-service process notify this test that
# refresh was called as expected
pipe[1].send("refreshed")
groups = _Groups()
groups.refresh = refresh
self.create_helper(groups)
self.create_reader_service(groups)
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
# sending anything arbitrary does not stop the helper
self.reader._commands.send(856794)
# sending anything arbitrary does not stop the reader-service
self.reader_client._commands_pipe.send(856794)
time.sleep(0.2)
push_events(
fixtures.foo_device_2_gamepad,
@ -386,7 +384,7 @@ class TestReader(unittest.TestCase):
self.assertTrue(pipe[0].poll())
self.assertEqual(pipe[0].recv(), "refreshed")
self.reader._read()
self.reader_client._read()
self.assertEqual(
l1.calls[-1].combination,
((EV_KEY, CODE_1, 1), (EV_KEY, CODE_3, 1), (EV_ABS, ABS_HAT0X, -1)),
@ -405,11 +403,11 @@ class TestReader(unittest.TestCase):
],
force=True,
)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
l1.calls[-1].combination, EventCombination((EV_KEY, CODE_2, 1))
)
@ -423,11 +421,11 @@ class TestReader(unittest.TestCase):
[new_event(EV_ABS, ABS_HAT0X, 1), new_event(EV_KEY, CODE_3, 2)],
force=True,
)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
time.sleep(0.2)
self.reader._read()
self.reader_client._read()
self.assertEqual(
l1.calls[-1].combination, EventCombination((EV_ABS, ABS_HAT0X, 1))
)
@ -443,11 +441,11 @@ class TestReader(unittest.TestCase):
new_event(EV_KEY, CODE_3, 0, 12),
],
)
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
time.sleep(0.1)
self.reader._read()
self.reader_client._read()
self.assertEqual(
l1.calls[-1].combination, EventCombination((EV_KEY, CODE_2, 1))
)
@ -464,11 +462,11 @@ class TestReader(unittest.TestCase):
new_event(EV_KEY, CODE_3, 1),
],
)
self.create_helper()
self.reader.set_group(self.groups.find(name="Bar Device"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(name="Bar Device"))
self.reader_client.start_recorder()
time.sleep(EVENT_READ_TIMEOUT * 5)
self.reader._read()
self.reader_client._read()
self.assertEqual(len(l1.calls), 0)
def test_inputremapper_devices(self):
@ -486,43 +484,43 @@ class TestReader(unittest.TestCase):
new_event(EV_KEY, CODE_3, 1),
],
)
self.create_helper()
self.reader.set_group(self.groups.find(name="Bar Device"))
self.reader.start_recorder()
self.create_reader_service()
self.reader_client.set_group(self.groups.find(name="Bar Device"))
self.reader_client.start_recorder()
time.sleep(EVENT_READ_TIMEOUT * 5)
self.reader._read()
self.reader_client._read()
self.assertEqual(len(l1.calls), 0)
def test_terminate(self):
self.create_helper()
self.reader.set_group(self.groups.find(key="Foo Device 2"))
self.create_reader_service()
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
push_events(fixtures.foo_device_2_keyboard, [new_event(EV_KEY, CODE_3, 1)])
time.sleep(START_READING_DELAY + EVENT_READ_TIMEOUT)
self.assertTrue(self.reader._results.poll())
self.assertTrue(self.reader_client._results_pipe.poll())
self.reader.terminate()
self.reader_client.terminate()
time.sleep(EVENT_READ_TIMEOUT)
self.assertFalse(self.reader._results.poll())
self.assertFalse(self.reader_client._results_pipe.poll())
# no new events arrive after terminating
push_events(fixtures.foo_device_2_keyboard, [new_event(EV_KEY, CODE_3, 1)])
time.sleep(EVENT_READ_TIMEOUT * 3)
self.assertFalse(self.reader._results.poll())
self.assertFalse(self.reader_client._results_pipe.poll())
def test_are_new_groups_available(self):
l1 = Listener()
self.message_broker.subscribe(MessageType.groups, l1)
self.create_helper()
self.reader.groups.set_groups({})
self.create_reader_service()
self.reader_client.groups.set_groups({})
time.sleep(0.1) # let the helper send the groups
# read stuff from the helper, which includes the devices
self.assertEqual("[]", self.reader.groups.dumps())
self.reader._read()
time.sleep(0.1) # let the reader-service send the groups
# read stuff from the reader-service, which includes the devices
self.assertEqual("[]", self.reader_client.groups.dumps())
self.reader_client._read()
self.assertEqual(
self.reader.groups.dumps(),
self.reader_client.groups.dumps(),
json.dumps(
[
json.dumps(
@ -595,6 +593,101 @@ class TestReader(unittest.TestCase):
self.assertEqual(len(l1.calls), 1) # ensure we got the event
def test_starts_the_service(self):
# if ReaderClient can't see the ReaderService, a new ReaderService should
# be started via pkexec
with patch.object(ReaderService, "is_running", lambda: False):
os_system_mock = MagicMock(return_value=0)
with patch.object(os, "system", os_system_mock):
# the status message enables the reader-client to see, that the
# reader-service has started
self.reader_client._results_pipe.send(
{"type": "status", "message": "ready"}
)
self.reader_client._send_command("foo")
os_system_mock.assert_called_once_with(
"pkexec input-remapper-control --command start-reader-service -d"
)
def test_wont_start_the_service(self):
# already running, no call to os.system
with patch.object(ReaderService, "is_running", lambda: True):
mock = MagicMock(return_value=0)
with patch.object(os, "system", mock):
self.reader_client._send_command("foo")
mock.assert_not_called()
def test_reader_service_wont_start(self):
# test for the "The reader-service did not start" message
expected_msg = "The reader-service did not start"
subscribe_mock = MagicMock()
self.message_broker.subscribe(MessageType.status_msg, subscribe_mock)
with patch.object(ReaderClient, "_timeout", 1):
with patch.object(ReaderService, "is_running", lambda: False):
os_system_mock = MagicMock(return_value=0)
with patch.object(os, "system", os_system_mock):
self.reader_client._send_command("foo")
# no message is sent into _results_pipe, so the reader-client will
# think the reader-service didn't manage to start
os_system_mock.assert_called_once_with(
"pkexec input-remapper-control "
"--command start-reader-service -d"
)
subscribe_mock.assert_called_once()
status = subscribe_mock.call_args[0][0]
self.assertEqual(status.msg, expected_msg)
def test_reader_service_times_out(self):
# after some time the reader-service just stops, to avoid leaving a hole
# that exposes user-input forever
with patch.object(ReaderService, "_maximum_lifetime", 1):
self.create_reader_service()
self.assertTrue(self.reader_service.is_alive())
time.sleep(0.5)
self.assertTrue(self.reader_service.is_alive())
time.sleep(1)
self.assertFalse(self.reader_service.is_alive())
def test_reader_service_waits_for_client_to_finish(self):
# if the client is currently reading, it waits a bit longer until the
# client finishes reading
with patch.object(ReaderService, "_maximum_lifetime", 1):
self.create_reader_service()
self.assertTrue(self.reader_service.is_alive())
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
time.sleep(2)
# still alive, without start_recorder it should have already exited
self.assertTrue(self.reader_service.is_alive())
self.reader_client.stop_recorder()
time.sleep(1)
self.assertFalse(self.reader_service.is_alive())
def test_reader_service_wont_wait_forever(self):
# if the client is reading forever, stop it after another timeout
with patch.object(ReaderService, "_maximum_lifetime", 1):
with patch.object(ReaderService, "_timeout_tolerance", 1):
self.create_reader_service()
self.assertTrue(self.reader_service.is_alive())
self.reader_client.set_group(self.groups.find(key="Foo Device 2"))
self.reader_client.start_recorder()
time.sleep(1.5)
# still alive, without start_recorder it should have already exited
self.assertTrue(self.reader_service.is_alive())
time.sleep(1)
# now it stopped, even though the reader is still reading
self.assertFalse(self.reader_service.is_alive())
if __name__ == "__main__":
unittest.main()

@ -27,6 +27,7 @@ from tests.test import (
push_events,
EVENT_READ_TIMEOUT,
START_READING_DELAY,
logger,
)
import os
@ -38,8 +39,8 @@ import evdev
from evdev.ecodes import EV_ABS, EV_KEY
from inputremapper.groups import groups, _Groups
from inputremapper.gui.reader import Reader
from inputremapper.gui.helper import RootHelper
from inputremapper.gui.reader_client import ReaderClient
from inputremapper.gui.reader_service import ReaderService
class TestTest(unittest.TestCase):
@ -82,53 +83,56 @@ class TestTest(unittest.TestCase):
self.assertNotIn("foo", environ)
def test_push_events(self):
"""Test that push_event works properly between helper and reader.
"""Test that push_event works properly between reader service and client.
Using push_events after the helper is already forked should work,
Using push_events after the reader-service is already started should work,
as well as using push_event twice
"""
reader = Reader(MessageBroker(), groups)
reader_client = ReaderClient(MessageBroker(), groups)
def create_helper():
# this will cause pending events to be copied over to the helper
def create_reader_service():
# this will cause pending events to be copied over to the reader-service
# process
def start_helper():
def start_reader_service():
# there is no point in using the global groups object
# because the helper runs in a different process
helper = RootHelper(_Groups())
helper.run()
# because the reader-service runs in a different process
reader_service = ReaderService(_Groups())
reader_service.run()
self.helper = multiprocessing.Process(target=start_helper)
self.helper.start()
self.reader_service = multiprocessing.Process(target=start_reader_service)
self.reader_service.start()
time.sleep(0.1)
def wait_for_results():
# wait for the helper to send stuff
# wait for the reader-service to send stuff
for _ in range(10):
time.sleep(EVENT_READ_TIMEOUT)
if reader._results.poll():
if reader_client._results_pipe.poll():
break
create_helper()
reader.set_group(groups.find(key="Foo Device 2"))
create_reader_service()
reader_client.set_group(groups.find(key="Foo Device 2"))
reader_client.start_recorder()
time.sleep(START_READING_DELAY)
event = new_event(EV_KEY, 102, 1)
push_events(fixtures.foo_device_2_keyboard, [event])
wait_for_results()
self.assertTrue(reader._results.poll())
self.assertTrue(reader_client._results_pipe.poll())
reader._read()
self.assertFalse(reader._results.poll())
reader_client._read()
self.assertFalse(reader_client._results_pipe.poll())
# can push more events to the helper that is inside a separate
# can push more events to the reader-service that is inside a separate
# process, which end up being sent to the reader
event = new_event(EV_KEY, 102, 0)
logger.info("push_events")
push_events(fixtures.foo_device_2_keyboard, [event])
wait_for_results()
self.assertTrue(reader._results.poll())
logger.info("assert")
self.assertTrue(reader_client._results_pipe.poll())
reader.terminate()
reader_client.terminate()
if __name__ == "__main__":

Loading…
Cancel
Save