From 1e4ac8759b4f3956095b0905ab0de4f5771a167c Mon Sep 17 00:00:00 2001 From: Tobi Date: Wed, 23 Nov 2022 22:55:28 +0100 Subject: [PATCH] Test setup cleanup (#557) --- .gitignore | 2 +- tests/integration/test_components.py | 7 +- tests/integration/test_gui.py | 22 +- tests/integration/test_numlock.py | 2 +- tests/integration/test_user_interface.py | 2 +- tests/lib/__init__.py | 0 tests/lib/cleanup.py | 179 ++++ tests/lib/constants.py | 32 + tests/lib/fixtures.py | 375 +++++++ tests/lib/logger.py | 54 + tests/lib/patches.py | 344 +++++++ tests/lib/pipes.py | 85 ++ tests/lib/stuff.py | 38 + tests/lib/tmp.py | 28 + tests/{ => lib}/xmodmap.py | 20 + tests/test.py | 933 +----------------- tests/unit/test_config.py | 3 +- tests/unit/test_context.py | 5 +- tests/unit/test_control.py | 3 +- tests/unit/test_controller.py | 11 +- tests/unit/test_daemon.py | 16 +- tests/unit/test_data_manager.py | 10 +- .../test_event_pipeline.py | 13 +- .../test_mapping_handlers.py | 10 +- tests/unit/test_event_reader.py | 4 +- tests/unit/test_global_uinputs.py | 2 +- tests/unit/test_groups.py | 3 +- tests/unit/test_injector.py | 17 +- tests/unit/test_ipc.py | 3 +- tests/unit/test_logger.py | 2 +- tests/unit/test_macros.py | 4 +- tests/unit/test_migrations.py | 4 +- tests/unit/test_paths.py | 3 +- tests/unit/test_preset.py | 3 +- tests/unit/test_reader.py | 11 +- tests/unit/test_system_mapping.py | 2 +- tests/unit/test_test.py | 19 +- tests/unit/test_user.py | 2 +- 38 files changed, 1269 insertions(+), 1004 deletions(-) create mode 100644 tests/lib/__init__.py create mode 100644 tests/lib/cleanup.py create mode 100644 tests/lib/constants.py create mode 100644 tests/lib/fixtures.py create mode 100644 tests/lib/logger.py create mode 100644 tests/lib/patches.py create mode 100644 tests/lib/pipes.py create mode 100644 tests/lib/stuff.py create mode 100644 tests/lib/tmp.py rename tests/{ => lib}/xmodmap.py (93%) diff --git a/.gitignore b/.gitignore index 6b97c1e1..b7780731 100644 --- a/.gitignore +++ b/.gitignore @@ -22,7 +22,7 @@ dist/ downloads/ eggs/ .eggs/ -lib/ +/lib/ lib64/ parts/ sdist/ diff --git a/tests/integration/test_components.py b/tests/integration/test_components.py index 40ca0084..319b5682 100644 --- a/tests/integration/test_components.py +++ b/tests/integration/test_components.py @@ -24,7 +24,7 @@ from unittest.mock import MagicMock import time import evdev -from evdev.ecodes import BTN_LEFT, KEY_A, KEY_B, KEY_C +from evdev.ecodes import KEY_A, KEY_B, KEY_C import gi @@ -36,7 +36,10 @@ gi.require_version("GLib", "2.0") gi.require_version("GtkSource", "4") from gi.repository import Gtk, GLib, GtkSource, Gdk -from tests.test import quick_cleanup, spy, logger +from tests.lib.cleanup import quick_cleanup +from tests.lib.stuff import spy +from tests.lib.logger import logger + from inputremapper.input_event import InputEvent from inputremapper.gui.utils import CTX_ERROR, CTX_WARNING, gtk_iteration from inputremapper.gui.messages.message_broker import ( diff --git a/tests/integration/test_gui.py b/tests/integration/test_gui.py index c14be330..cba93782 100644 --- a/tests/integration/test_gui.py +++ b/tests/integration/test_gui.py @@ -23,19 +23,15 @@ from contextlib import contextmanager from typing import Tuple, List, Optional -from tests.test import ( - get_project_root, - logger, - push_events, - new_event, - spy, - cleanup, - uinput_write_history_pipe, - EVENT_READ_TIMEOUT, - prepare_presets, - fixtures, - push_event, -) +from tests.test import get_project_root +from tests.lib.fixtures import new_event +from tests.lib.cleanup import cleanup +from tests.lib.stuff import spy +from tests.lib.constants import EVENT_READ_TIMEOUT +from tests.lib.fixtures import prepare_presets +from tests.lib.logger import logger +from tests.lib.fixtures import fixtures +from tests.lib.pipes import push_event, push_events, uinput_write_history_pipe from tests.integration.test_components import FlowBoxTestUtils import sys diff --git a/tests/integration/test_numlock.py b/tests/integration/test_numlock.py index 637981f2..7a9dbae0 100644 --- a/tests/integration/test_numlock.py +++ b/tests/integration/test_numlock.py @@ -19,7 +19,7 @@ # along with input-remapper. If not, see . -from tests.test import quick_cleanup +from tests.lib.cleanup import quick_cleanup import unittest diff --git a/tests/integration/test_user_interface.py b/tests/integration/test_user_interface.py index e1623ff1..36f80454 100644 --- a/tests/integration/test_user_interface.py +++ b/tests/integration/test_user_interface.py @@ -11,7 +11,7 @@ gi.require_version("GtkSource", "4") from gi.repository import Gtk, Gdk, GLib from inputremapper.gui.utils import gtk_iteration -from tests.test import quick_cleanup +from tests.lib.cleanup import quick_cleanup from inputremapper.gui.messages.message_broker import MessageBroker, MessageType from inputremapper.gui.user_interface import UserInterface from inputremapper.configs.mapping import MappingData diff --git a/tests/lib/__init__.py b/tests/lib/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/lib/cleanup.py b/tests/lib/cleanup.py new file mode 100644 index 00000000..e33aa155 --- /dev/null +++ b/tests/lib/cleanup.py @@ -0,0 +1,179 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +from __future__ import annotations + +import os +import sys +import shutil +import time +import asyncio +import psutil +from pickle import UnpicklingError +from unittest.mock import patch + +from tests.lib.logger import logger +from tests.lib.pipes import ( + uinput_write_history_pipe, + uinput_write_history, + pending_events, + setup_pipe, +) +from tests.lib.constants import EVENT_READ_TIMEOUT +from tests.lib.tmp import tmp +from tests.lib.fixtures import fixtures +from tests.lib.stuff import environ_copy +from tests.lib.patches import uinputs + + +def join_children(): + """Wait for child processes to exit. Stop them if it takes too long.""" + this = psutil.Process(os.getpid()) + + i = 0 + time.sleep(EVENT_READ_TIMEOUT) + children = this.children(recursive=True) + while len([c for c in children if c.status() != "zombie"]) > 0: + for child in children: + if i > 10: + child.kill() + logger.info("Killed pid %s because it didn't finish in time", child.pid) + + children = this.children(recursive=True) + time.sleep(EVENT_READ_TIMEOUT) + i += 1 + + +def clear_write_history(): + """Empty the history in preparation for the next test.""" + while len(uinput_write_history) > 0: + uinput_write_history.pop() + while uinput_write_history_pipe[0].poll(): + uinput_write_history_pipe[0].recv() + + +def quick_cleanup(log=True): + """Reset the applications state.""" + # Reminder: before patches are applied in test.py, no inputremapper module + # may be imported. So tests.lib imports them just-in-time in functions instead. + from inputremapper.injection.macros.macro import macro_variables + from inputremapper.configs.global_config import global_config + from inputremapper.configs.system_mapping import system_mapping + from inputremapper.gui.utils import debounce_manager + from inputremapper.configs.paths import get_config_path + from inputremapper.injection.global_uinputs import global_uinputs + + if log: + print("Quick cleanup...") + + debounce_manager.stop_all() + + for device in list(pending_events.keys()): + try: + while pending_events[device][1].poll(): + pending_events[device][1].recv() + except (UnpicklingError, EOFError): + pass + + # setup new pipes for the next test + pending_events[device][1].close() + pending_events[device][0].close() + del pending_events[device] + setup_pipe(device) + + try: + if asyncio.get_event_loop().is_running(): + for task in asyncio.all_tasks(): + task.cancel() + except RuntimeError: + # happens when the event loop disappears for magical reasons + # create a fresh event loop + asyncio.set_event_loop(asyncio.new_event_loop()) + + if macro_variables.process is not None and not macro_variables.process.is_alive(): + # nothing should stop the process during runtime, if it has been started by + # the injector once + raise AssertionError("the SharedDict manager is not running anymore") + + if macro_variables.process is not None: + macro_variables._stop() + + join_children() + + macro_variables.start() + + if os.path.exists(tmp): + shutil.rmtree(tmp) + + global_config.path = os.path.join(get_config_path(), "config.json") + global_config.clear_config() + global_config._save_config() + + system_mapping.populate() + + clear_write_history() + + for name in list(uinputs.keys()): + del uinputs[name] + + # for device in list(active_macros.keys()): + # del active_macros[device] + # for device in list(unreleased.keys()): + # del unreleased[device] + fixtures.reset() + os.environ.update(environ_copy) + for device in list(os.environ.keys()): + if device not in environ_copy: + del os.environ[device] + + for _, pipe in pending_events.values(): + assert not pipe.poll() + + assert macro_variables.is_alive(1) + for uinput in global_uinputs.devices.values(): + uinput.write_count = 0 + uinput.write_history = [] + + global_uinputs.is_service = True + + if log: + print("Quick cleanup done") + + +def cleanup(): + """Reset the applications state. + + Using this is slower, usually quick_cleanup() is sufficient. + """ + from inputremapper.groups import groups + from inputremapper.injection.global_uinputs import global_uinputs + + print("Cleanup...") + + os.system("pkill -f input-remapper-service") + os.system("pkill -f input-remapper-control") + time.sleep(0.05) + + quick_cleanup(log=False) + groups.refresh() + with patch.object(sys, "argv", ["input-remapper-service"]): + global_uinputs.prepare_all() + + print("Cleanup done") diff --git a/tests/lib/constants.py b/tests/lib/constants.py new file mode 100644 index 00000000..1b009967 --- /dev/null +++ b/tests/lib/constants.py @@ -0,0 +1,32 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +# give tests some time to test stuff while the process +# is still running +EVENT_READ_TIMEOUT = 0.01 + +# based on experience how much time passes at most until +# the reader-service starts receiving previously pushed events after a +# call to start_reading +START_READING_DELAY = 0.05 + +# for joysticks +MIN_ABS = -(2**15) +MAX_ABS = 2**15 diff --git a/tests/lib/fixtures.py b/tests/lib/fixtures.py new file mode 100644 index 00000000..7c7b385e --- /dev/null +++ b/tests/lib/fixtures.py @@ -0,0 +1,375 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +from __future__ import annotations + +import dataclasses +import json +from typing import Dict, Optional + +import time + +import evdev + + +# input-remapper is only interested in devices that have EV_KEY, add some +# random other stuff to test that they are ignored. +phys_foo = "usb-0000:03:00.0-1/input2" +info_foo = evdev.device.DeviceInfo(1, 1, 1, 1) + +keyboard_keys = sorted(evdev.ecodes.keys.keys())[:255] + + +@dataclasses.dataclass(frozen=True) +class Fixture: + capabilities: Dict = dataclasses.field(default_factory=dict) + path: str = "" + name: str = "unset" + info: evdev.device.DeviceInfo = evdev.device.DeviceInfo(None, None, None, None) + phys: str = "unset" + group_key: Optional[str] = None + + def __hash__(self): + return hash(self.path) + + +class _Fixtures: + """contains all predefined Fixtures. + Can be extended with new Fixtures during runtime""" + + dev_input_event1 = Fixture( + capabilities={ + evdev.ecodes.EV_KEY: [evdev.ecodes.KEY_A], + }, + phys="usb-0000:03:00.0-0/input1", + info=info_foo, + name="Foo Device", + path="/dev/input/event1", + ) + # Another "Foo Device", which will get an incremented key. + # If possible write tests using this one, because name != key here and + # that would be important to test as well. Otherwise the tests can't + # see if the groups correct attribute is used in functions and paths. + dev_input_event11 = Fixture( + capabilities={ + evdev.ecodes.EV_KEY: [ + evdev.ecodes.BTN_LEFT, + evdev.ecodes.BTN_TOOL_DOUBLETAP, + ], + evdev.ecodes.EV_REL: [ + evdev.ecodes.REL_X, + evdev.ecodes.REL_Y, + evdev.ecodes.REL_WHEEL, + evdev.ecodes.REL_HWHEEL, + ], + }, + phys=f"{phys_foo}/input2", + info=info_foo, + name="Foo Device foo", + group_key="Foo Device 2", # expected key + path="/dev/input/event11", + ) + dev_input_event10 = Fixture( + capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, + phys=f"{phys_foo}/input3", + info=info_foo, + name="Foo Device", + group_key="Foo Device 2", + path="/dev/input/event10", + ) + dev_input_event13 = Fixture( + capabilities={evdev.ecodes.EV_KEY: [], evdev.ecodes.EV_SYN: []}, + phys=f"{phys_foo}/input1", + info=info_foo, + name="Foo Device", + group_key="Foo Device 2", + path="/dev/input/event13", + ) + dev_input_event14 = Fixture( + capabilities={evdev.ecodes.EV_SYN: []}, + phys=f"{phys_foo}/input0", + info=info_foo, + name="Foo Device qux", + group_key="Foo Device 2", + path="/dev/input/event14", + ) + dev_input_event15 = Fixture( + capabilities={ + evdev.ecodes.EV_SYN: [], + evdev.ecodes.EV_ABS: [ + evdev.ecodes.ABS_X, + evdev.ecodes.ABS_Y, + evdev.ecodes.ABS_RX, + evdev.ecodes.ABS_RY, + evdev.ecodes.ABS_Z, + evdev.ecodes.ABS_RZ, + evdev.ecodes.ABS_HAT0X, + evdev.ecodes.ABS_HAT0Y, + ], + evdev.ecodes.EV_KEY: [evdev.ecodes.BTN_A], + }, + phys=f"{phys_foo}/input4", + info=info_foo, + name="Foo Device bar", + group_key="Foo Device 2", + path="/dev/input/event15", + ) + # Bar Device + dev_input_event20 = Fixture( + capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, + phys="usb-0000:03:00.0-2/input1", + info=evdev.device.DeviceInfo(2, 1, 2, 1), + name="Bar Device", + path="/dev/input/event20", + ) + dev_input_event30 = Fixture( + capabilities={ + evdev.ecodes.EV_SYN: [], + evdev.ecodes.EV_ABS: [ + evdev.ecodes.ABS_X, + evdev.ecodes.ABS_Y, + evdev.ecodes.ABS_RX, + evdev.ecodes.ABS_RY, + evdev.ecodes.ABS_Z, + evdev.ecodes.ABS_RZ, + evdev.ecodes.ABS_HAT0X, + evdev.ecodes.ABS_HAT0Y, + ], + evdev.ecodes.EV_KEY: [ + evdev.ecodes.BTN_A, + evdev.ecodes.BTN_B, + evdev.ecodes.BTN_X, + evdev.ecodes.BTN_Y, + ], + }, + phys="", # this is empty sometimes + info=evdev.device.DeviceInfo(3, 1, 3, 1), + name="gamepad", + path="/dev/input/event30", + ) + # device that is completely ignored + dev_input_event31 = Fixture( + capabilities={evdev.ecodes.EV_SYN: []}, + phys="usb-0000:03:00.0-4/input1", + info=evdev.device.DeviceInfo(4, 1, 4, 1), + name="Power Button", + path="/dev/input/event31", + ) + # input-remapper devices are not displayed in the ui, some instance + # of input-remapper started injecting apparently. + dev_input_event40 = Fixture( + capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, + phys="input-remapper/input1", + info=evdev.device.DeviceInfo(5, 1, 5, 1), + name="input-remapper Bar Device", + path="/dev/input/event40", + ) + # denylisted + dev_input_event51 = Fixture( + capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, + phys="usb-0000:03:00.0-5/input1", + info=evdev.device.DeviceInfo(6, 1, 6, 1), + name="YuBiCofooYuBiKeYbar", + path="/dev/input/event51", + ) + # name requires sanitation + dev_input_event52 = Fixture( + capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, + phys="usb-0000:03:00.0-3/input1", + info=evdev.device.DeviceInfo(2, 1, 2, 1), + name="Qux/Device?", + path="/dev/input/event52", + ) + + def __init__(self): + self._iter = [ + self.dev_input_event1, + self.dev_input_event11, + self.dev_input_event10, + self.dev_input_event13, + self.dev_input_event14, + self.dev_input_event15, + self.dev_input_event20, + self.dev_input_event30, + self.dev_input_event31, + self.dev_input_event40, + self.dev_input_event51, + self.dev_input_event52, + ] + self._dynamic_fixtures = {} + + def __getitem__(self, path: str) -> Fixture: + """get a Fixture by it's unique /dev/input/eventX path""" + if fixture := self._dynamic_fixtures.get(path): + return fixture + path = self._path_to_attribute(path) + + try: + return getattr(self, path) + except AttributeError as e: + raise KeyError(str(e)) + + def __setitem__(self, key: str, value: [Fixture | dict]): + if isinstance(value, Fixture): + self._dynamic_fixtures[key] = value + elif isinstance(value, dict): + self._dynamic_fixtures[key] = Fixture(path=key, **value) + + def __iter__(self): + return iter([*self._iter, *self._dynamic_fixtures.values()]) + + def reset(self): + self._dynamic_fixtures = {} + + @staticmethod + def _path_to_attribute(path) -> str: + if path.startswith("/"): + path = path[1:] + if "/" in path: + path = path.replace("/", "_") + return path + + def get(self, item) -> Optional[Fixture]: + try: + return self[item] + except KeyError: + return None + + @property + def foo_device_1_1(self): + return self["/dev/input/event1"] + + @property + def foo_device_2_mouse(self): + return self["/dev/input/event11"] + + @property + def foo_device_2_keyboard(self): + return self["/dev/input/event10"] + + @property + def foo_device_2_13(self): + return self["/dev/input/event13"] + + @property + def foo_device_2_qux(self): + return self["/dev/input/event14"] + + @property + def foo_device_2_gamepad(self): + return self["/dev/input/event15"] + + @property + def bar_device(self): + return self["/dev/input/event20"] + + @property + def gamepad(self): + return self["/dev/input/event30"] + + @property + def power_button(self): + return self["/dev/input/event31"] + + @property + def input_remapper_bar_device(self): + return self["/dev/input/event40"] + + @property + def YuBiCofooYuBiKeYbar(self): + return self["/dev/input/event51"] + + @property + def QuxSlashDeviceQuestionmark(self): + return self["/dev/input/event52"] + + +fixtures = _Fixtures() + + +def get_ui_mapping(combination="99,99,99", target_uinput="keyboard", output_symbol="a"): + """Convenient function to get a valid mapping.""" + from inputremapper.configs.mapping import UIMapping + + return UIMapping( + event_combination=combination, + target_uinput=target_uinput, + output_symbol=output_symbol, + ) + + +def get_key_mapping( + combination="99,99,99", target_uinput="keyboard", output_symbol="a" +): + """Convenient function to get a valid mapping.""" + from inputremapper.configs.mapping import Mapping + + return Mapping( + event_combination=combination, + target_uinput=target_uinput, + output_symbol=output_symbol, + ) + + +def new_event(type, code, value, timestamp=None, offset=0): + """Create a new input_event.""" + from tests.lib.patches import InputEvent + + if timestamp is None: + timestamp = time.time() + offset + + sec = int(timestamp) + usec = timestamp % 1 * 1000000 + event = InputEvent(sec, usec, type, code, value) + return event + + +def prepare_presets(): + """prepare a few presets for use in tests + "Foo Device 2/preset3" is the newest and "Foo Device 2/preset2" is set to autoload + """ + from inputremapper.configs.preset import Preset + from inputremapper.configs.paths import get_config_path, get_preset_path + from inputremapper.configs.global_config import global_config + + preset1 = Preset(get_preset_path("Foo Device", "preset1")) + preset1.add(get_key_mapping(combination="1,1,1", output_symbol="b")) + preset1.add(get_key_mapping(combination="1,2,1")) + preset1.save() + + time.sleep(0.1) + preset2 = Preset(get_preset_path("Foo Device", "preset2")) + preset2.add(get_key_mapping(combination="1,3,1")) + preset2.add(get_key_mapping(combination="1,4,1")) + preset2.save() + + # make sure the timestamp of preset 3 is the newest, + # so that it will be automatically loaded by the GUI + time.sleep(0.1) + preset3 = Preset(get_preset_path("Foo Device", "preset3")) + preset3.add(get_key_mapping(combination="1,5,1")) + preset3.save() + + with open(get_config_path("config.json"), "w") as file: + json.dump({"autoload": {"Foo Device 2": "preset2"}}, file, indent=4) + + global_config.load_config() + + return preset1, preset2, preset3 diff --git a/tests/lib/logger.py b/tests/lib/logger.py new file mode 100644 index 00000000..c462ca1a --- /dev/null +++ b/tests/lib/logger.py @@ -0,0 +1,54 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +from __future__ import annotations + +import sys +import traceback +import tracemalloc +import warnings +import logging + + +tracemalloc.start() + +logger = logging.getLogger("input-remapper-test") +handler = logging.StreamHandler() +handler.setFormatter(logging.Formatter("\033[90mTest: %(message)s\033[0m")) +logger.addHandler(handler) +logger.setLevel(logging.INFO) + + +def update_inputremapper_verbosity(): + from inputremapper.logger import update_verbosity + + update_verbosity(True) + + +def warn_with_traceback(message, category, filename, lineno, file=None, line=None): + log = file if hasattr(file, "write") else sys.stderr + traceback.print_stack(file=log) + log.write(warnings.formatwarning(message, category, filename, lineno, line)) + + +def patch_warnings(): + # show traceback + warnings.showwarning = warn_with_traceback + warnings.simplefilter("always") diff --git a/tests/lib/patches.py b/tests/lib/patches.py new file mode 100644 index 00000000..c00ae7f7 --- /dev/null +++ b/tests/lib/patches.py @@ -0,0 +1,344 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +from __future__ import annotations + +import asyncio +import copy +import os +import subprocess +import time +from pickle import UnpicklingError + +import evdev + +from tests.lib.constants import EVENT_READ_TIMEOUT, MIN_ABS, MAX_ABS +from tests.lib.fixtures import Fixture, fixtures, new_event +from tests.lib.pipes import ( + setup_pipe, + push_events, + uinput_write_history, + uinput_write_history_pipe, + pending_events, +) +from tests.lib.xmodmap import xmodmap +from tests.lib.tmp import tmp +from tests.lib.logger import logger + + +def patch_paths(): + from inputremapper import user + + user.HOME = tmp + + +class InputDevice: + # expose as existing attribute, otherwise the patch for + # evdev < 1.0.0 will crash the test + path = None + + def __init__(self, path): + if path != "justdoit" and not fixtures.get(path): + raise FileNotFoundError() + if path == "justdoit": + self._fixture = Fixture() + else: + self._fixture = fixtures[path] + + self.path = path + self.phys = self._fixture.phys + self.info = self._fixture.info + self.name = self._fixture.name + + # this property exists only for test purposes and is not part of + # the original evdev.InputDevice class + self.group_key = self._fixture.group_key or self._fixture.name + + # ensure a pipe exists to make this object act like + # it is reading events from a device + setup_pipe(self._fixture) + + self.fd = pending_events[self._fixture][1].fileno() + + def push_events(self, events): + push_events(self._fixture, events) + + def fileno(self): + """Compatibility to select.select.""" + return self.fd + + def log(self, key, msg): + logger.info(f'%s "%s" "%s" %s', msg, self.name, self.path, key) + + def absinfo(self, *args): + raise Exception("Ubuntus version of evdev doesn't support .absinfo") + + def grab(self): + logger.info("grab %s %s", self.name, self.path) + + def ungrab(self): + logger.info("ungrab %s %s", self.name, self.path) + + async def async_read_loop(self): + logger.info("starting read loop for %s", self.path) + new_frame = asyncio.Event() + asyncio.get_running_loop().add_reader(self.fd, new_frame.set) + while True: + await new_frame.wait() + new_frame.clear() + if not pending_events[self._fixture][1].poll(): + # todo: why? why do we need this? + # sometimes this happens, as if a other process calls recv on + # the pipe + continue + + event = pending_events[self._fixture][1].recv() + logger.info("got %s at %s", event, self.path) + yield event + + def read(self): + # the patched fake InputDevice objects read anything pending from + # that group. + # To be realistic it would have to check if the provided + # element is in its capabilities. + if self.group_key not in pending_events: + self.log("no events to read", self.group_key) + return + + # consume all of them + while pending_events[self._fixture][1].poll(): + event = pending_events[self._fixture][1].recv() + self.log(event, "read") + yield event + time.sleep(EVENT_READ_TIMEOUT) + + def read_loop(self): + """Endless loop that yields events.""" + while True: + event = pending_events[self._fixture][1].recv() + if event is not None: + self.log(event, "read_loop") + yield event + time.sleep(EVENT_READ_TIMEOUT) + + def read_one(self): + """Read one event or none if nothing available.""" + if not pending_events.get(self._fixture): + return None + + if not pending_events[self._fixture][1].poll(): + return None + + try: + event = pending_events[self._fixture][1].recv() + except (UnpicklingError, EOFError): + # failed in tests sometimes + return None + + self.log(event, "read_one") + return event + + def capabilities(self, absinfo=True, verbose=False): + result = copy.deepcopy(self._fixture.capabilities) + + if absinfo and evdev.ecodes.EV_ABS in result: + absinfo_obj = evdev.AbsInfo( + value=None, + min=MIN_ABS, + fuzz=None, + flat=None, + resolution=None, + max=MAX_ABS, + ) + + ev_abs = [] + for ev_code in result[evdev.ecodes.EV_ABS]: + if ev_code in range(0x10, 0x18): # ABS_HAT0X - ABS_HAT3Y + absinfo_obj = evdev.AbsInfo( + value=None, + min=-1, + fuzz=None, + flat=None, + resolution=None, + max=1, + ) + ev_abs.append((ev_code, absinfo_obj)) + + result[evdev.ecodes.EV_ABS] = ev_abs + + return result + + def input_props(self): + return [] + + +uinputs = {} + + +class UInput: + def __init__(self, events=None, name="unnamed", *args, **kwargs): + self.fd = 0 + self.write_count = 0 + self.device = InputDevice("justdoit") + self.name = name + self.events = events + self.write_history = [] + + global uinputs + uinputs[name] = self + + def capabilities(self, verbose=False, absinfo=True): + if absinfo or 3 not in self.events: + return self.events + else: + events = self.events.copy() + events[3] = [code for code, _ in self.events[3]] + return events + + def write(self, type, code, value): + self.write_count += 1 + event = new_event(type, code, value) + uinput_write_history.append(event) + uinput_write_history_pipe[1].send(event) + self.write_history.append(event) + logger.info("%s written", (type, code, value)) + + def syn(self): + pass + + +# TODO inherit from input-remappers InputEvent? +# makes convert_to_internal_events obsolete +class InputEvent(evdev.InputEvent): + def __init__(self, sec, usec, type, code, value): + self.t = (type, code, value) + super().__init__(sec, usec, type, code, value) + + def copy(self): + return InputEvent(self.sec, self.usec, self.type, self.code, self.value) + + +def patch_evdev(): + def list_devices(): + return [fixture_.path for fixture_ in fixtures] + + evdev.list_devices = list_devices + evdev.InputDevice = InputDevice + evdev.UInput = UInput + evdev.InputEvent = InputEvent + + +def patch_events(): + # improve logging of stuff + evdev.InputEvent.__str__ = lambda self: ( + f"InputEvent{(self.type, self.code, self.value)}" + ) + + +def patch_os_system(): + """Avoid running pkexec.""" + original_system = os.system + + def system(command): + if "pkexec" in command: + # because it + # - will open a window for user input + # - has no knowledge of the fixtures and patches + raise Exception("Write patches to avoid running pkexec stuff") + return original_system(command) + + os.system = system + + +def patch_check_output(): + """Xmodmap -pke should always return a fixed set of symbols. + + On some installations the `xmodmap` command might be missig completely, + which would break the tests. + """ + original_check_output = subprocess.check_output + + def check_output(command, *args, **kwargs): + if "xmodmap" in command and "-pke" in command: + return xmodmap + return original_check_output(command, *args, **kwargs) + + subprocess.check_output = check_output + + +def patch_regrab_timeout(): + # no need for a high number in tests + from inputremapper.injection.injector import Injector + + Injector.regrab_timeout = 0.05 + + +def is_running_patch(): + logger.info("is_running is patched to always return True") + return True + + +def patch_is_running(): + from inputremapper.gui.reader_service import ReaderService + + setattr(ReaderService, "is_running", is_running_patch) + + +class FakeDaemonProxy: + def __init__(self): + self.calls = { + "stop_injecting": [], + "get_state": [], + "start_injecting": [], + "stop_all": 0, + "set_config_dir": [], + "autoload": 0, + "autoload_single": [], + "hello": [], + } + + def stop_injecting(self, group_key: str) -> None: + self.calls["stop_injecting"].append(group_key) + + def get_state(self, group_key: str): + from inputremapper.injection.injector import InjectorState + + self.calls["get_state"].append(group_key) + return InjectorState.STOPPED + + def start_injecting(self, group_key: str, preset: str) -> bool: + self.calls["start_injecting"].append((group_key, preset)) + return True + + def stop_all(self) -> None: + self.calls["stop_all"] += 1 + + def set_config_dir(self, config_dir: str) -> None: + self.calls["set_config_dir"].append(config_dir) + + def autoload(self) -> None: + self.calls["autoload"] += 1 + + def autoload_single(self, group_key: str) -> None: + self.calls["autoload_single"].append(group_key) + + def hello(self, out: str) -> str: + self.calls["hello"].append(out) + return out diff --git a/tests/lib/pipes.py b/tests/lib/pipes.py new file mode 100644 index 00000000..7e9bb24b --- /dev/null +++ b/tests/lib/pipes.py @@ -0,0 +1,85 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +from __future__ import annotations + +import multiprocessing +from multiprocessing.connection import Connection +from typing import Dict, Tuple + +from tests.lib.fixtures import Fixture +from tests.lib.logger import logger + +uinput_write_history = [] +# for tests that makes the injector create its processes +uinput_write_history_pipe = multiprocessing.Pipe() +pending_events: Dict[Fixture, Tuple[Connection, Connection]] = {} + + +def read_write_history_pipe(): + """Convert the write history from the pipe to some easier to manage list.""" + history = [] + while uinput_write_history_pipe[0].poll(): + event = uinput_write_history_pipe[0].recv() + history.append((event.type, event.code, event.value)) + return history + + +def setup_pipe(fixture: Fixture): + """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() + + +def get_events(): + """Get all events written by the injector.""" + return uinput_write_history + + +def push_event(fixture: Fixture, event, force: bool = False): + """Make a device act like it is reading events from evdev. + + push_event is like hitting a key on a keyboard for stuff that reads from + evdev.InputDevice (which is patched in test.py to work that way) + + Parameters + ---------- + fixture + For example 'Foo Device' + event + force + don't check if the event is in fixture.capabilities + """ + setup_pipe(fixture) + if not force and ( + not fixture.capabilities.get(event.type) + or event.code not in fixture.capabilities[event.type] + ): + raise AssertionError(f"Fixture {fixture.path} cannot send {event}") + logger.info("Simulating %s for %s", event, fixture.path) + pending_events[fixture][0].send(event) + + +def push_events(fixture: Fixture, events, force=False): + """Push multiple events.""" + for event in events: + push_event(fixture, event, force) diff --git a/tests/lib/stuff.py b/tests/lib/stuff.py new file mode 100644 index 00000000..ccc07cc6 --- /dev/null +++ b/tests/lib/stuff.py @@ -0,0 +1,38 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +import os +import copy +from unittest.mock import patch + + +def convert_to_internal_events(events): + """Convert an iterable of InputEvent to a list of inputremapper.InputEvent.""" + from inputremapper.input_event import InputEvent as InternalInputEvent + + return [InternalInputEvent.from_event(event) for event in events] + + +def spy(obj, name): + """Convenient wrapper for patch.object(..., ..., wraps=...).""" + return patch.object(obj, name, wraps=obj.__getattribute__(name)) + + +environ_copy = copy.deepcopy(os.environ) diff --git a/tests/lib/tmp.py b/tests/lib/tmp.py new file mode 100644 index 00000000..967265e8 --- /dev/null +++ b/tests/lib/tmp.py @@ -0,0 +1,28 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + +from __future__ import annotations + +import tempfile + +# When it gets garbage collected it cleans up the temporary directory so it needs to +# stay reachable while the tests are ran. +temporary_directory = tempfile.TemporaryDirectory(prefix="input-remapper-test") +tmp = temporary_directory.name diff --git a/tests/xmodmap.py b/tests/lib/xmodmap.py similarity index 93% rename from tests/xmodmap.py rename to tests/lib/xmodmap.py index 2f3b9a5e..d13e09cd 100644 --- a/tests/xmodmap.py +++ b/tests/lib/xmodmap.py @@ -1,3 +1,23 @@ +#!/usr/bin/python3 +# -*- coding: utf-8 -*- +# input-remapper - GUI for device specific keyboard mappings +# Copyright (C) 2022 sezanzeb +# +# This file is part of input-remapper. +# +# input-remapper is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# input-remapper is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with input-remapper. If not, see . + xmodmap = ( b"keycode 8 =\nkeycode 9 = Escape NoSymbol Escape\nkeycode 10 = 1 exclam 1 exclam onesuperior exclamdown ones" b"uperior\nkeycode 11 = 2 quotedbl 2 quotedbl twosuperior oneeighth twosuperior\nkeycode 12 = 3 section 3 sectio" diff --git a/tests/test.py b/tests/test.py index a1712867..084397d2 100644 --- a/tests/test.py +++ b/tests/test.py @@ -18,24 +18,11 @@ # You should have received a copy of the GNU General Public License # along with input-remapper. If not, see . - -"""Sets up input-remapper for the tests and runs them. - -This module needs to be imported first in test files. -""" - from __future__ import annotations import argparse -import dataclasses -import json import os import sys -import tempfile -import traceback -import warnings -from multiprocessing.connection import Connection -from typing import Dict, Tuple, Optional import tracemalloc tracemalloc.start() @@ -84,29 +71,24 @@ if __name__ == "__main__": tests.test.main() -import shutil -import time -import copy import unittest import subprocess -import multiprocessing -import asyncio -import psutil -import logging -from pickle import UnpicklingError -from unittest.mock import patch - -import evdev - -from tests.xmodmap import xmodmap os.environ["UNITTEST"] = "1" -logger = logging.getLogger("input-remapper-test") -handler = logging.StreamHandler() -handler.setFormatter(logging.Formatter("\033[90mTest: %(message)s\033[0m")) -logger.addHandler(handler) -logger.setLevel(logging.INFO) +from tests.lib.fixtures import fixtures +from tests.lib.pipes import setup_pipe +from tests.lib.patches import ( + patch_paths, + patch_events, + patch_os_system, + patch_check_output, + patch_regrab_timeout, + patch_is_running, + patch_evdev, +) +from tests.lib.cleanup import cleanup +from tests.lib.logger import update_inputremapper_verbosity def is_service_running(): @@ -118,905 +100,36 @@ def is_service_running(): return False -def join_children(): - """Wait for child processes to exit. Stop them if it takes too long.""" - this = psutil.Process(os.getpid()) - - i = 0 - time.sleep(EVENT_READ_TIMEOUT) - children = this.children(recursive=True) - while len([c for c in children if c.status() != "zombie"]) > 0: - for child in children: - if i > 10: - child.kill() - logger.info("Killed pid %s because it didn't finish in time", child.pid) - - children = this.children(recursive=True) - time.sleep(EVENT_READ_TIMEOUT) - i += 1 - - if is_service_running(): # let tests control daemon existance raise Exception("Expected the service not to be running already.") -# give tests some time to test stuff while the process -# is still running -EVENT_READ_TIMEOUT = 0.01 - -# based on experience how much time passes at most until -# the reader-service starts receiving previously pushed events after a -# call to start_reading -START_READING_DELAY = 0.05 - -# for joysticks -MIN_ABS = -(2**15) -MAX_ABS = 2**15 - -# When it gets garbage collected it cleans up the temporary directory so it needs to -# stay reachable while the tests are ran. -temporary_directory = tempfile.TemporaryDirectory(prefix="input-remapper-test") -tmp = temporary_directory.name - -uinput_write_history = [] -# for tests that makes the injector create its processes -uinput_write_history_pipe = multiprocessing.Pipe() -pending_events: Dict[Fixture, Tuple[Connection, Connection]] = {} - - -def read_write_history_pipe(): - """Convert the write history from the pipe to some easier to manage list.""" - history = [] - while uinput_write_history_pipe[0].poll(): - event = uinput_write_history_pipe[0].recv() - history.append((event.type, event.code, event.value)) - return history - - -# input-remapper is only interested in devices that have EV_KEY, add some -# random other stuff to test that they are ignored. -phys_foo = "usb-0000:03:00.0-1/input2" -info_foo = evdev.device.DeviceInfo(1, 1, 1, 1) - -keyboard_keys = sorted(evdev.ecodes.keys.keys())[:255] - - -@dataclasses.dataclass(frozen=True) -class Fixture: - capabilities: Dict = dataclasses.field(default_factory=dict) - path: str = "" - name: str = "unset" - info: evdev.device.DeviceInfo = evdev.device.DeviceInfo(None, None, None, None) - phys: str = "unset" - group_key: Optional[str] = None - - def __hash__(self): - return hash(self.path) - - -class _Fixtures: - """contains all predefined Fixtures. - Can be extended with new Fixtures during runtime""" - - dev_input_event1 = Fixture( - capabilities={ - evdev.ecodes.EV_KEY: [evdev.ecodes.KEY_A], - }, - phys="usb-0000:03:00.0-0/input1", - info=info_foo, - name="Foo Device", - path="/dev/input/event1", - ) - # Another "Foo Device", which will get an incremented key. - # If possible write tests using this one, because name != key here and - # that would be important to test as well. Otherwise the tests can't - # see if the groups correct attribute is used in functions and paths. - dev_input_event11 = Fixture( - capabilities={ - evdev.ecodes.EV_KEY: [ - evdev.ecodes.BTN_LEFT, - evdev.ecodes.BTN_TOOL_DOUBLETAP, - ], - evdev.ecodes.EV_REL: [ - evdev.ecodes.REL_X, - evdev.ecodes.REL_Y, - evdev.ecodes.REL_WHEEL, - evdev.ecodes.REL_HWHEEL, - ], - }, - phys=f"{phys_foo}/input2", - info=info_foo, - name="Foo Device foo", - group_key="Foo Device 2", # expected key - path="/dev/input/event11", - ) - dev_input_event10 = Fixture( - capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, - phys=f"{phys_foo}/input3", - info=info_foo, - name="Foo Device", - group_key="Foo Device 2", - path="/dev/input/event10", - ) - dev_input_event13 = Fixture( - capabilities={evdev.ecodes.EV_KEY: [], evdev.ecodes.EV_SYN: []}, - phys=f"{phys_foo}/input1", - info=info_foo, - name="Foo Device", - group_key="Foo Device 2", - path="/dev/input/event13", - ) - dev_input_event14 = Fixture( - capabilities={evdev.ecodes.EV_SYN: []}, - phys=f"{phys_foo}/input0", - info=info_foo, - name="Foo Device qux", - group_key="Foo Device 2", - path="/dev/input/event14", - ) - dev_input_event15 = Fixture( - capabilities={ - evdev.ecodes.EV_SYN: [], - evdev.ecodes.EV_ABS: [ - evdev.ecodes.ABS_X, - evdev.ecodes.ABS_Y, - evdev.ecodes.ABS_RX, - evdev.ecodes.ABS_RY, - evdev.ecodes.ABS_Z, - evdev.ecodes.ABS_RZ, - evdev.ecodes.ABS_HAT0X, - evdev.ecodes.ABS_HAT0Y, - ], - evdev.ecodes.EV_KEY: [evdev.ecodes.BTN_A], - }, - phys=f"{phys_foo}/input4", - info=info_foo, - name="Foo Device bar", - group_key="Foo Device 2", - path="/dev/input/event15", - ) - # Bar Device - dev_input_event20 = Fixture( - capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, - phys="usb-0000:03:00.0-2/input1", - info=evdev.device.DeviceInfo(2, 1, 2, 1), - name="Bar Device", - path="/dev/input/event20", - ) - dev_input_event30 = Fixture( - capabilities={ - evdev.ecodes.EV_SYN: [], - evdev.ecodes.EV_ABS: [ - evdev.ecodes.ABS_X, - evdev.ecodes.ABS_Y, - evdev.ecodes.ABS_RX, - evdev.ecodes.ABS_RY, - evdev.ecodes.ABS_Z, - evdev.ecodes.ABS_RZ, - evdev.ecodes.ABS_HAT0X, - evdev.ecodes.ABS_HAT0Y, - ], - evdev.ecodes.EV_KEY: [ - evdev.ecodes.BTN_A, - evdev.ecodes.BTN_B, - evdev.ecodes.BTN_X, - evdev.ecodes.BTN_Y, - ], - }, - phys="", # this is empty sometimes - info=evdev.device.DeviceInfo(3, 1, 3, 1), - name="gamepad", - path="/dev/input/event30", - ) - # device that is completely ignored - dev_input_event31 = Fixture( - capabilities={evdev.ecodes.EV_SYN: []}, - phys="usb-0000:03:00.0-4/input1", - info=evdev.device.DeviceInfo(4, 1, 4, 1), - name="Power Button", - path="/dev/input/event31", - ) - # input-remapper devices are not displayed in the ui, some instance - # of input-remapper started injecting apparently. - dev_input_event40 = Fixture( - capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, - phys="input-remapper/input1", - info=evdev.device.DeviceInfo(5, 1, 5, 1), - name="input-remapper Bar Device", - path="/dev/input/event40", - ) - # denylisted - dev_input_event51 = Fixture( - capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, - phys="usb-0000:03:00.0-5/input1", - info=evdev.device.DeviceInfo(6, 1, 6, 1), - name="YuBiCofooYuBiKeYbar", - path="/dev/input/event51", - ) - # name requires sanitation - dev_input_event52 = Fixture( - capabilities={evdev.ecodes.EV_KEY: keyboard_keys}, - phys="usb-0000:03:00.0-3/input1", - info=evdev.device.DeviceInfo(2, 1, 2, 1), - name="Qux/Device?", - path="/dev/input/event52", - ) - - def __init__(self): - self._iter = [ - self.dev_input_event1, - self.dev_input_event11, - self.dev_input_event10, - self.dev_input_event13, - self.dev_input_event14, - self.dev_input_event15, - self.dev_input_event20, - self.dev_input_event30, - self.dev_input_event31, - self.dev_input_event40, - self.dev_input_event51, - self.dev_input_event52, - ] - self._dynamic_fixtures = {} - - def __getitem__(self, path: str) -> Fixture: - """get a Fixture by it's unique /dev/input/eventX path""" - if fixture := self._dynamic_fixtures.get(path): - return fixture - path = self._path_to_attribute(path) - - try: - return getattr(self, path) - except AttributeError as e: - raise KeyError(str(e)) - - def __setitem__(self, key: str, value: [Fixture | dict]): - if isinstance(value, Fixture): - self._dynamic_fixtures[key] = value - elif isinstance(value, dict): - self._dynamic_fixtures[key] = Fixture(path=key, **value) - - def __iter__(self): - return iter([*self._iter, *self._dynamic_fixtures.values()]) - - def reset(self): - self._dynamic_fixtures = {} - - @staticmethod - def _path_to_attribute(path) -> str: - if path.startswith("/"): - path = path[1:] - if "/" in path: - path = path.replace("/", "_") - return path - - def get(self, item) -> Optional[Fixture]: - try: - return self[item] - except KeyError: - return None - - @property - def foo_device_1_1(self): - return self["/dev/input/event1"] - - @property - def foo_device_2_mouse(self): - return self["/dev/input/event11"] - - @property - def foo_device_2_keyboard(self): - return self["/dev/input/event10"] - - @property - def foo_device_2_13(self): - return self["/dev/input/event13"] - - @property - def foo_device_2_qux(self): - return self["/dev/input/event14"] - - @property - def foo_device_2_gamepad(self): - return self["/dev/input/event15"] - - @property - def bar_device(self): - return self["/dev/input/event20"] - - @property - def gamepad(self): - return self["/dev/input/event30"] - - @property - def power_button(self): - return self["/dev/input/event31"] - - @property - def input_remapper_bar_device(self): - return self["/dev/input/event40"] - - @property - def YuBiCofooYuBiKeYbar(self): - return self["/dev/input/event51"] - - @property - def QuxSlashDeviceQuestionmark(self): - return self["/dev/input/event52"] - - -fixtures = _Fixtures() - - -def setup_pipe(fixture: Fixture): - """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 reader-service) gets forked, # so that events can be pushed after the fork. for _fixture in fixtures: setup_pipe(_fixture) -def get_events(): - """Get all events written by the injector.""" - return uinput_write_history - - -def push_event(fixture: Fixture, event: InputEvent, force: bool = False): - """Make a device act like it is reading events from evdev. - - push_event is like hitting a key on a keyboard for stuff that reads from - evdev.InputDevice (which is patched in test.py to work that way) - - Parameters - ---------- - fixture - For example 'Foo Device' - event - force - don't check if the event is in fixture.capabilities - """ - setup_pipe(fixture) - if not force and ( - not fixture.capabilities.get(event.type) - or event.code not in fixture.capabilities[event.type] - ): - raise AssertionError(f"Fixture {fixture.path} cannot send {event}") - logger.info("Simulating %s for %s", event, fixture.path) - pending_events[fixture][0].send(event) - - -def push_events(fixture: Fixture, events, force=False): - """Push multiple events.""" - for event in events: - push_event(fixture, event, force) - - -def new_event(type, code, value, timestamp=None, offset=0): - """Create a new input_event.""" - if timestamp is None: - timestamp = time.time() + offset - - sec = int(timestamp) - usec = timestamp % 1 * 1000000 - event = InputEvent(sec, usec, type, code, value) - return event - - -def patch_paths(): - from inputremapper import user - - user.HOME = tmp - - -class InputDevice: - # expose as existing attribute, otherwise the patch for - # evdev < 1.0.0 will crash the test - path = None - - def __init__(self, path): - if path != "justdoit" and not fixtures.get(path): - raise FileNotFoundError() - if path == "justdoit": - self._fixture = Fixture() - else: - self._fixture = fixtures[path] - - self.path = path - self.phys = self._fixture.phys - self.info = self._fixture.info - self.name = self._fixture.name - - # this property exists only for test purposes and is not part of - # the original evdev.InputDevice class - self.group_key = self._fixture.group_key or self._fixture.name - - # ensure a pipe exists to make this object act like - # it is reading events from a device - setup_pipe(self._fixture) - - self.fd = pending_events[self._fixture][1].fileno() - - def push_events(self, events): - push_events(self._fixture, events) - - def fileno(self): - """Compatibility to select.select.""" - return self.fd - - def log(self, key, msg): - logger.info(f'%s "%s" "%s" %s', msg, self.name, self.path, key) - - def absinfo(self, *args): - raise Exception("Ubuntus version of evdev doesn't support .absinfo") - - def grab(self): - logger.info("grab %s %s", self.name, self.path) - - def ungrab(self): - logger.info("ungrab %s %s", self.name, self.path) - - async def async_read_loop(self): - logger.info("starting read loop for %s", self.path) - new_frame = asyncio.Event() - asyncio.get_running_loop().add_reader(self.fd, new_frame.set) - while True: - await new_frame.wait() - new_frame.clear() - if not pending_events[self._fixture][1].poll(): - # todo: why? why do we need this? - # sometimes this happens, as if a other process calls recv on - # the pipe - continue - - event = pending_events[self._fixture][1].recv() - logger.info("got %s at %s", event, self.path) - yield event - - def read(self): - # the patched fake InputDevice objects read anything pending from - # that group. - # To be realistic it would have to check if the provided - # element is in its capabilities. - if self.group_key not in pending_events: - self.log("no events to read", self.group_key) - return - - # consume all of them - while pending_events[self._fixture][1].poll(): - event = pending_events[self._fixture][1].recv() - self.log(event, "read") - yield event - time.sleep(EVENT_READ_TIMEOUT) - - def read_loop(self): - """Endless loop that yields events.""" - while True: - event = pending_events[self._fixture][1].recv() - if event is not None: - self.log(event, "read_loop") - yield event - time.sleep(EVENT_READ_TIMEOUT) - - def read_one(self): - """Read one event or none if nothing available.""" - if not pending_events.get(self._fixture): - return None - - if not pending_events[self._fixture][1].poll(): - return None - - try: - event = pending_events[self._fixture][1].recv() - except (UnpicklingError, EOFError): - # failed in tests sometimes - return None - - self.log(event, "read_one") - return event - - def capabilities(self, absinfo=True, verbose=False): - result = copy.deepcopy(self._fixture.capabilities) - - if absinfo and evdev.ecodes.EV_ABS in result: - absinfo_obj = evdev.AbsInfo( - value=None, - min=MIN_ABS, - fuzz=None, - flat=None, - resolution=None, - max=MAX_ABS, - ) - - ev_abs = [] - for ev_code in result[evdev.ecodes.EV_ABS]: - if ev_code in range(0x10, 0x18): # ABS_HAT0X - ABS_HAT3Y - absinfo_obj = evdev.AbsInfo( - value=None, - min=-1, - fuzz=None, - flat=None, - resolution=None, - max=1, - ) - ev_abs.append((ev_code, absinfo_obj)) - - result[evdev.ecodes.EV_ABS] = ev_abs - - return result - - def input_props(self): - return [] - - -uinputs = {} - - -class UInput: - def __init__(self, events=None, name="unnamed", *args, **kwargs): - self.fd = 0 - self.write_count = 0 - self.device = InputDevice("justdoit") - self.name = name - self.events = events - self.write_history = [] - - global uinputs - uinputs[name] = self - - def capabilities(self, verbose=False, absinfo=True): - if absinfo or 3 not in self.events: - return self.events - else: - events = self.events.copy() - events[3] = [code for code, _ in self.events[3]] - return events - - def write(self, type, code, value): - self.write_count += 1 - event = new_event(type, code, value) - uinput_write_history.append(event) - uinput_write_history_pipe[1].send(event) - self.write_history.append(event) - logger.info("%s written", (type, code, value)) - - def syn(self): - pass - - -# TODO inherit from input-remappers InputEvent? -# makes convert_to_internal_events obsolete -class InputEvent(evdev.InputEvent): - def __init__(self, sec, usec, type, code, value): - self.t = (type, code, value) - super().__init__(sec, usec, type, code, value) - - def copy(self): - return InputEvent(self.sec, self.usec, self.type, self.code, self.value) - - -def patch_evdev(): - def list_devices(): - return [fixture_.path for fixture_ in fixtures] - - evdev.list_devices = list_devices - evdev.InputDevice = InputDevice - evdev.UInput = UInput - evdev.InputEvent = InputEvent - - -def patch_events(): - # improve logging of stuff - evdev.InputEvent.__str__ = lambda self: ( - f"InputEvent{(self.type, self.code, self.value)}" - ) - - -def patch_os_system(): - """Avoid running pkexec.""" - original_system = os.system - - def system(command): - if "pkexec" in command: - # because it - # - will open a window for user input - # - has no knowledge of the fixtures and patches - raise Exception("Write patches to avoid running pkexec stuff") - return original_system(command) - - os.system = system - - -def patch_check_output(): - """Xmodmap -pke should always return a fixed set of symbols. - - On some installations the `xmodmap` command might be missig completely, - which would break the tests. - """ - original_check_output = subprocess.check_output - - def check_output(command, *args, **kwargs): - if "xmodmap" in command and "-pke" in command: - return xmodmap - return original_check_output(command, *args, **kwargs) - - subprocess.check_output = check_output - - -def clear_write_history(): - """Empty the history in preparation for the next test.""" - while len(uinput_write_history) > 0: - uinput_write_history.pop() - while uinput_write_history_pipe[0].poll(): - uinput_write_history_pipe[0].recv() - - -def warn_with_traceback(message, category, filename, lineno, file=None, line=None): - - log = file if hasattr(file, "write") else sys.stderr - traceback.print_stack(file=log) - log.write(warnings.formatwarning(message, category, filename, lineno, line)) - - -def patch_warnings(): - # show traceback - warnings.showwarning = warn_with_traceback - warnings.simplefilter("always") - - -# quickly fake some stuff before any other file gets a chance to import -# the original versions +# applying patches before importing input-remappers modules is important, otherwise +# input-remapper might use non-patched modules. Importing modules from inputremapper +# just-in-time in the test-setup functions instead of globally helps. This way, +# it is ensured that the patches on evdev and such are already applied, without having +# to take care about ordering the files in a special way. patch_paths() patch_evdev() patch_events() patch_os_system() patch_check_output() +patch_regrab_timeout() +patch_is_running() # patch_warnings() -from inputremapper.logger import update_verbosity - -update_verbosity(True) - -from inputremapper.input_event import InputEvent as InternalInputEvent -from inputremapper.injection.injector import Injector, InjectorState -from inputremapper.injection.macros.macro import macro_variables -from inputremapper.configs.global_config import global_config -from inputremapper.configs.mapping import Mapping, UIMapping -from inputremapper.groups import groups -from inputremapper.configs.system_mapping import system_mapping -from inputremapper.gui.reader_service import ReaderService -from inputremapper.gui.utils import debounce_manager -from inputremapper.configs.paths import get_config_path, get_preset_path -from inputremapper.configs.preset import Preset - -from inputremapper.injection.global_uinputs import global_uinputs - -# no need for a high number in tests -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] - - -def get_key_mapping( - combination="99,99,99", target_uinput="keyboard", output_symbol="a" -) -> Mapping: - """Convenient function to get a valid mapping.""" - return Mapping( - event_combination=combination, - target_uinput=target_uinput, - output_symbol=output_symbol, - ) - - -def get_ui_mapping( - combination="99,99,99", target_uinput="keyboard", output_symbol="a" -) -> UIMapping: - """Convenient function to get a valid mapping.""" - return UIMapping( - event_combination=combination, - target_uinput=target_uinput, - output_symbol=output_symbol, - ) - - -def quick_cleanup(log=True): - """Reset the applications state.""" - if log: - print("Quick cleanup...") - - debounce_manager.stop_all() - - for device in list(pending_events.keys()): - try: - while pending_events[device][1].poll(): - pending_events[device][1].recv() - except (UnpicklingError, EOFError): - pass - - # setup new pipes for the next test - pending_events[device][1].close() - pending_events[device][0].close() - del pending_events[device] - setup_pipe(device) - - try: - if asyncio.get_event_loop().is_running(): - for task in asyncio.all_tasks(): - task.cancel() - except RuntimeError: - # happens when the event loop disappears for magical reasons - # create a fresh event loop - asyncio.set_event_loop(asyncio.new_event_loop()) - - if macro_variables.process is not None and not macro_variables.process.is_alive(): - # nothing should stop the process during runtime, if it has been started by - # the injector once - raise AssertionError("the SharedDict manager is not running anymore") - - if macro_variables.process is not None: - macro_variables._stop() - - join_children() - - macro_variables.start() - - if os.path.exists(tmp): - shutil.rmtree(tmp) - - global_config.path = os.path.join(get_config_path(), "config.json") - global_config.clear_config() - global_config._save_config() - - system_mapping.populate() - - clear_write_history() - - for name in list(uinputs.keys()): - del uinputs[name] - - # for device in list(active_macros.keys()): - # del active_macros[device] - # for device in list(unreleased.keys()): - # del unreleased[device] - fixtures.reset() - os.environ.update(environ_copy) - for device in list(os.environ.keys()): - if device not in environ_copy: - del os.environ[device] - - for _, pipe in pending_events.values(): - assert not pipe.poll() - - assert macro_variables.is_alive(1) - for uinput in global_uinputs.devices.values(): - uinput.write_count = 0 - uinput.write_history = [] - - global_uinputs.is_service = True - - if log: - print("Quick cleanup done") - - -def cleanup(): - """Reset the applications state. - - Using this is slower, usually quick_cleanup() is sufficient. - """ - print("Cleanup...") - - os.system("pkill -f input-remapper-service") - os.system("pkill -f input-remapper-control") - time.sleep(0.05) - - quick_cleanup(log=False) - groups.refresh() - with patch.object(sys, "argv", ["input-remapper-service"]): - global_uinputs.prepare_all() - - print("Cleanup done") - - -def spy(obj, name): - """Convenient wrapper for patch.object(..., ..., wraps=...).""" - return patch.object(obj, name, wraps=obj.__getattribute__(name)) - - -class FakeDaemonProxy: - def __init__(self): - self.calls = { - "stop_injecting": [], - "get_state": [], - "start_injecting": [], - "stop_all": 0, - "set_config_dir": [], - "autoload": 0, - "autoload_single": [], - "hello": [], - } - - def stop_injecting(self, group_key: str) -> None: - self.calls["stop_injecting"].append(group_key) - - def get_state(self, group_key: str) -> InjectorState: - self.calls["get_state"].append(group_key) - return InjectorState.STOPPED - - def start_injecting(self, group_key: str, preset: str) -> bool: - self.calls["start_injecting"].append((group_key, preset)) - return True - - def stop_all(self) -> None: - self.calls["stop_all"] += 1 - - def set_config_dir(self, config_dir: str) -> None: - self.calls["set_config_dir"].append(config_dir) - - def autoload(self) -> None: - self.calls["autoload"] += 1 - - def autoload_single(self, group_key: str) -> None: - self.calls["autoload_single"].append(group_key) - - def hello(self, out: str) -> str: - self.calls["hello"].append(out) - return out - - -def prepare_presets(): - """prepare a few presets for use in tests - "Foo Device 2/preset3" is the newest and "Foo Device 2/preset2" is set to autoload - """ - preset1 = Preset(get_preset_path("Foo Device", "preset1")) - preset1.add(get_key_mapping(combination="1,1,1", output_symbol="b")) - preset1.add(get_key_mapping(combination="1,2,1")) - preset1.save() - - time.sleep(0.1) - preset2 = Preset(get_preset_path("Foo Device", "preset2")) - preset2.add(get_key_mapping(combination="1,3,1")) - preset2.add(get_key_mapping(combination="1,4,1")) - preset2.save() - - # make sure the timestamp of preset 3 is the newest, - # so that it will be automatically loaded by the GUI - time.sleep(0.1) - preset3 = Preset(get_preset_path("Foo Device", "preset3")) - preset3.add(get_key_mapping(combination="1,5,1")) - preset3.save() - - with open(get_config_path("config.json"), "w") as file: - json.dump({"autoload": {"Foo Device 2": "preset2"}}, file, indent=4) - - global_config.load_config() - - return preset1, preset2, preset3 - - -cleanup() - def main(): + update_inputremapper_verbosity() + + cleanup() # https://docs.python.org/3/library/argparse.html parser = argparse.ArgumentParser(description=__doc__) # repeated argument 0 or more times with modules diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index d7de9e33..79567c51 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -19,7 +19,8 @@ # along with input-remapper. If not, see . -from tests.test import quick_cleanup, tmp +from tests.lib.cleanup import quick_cleanup +from tests.lib.tmp import tmp import os import unittest diff --git a/tests/unit/test_context.py b/tests/unit/test_context.py index c64d6f25..094a69d3 100644 --- a/tests/unit/test_context.py +++ b/tests/unit/test_context.py @@ -17,8 +17,9 @@ # # You should have received a copy of the GNU General Public License # along with input-remapper. If not, see . -from inputremapper.injection.mapping_handlers.hierarchy_handler import HierarchyHandler -from tests.test import quick_cleanup, get_key_mapping + +from tests.lib.cleanup import quick_cleanup +from tests.lib.fixtures import get_key_mapping from evdev.ecodes import ( EV_REL, EV_ABS, diff --git a/tests/unit/test_control.py b/tests/unit/test_control.py index 84866ea6..dbac0ba5 100644 --- a/tests/unit/test_control.py +++ b/tests/unit/test_control.py @@ -22,7 +22,8 @@ """Testing the input-remapper-control command""" -from tests.test import quick_cleanup, tmp +from tests.lib.cleanup import quick_cleanup +from tests.lib.tmp import tmp import os import time diff --git a/tests/unit/test_controller.py b/tests/unit/test_controller.py index 9ecad84e..5d6b3c22 100644 --- a/tests/unit/test_controller.py +++ b/tests/unit/test_controller.py @@ -51,13 +51,10 @@ from inputremapper.gui.utils import CTX_ERROR, CTX_APPLY, gtk_iteration from inputremapper.gui.gettext import _ from inputremapper.injection.global_uinputs import GlobalUInputs from inputremapper.configs.mapping import UIMapping, MappingData -from tests.test import ( - quick_cleanup, - FakeDaemonProxy, - fixtures, - prepare_presets, - spy, -) +from tests.lib.cleanup import quick_cleanup +from tests.lib.stuff import spy +from tests.lib.patches import FakeDaemonProxy +from tests.lib.fixtures import fixtures, prepare_presets from inputremapper.configs.global_config import GlobalConfig from inputremapper.gui.controller import Controller, MAPPING_DEFAULTS from inputremapper.gui.data_manager import DataManager, DEFAULT_PRESET_NAME diff --git a/tests/unit/test_daemon.py b/tests/unit/test_daemon.py index bd135060..e68d1033 100644 --- a/tests/unit/test_daemon.py +++ b/tests/unit/test_daemon.py @@ -19,16 +19,12 @@ # along with input-remapper. If not, see . -from tests.test import ( - cleanup, - uinput_write_history_pipe, - new_event, - push_events, - is_service_running, - fixtures, - tmp, - get_key_mapping, -) +from tests.test import is_service_running +from tests.lib.cleanup import cleanup +from tests.lib.fixtures import new_event +from tests.lib.pipes import push_events, uinput_write_history_pipe +from tests.lib.tmp import tmp +from tests.lib.fixtures import fixtures, get_key_mapping import os import unittest diff --git a/tests/unit/test_data_manager.py b/tests/unit/test_data_manager.py index a9adb3c7..d01099c2 100644 --- a/tests/unit/test_data_manager.py +++ b/tests/unit/test_data_manager.py @@ -17,12 +17,11 @@ # # You should have received a copy of the GNU General Public License # along with input-remapper. If not, see . -import json import os import time import unittest from itertools import permutations -from typing import List, Dict, Any +from typing import List from unittest.mock import MagicMock, call from inputremapper.configs.global_config import global_config @@ -37,15 +36,16 @@ from inputremapper.gui.messages.message_broker import ( ) from inputremapper.gui.messages.message_data import ( GroupData, - PresetData, CombinationUpdate, ) 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 +from tests.lib.cleanup import quick_cleanup +from tests.lib.patches import FakeDaemonProxy +from tests.lib.fixtures import prepare_presets -from inputremapper.configs.paths import get_preset_path, get_config_path +from inputremapper.configs.paths import get_preset_path from inputremapper.configs.preset import Preset from inputremapper.gui.data_manager import DataManager, DEFAULT_PRESET_NAME diff --git a/tests/unit/test_event_pipeline/test_event_pipeline.py b/tests/unit/test_event_pipeline/test_event_pipeline.py index a8ee373d..421bf206 100644 --- a/tests/unit/test_event_pipeline/test_event_pipeline.py +++ b/tests/unit/test_event_pipeline/test_event_pipeline.py @@ -59,15 +59,14 @@ from inputremapper.injection.context import Context from inputremapper.injection.event_reader import EventReader from inputremapper.injection.global_uinputs import global_uinputs from inputremapper.input_event import InputEvent, USE_AS_ANALOG_VALUE -from tests.test import ( - get_key_mapping, - cleanup, - convert_to_internal_events, - MAX_ABS, - MIN_ABS, +from tests.lib.cleanup import cleanup +from tests.lib.logger import logger +from tests.lib.constants import MAX_ABS, MIN_ABS +from tests.lib.stuff import convert_to_internal_events +from tests.lib.fixtures import ( Fixture, fixtures, - logger, + get_key_mapping, ) diff --git a/tests/unit/test_event_pipeline/test_mapping_handlers.py b/tests/unit/test_event_pipeline/test_mapping_handlers.py index ff013128..4e4c95c4 100644 --- a/tests/unit/test_event_pipeline/test_mapping_handlers.py +++ b/tests/unit/test_event_pipeline/test_mapping_handlers.py @@ -56,12 +56,10 @@ from inputremapper.injection.mapping_handlers.macro_handler import MacroHandler from inputremapper.injection.mapping_handlers.mapping_handler import MappingHandler from inputremapper.injection.mapping_handlers.rel_to_abs_handler import RelToAbsHandler from inputremapper.input_event import InputEvent, EventActions, USE_AS_ANALOG_VALUE -from tests.test import ( - InputDevice, - cleanup, - convert_to_internal_events, - MAX_ABS, -) +from tests.lib.cleanup import cleanup +from tests.lib.patches import InputDevice +from tests.lib.constants import MAX_ABS +from tests.lib.stuff import convert_to_internal_events class BaseTests: diff --git a/tests/unit/test_event_reader.py b/tests/unit/test_event_reader.py index bcfcc400..2e0dc872 100644 --- a/tests/unit/test_event_reader.py +++ b/tests/unit/test_event_reader.py @@ -43,7 +43,9 @@ from inputremapper.event_combination import EventCombination from inputremapper.injection.context import Context from inputremapper.injection.event_reader import EventReader from inputremapper.injection.global_uinputs import global_uinputs -from tests.test import new_event, quick_cleanup, get_key_mapping +from tests.lib.fixtures import new_event +from tests.lib.cleanup import quick_cleanup +from tests.lib.fixtures import get_key_mapping class TestEventReader(unittest.IsolatedAsyncioTestCase): diff --git a/tests/unit/test_global_uinputs.py b/tests/unit/test_global_uinputs.py index 7049663b..8af872f2 100644 --- a/tests/unit/test_global_uinputs.py +++ b/tests/unit/test_global_uinputs.py @@ -19,7 +19,7 @@ # along with input-remapper. If not, see . -from tests.test import cleanup +from tests.lib.cleanup import cleanup import sys import unittest diff --git a/tests/unit/test_groups.py b/tests/unit/test_groups.py index a9e24d35..63641fcf 100644 --- a/tests/unit/test_groups.py +++ b/tests/unit/test_groups.py @@ -19,7 +19,8 @@ # along with input-remapper. If not, see . -from tests.test import quick_cleanup, fixtures, keyboard_keys +from tests.lib.cleanup import quick_cleanup +from tests.lib.fixtures import fixtures, keyboard_keys import os import unittest diff --git a/tests/unit/test_injector.py b/tests/unit/test_injector.py index 385fe6d4..86961793 100644 --- a/tests/unit/test_injector.py +++ b/tests/unit/test_injector.py @@ -19,15 +19,14 @@ # along with input-remapper. If not, see . from pydantic import ValidationError -from tests.test import ( - new_event, - push_events, - fixtures, - EVENT_READ_TIMEOUT, - uinput_write_history_pipe, - quick_cleanup, - read_write_history_pipe, - uinputs, +from tests.lib.fixtures import new_event +from tests.lib.patches import uinputs +from tests.lib.cleanup import quick_cleanup +from tests.lib.constants import EVENT_READ_TIMEOUT +from tests.lib.fixtures import fixtures +from tests.lib.pipes import uinput_write_history_pipe +from tests.lib.pipes import read_write_history_pipe, push_events +from tests.lib.fixtures import ( keyboard_keys, get_key_mapping, ) diff --git a/tests/unit/test_ipc.py b/tests/unit/test_ipc.py index 80e60091..4a6127a8 100644 --- a/tests/unit/test_ipc.py +++ b/tests/unit/test_ipc.py @@ -20,7 +20,8 @@ import asyncio import multiprocessing -from tests.test import quick_cleanup, tmp +from tests.lib.cleanup import quick_cleanup +from tests.lib.tmp import tmp import unittest import select diff --git a/tests/unit/test_logger.py b/tests/unit/test_logger.py index 9830c8e5..0d15f85c 100644 --- a/tests/unit/test_logger.py +++ b/tests/unit/test_logger.py @@ -19,7 +19,7 @@ # along with input-remapper. If not, see . -from tests.test import tmp +from tests.lib.tmp import tmp import os import shutil diff --git a/tests/unit/test_macros.py b/tests/unit/test_macros.py index 505e6506..965d0406 100644 --- a/tests/unit/test_macros.py +++ b/tests/unit/test_macros.py @@ -65,7 +65,9 @@ from inputremapper.injection.macros.parse import ( get_macro_argument_names, get_num_parameters, ) -from tests.test import logger, quick_cleanup, new_event +from tests.lib.fixtures import new_event +from tests.lib.logger import logger +from tests.lib.cleanup import quick_cleanup class MacroTestBase(unittest.IsolatedAsyncioTestCase): diff --git a/tests/unit/test_migrations.py b/tests/unit/test_migrations.py index f1afaace..bc5cec53 100644 --- a/tests/unit/test_migrations.py +++ b/tests/unit/test_migrations.py @@ -11,8 +11,10 @@ # # You should have received a copy of the GNU General Public License # along with input-remapper. If not, see . + from inputremapper.configs.mapping import UIMapping -from tests.test import quick_cleanup, tmp +from tests.lib.cleanup import quick_cleanup +from tests.lib.tmp import tmp import os import unittest diff --git a/tests/unit/test_paths.py b/tests/unit/test_paths.py index bc4eb233..a7c7521e 100644 --- a/tests/unit/test_paths.py +++ b/tests/unit/test_paths.py @@ -19,7 +19,8 @@ # along with input-remapper. If not, see . -from tests.test import quick_cleanup, tmp +from tests.lib.cleanup import quick_cleanup +from tests.lib.tmp import tmp import os import unittest diff --git a/tests/unit/test_preset.py b/tests/unit/test_preset.py index b9936413..2bdf7c08 100644 --- a/tests/unit/test_preset.py +++ b/tests/unit/test_preset.py @@ -30,7 +30,8 @@ from inputremapper.configs.paths import get_preset_path, get_config_path, CONFIG from inputremapper.configs.preset import Preset from inputremapper.event_combination import EventCombination from inputremapper.input_event import InputEvent -from tests.test import quick_cleanup, get_key_mapping +from tests.lib.cleanup import quick_cleanup +from tests.lib.fixtures import get_key_mapping class TestPreset(unittest.TestCase): diff --git a/tests/unit/test_reader.py b/tests/unit/test_reader.py index ada8456c..99fbdaef 100644 --- a/tests/unit/test_reader.py +++ b/tests/unit/test_reader.py @@ -50,17 +50,16 @@ 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, +from tests.lib.fixtures import new_event +from tests.lib.cleanup import quick_cleanup +from tests.lib.constants import ( EVENT_READ_TIMEOUT, START_READING_DELAY, - quick_cleanup, MAX_ABS, MIN_ABS, - fixtures, - push_event, ) +from tests.lib.pipes import push_event, push_events +from tests.lib.fixtures import fixtures CODE_1 = 100 CODE_2 = 101 diff --git a/tests/unit/test_system_mapping.py b/tests/unit/test_system_mapping.py index eb6bb005..6737ace3 100644 --- a/tests/unit/test_system_mapping.py +++ b/tests/unit/test_system_mapping.py @@ -28,7 +28,7 @@ from evdev.ecodes import BTN_LEFT, KEY_A from inputremapper.configs.paths import CONFIG_PATH from inputremapper.configs.system_mapping import SystemMapping, XMODMAP_FILENAME -from tests.test import quick_cleanup +from tests.lib.cleanup import quick_cleanup class TestSystemMapping(unittest.TestCase): diff --git a/tests/unit/test_test.py b/tests/unit/test_test.py index 1501eb9f..b97648c9 100644 --- a/tests/unit/test_test.py +++ b/tests/unit/test_test.py @@ -17,18 +17,15 @@ # # You should have received a copy of the GNU General Public License # along with input-remapper. If not, see . + from inputremapper.gui.messages.message_broker import MessageBroker -from tests.test import ( - InputDevice, - quick_cleanup, - cleanup, - fixtures, - new_event, - push_events, - EVENT_READ_TIMEOUT, - START_READING_DELAY, - logger, -) +from tests.lib.fixtures import new_event +from tests.lib.cleanup import cleanup, quick_cleanup +from tests.lib.constants import EVENT_READ_TIMEOUT, START_READING_DELAY +from tests.lib.logger import logger +from tests.lib.fixtures import fixtures +from tests.lib.pipes import push_events +from tests.lib.patches import InputDevice import os import unittest diff --git a/tests/unit/test_user.py b/tests/unit/test_user.py index d62b2011..dc3caa60 100644 --- a/tests/unit/test_user.py +++ b/tests/unit/test_user.py @@ -19,7 +19,7 @@ # along with input-remapper. If not, see . -from tests.test import quick_cleanup +from tests.lib.cleanup import quick_cleanup import os import unittest