1#!/bin/env python3
2# SPDX-License-Identifier: GPL-2.0
3# -*- coding: utf-8 -*-
4#
5# Copyright (c) 2017 Benjamin Tissoires <benjamin.tissoires@gmail.com>
6# Copyright (c) 2017 Red Hat, Inc.
7
8import dataclasses
9import libevdev
10import os
11import pytest
12import shutil
13import subprocess
14import time
15
16import logging
17
18from .base_device import BaseDevice, EvdevMatch, SysfsFile
19from pathlib import Path
20from typing import Final, List, Tuple
21
22logger = logging.getLogger("hidtools.test.base")
23
24# application to matches
25application_matches: Final = {
26    # pyright: ignore
27    "Accelerometer": EvdevMatch(
28        req_properties=[
29            libevdev.INPUT_PROP_ACCELEROMETER,
30        ]
31    ),
32    "Game Pad": EvdevMatch(  # in systemd, this is a lot more complex, but that will do
33        requires=[
34            libevdev.EV_ABS.ABS_X,
35            libevdev.EV_ABS.ABS_Y,
36            libevdev.EV_ABS.ABS_RX,
37            libevdev.EV_ABS.ABS_RY,
38            libevdev.EV_KEY.BTN_START,
39        ],
40        excl_properties=[
41            libevdev.INPUT_PROP_ACCELEROMETER,
42        ],
43    ),
44    "Joystick": EvdevMatch(  # in systemd, this is a lot more complex, but that will do
45        requires=[
46            libevdev.EV_ABS.ABS_RX,
47            libevdev.EV_ABS.ABS_RY,
48            libevdev.EV_KEY.BTN_START,
49        ],
50        excl_properties=[
51            libevdev.INPUT_PROP_ACCELEROMETER,
52        ],
53    ),
54    "Key": EvdevMatch(
55        requires=[
56            libevdev.EV_KEY.KEY_A,
57        ],
58        excl_properties=[
59            libevdev.INPUT_PROP_ACCELEROMETER,
60            libevdev.INPUT_PROP_DIRECT,
61            libevdev.INPUT_PROP_POINTER,
62        ],
63    ),
64    "Mouse": EvdevMatch(
65        requires=[
66            libevdev.EV_REL.REL_X,
67            libevdev.EV_REL.REL_Y,
68            libevdev.EV_KEY.BTN_LEFT,
69        ],
70        excl_properties=[
71            libevdev.INPUT_PROP_ACCELEROMETER,
72        ],
73    ),
74    "Pad": EvdevMatch(
75        requires=[
76            libevdev.EV_KEY.BTN_0,
77        ],
78        excludes=[
79            libevdev.EV_KEY.BTN_TOOL_PEN,
80            libevdev.EV_KEY.BTN_TOUCH,
81            libevdev.EV_ABS.ABS_DISTANCE,
82        ],
83        excl_properties=[
84            libevdev.INPUT_PROP_ACCELEROMETER,
85        ],
86    ),
87    "Pen": EvdevMatch(
88        requires=[
89            libevdev.EV_KEY.BTN_STYLUS,
90            libevdev.EV_ABS.ABS_X,
91            libevdev.EV_ABS.ABS_Y,
92        ],
93        excl_properties=[
94            libevdev.INPUT_PROP_ACCELEROMETER,
95        ],
96    ),
97    "Stylus": EvdevMatch(
98        requires=[
99            libevdev.EV_KEY.BTN_STYLUS,
100            libevdev.EV_ABS.ABS_X,
101            libevdev.EV_ABS.ABS_Y,
102        ],
103        excl_properties=[
104            libevdev.INPUT_PROP_ACCELEROMETER,
105        ],
106    ),
107    "Touch Pad": EvdevMatch(
108        requires=[
109            libevdev.EV_KEY.BTN_LEFT,
110            libevdev.EV_ABS.ABS_X,
111            libevdev.EV_ABS.ABS_Y,
112        ],
113        excludes=[libevdev.EV_KEY.BTN_TOOL_PEN, libevdev.EV_KEY.BTN_STYLUS],
114        req_properties=[
115            libevdev.INPUT_PROP_POINTER,
116        ],
117        excl_properties=[
118            libevdev.INPUT_PROP_ACCELEROMETER,
119        ],
120    ),
121    "Touch Screen": EvdevMatch(
122        requires=[
123            libevdev.EV_KEY.BTN_TOUCH,
124            libevdev.EV_ABS.ABS_X,
125            libevdev.EV_ABS.ABS_Y,
126        ],
127        excludes=[libevdev.EV_KEY.BTN_TOOL_PEN, libevdev.EV_KEY.BTN_STYLUS],
128        req_properties=[
129            libevdev.INPUT_PROP_DIRECT,
130        ],
131        excl_properties=[
132            libevdev.INPUT_PROP_ACCELEROMETER,
133        ],
134    ),
135}
136
137
138class UHIDTestDevice(BaseDevice):
139    def __init__(self, name, application, rdesc_str=None, rdesc=None, input_info=None):
140        super().__init__(name, application, rdesc_str, rdesc, input_info)
141        self.application_matches = application_matches
142        if name is None:
143            name = f"uhid test {self.__class__.__name__}"
144        if not name.startswith("uhid test "):
145            name = "uhid test " + self.name
146        self.name = name
147
148
149@dataclasses.dataclass
150class HidBpf:
151    object_name: str
152    has_rdesc_fixup: bool
153
154
155@dataclasses.dataclass
156class KernelModule:
157    driver_name: str
158    module_name: str
159
160
161class BaseTestCase:
162    class TestUhid(object):
163        syn_event = libevdev.InputEvent(libevdev.EV_SYN.SYN_REPORT)  # type: ignore
164        key_event = libevdev.InputEvent(libevdev.EV_KEY)  # type: ignore
165        abs_event = libevdev.InputEvent(libevdev.EV_ABS)  # type: ignore
166        rel_event = libevdev.InputEvent(libevdev.EV_REL)  # type: ignore
167        msc_event = libevdev.InputEvent(libevdev.EV_MSC.MSC_SCAN)  # type: ignore
168
169        # List of kernel modules to load before starting the test
170        # if any module is not available (not compiled), the test will skip.
171        # Each element is a KernelModule object, for example
172        # KernelModule("playstation", "hid-playstation")
173        kernel_modules: List[KernelModule] = []
174
175        # List of in kernel HID-BPF object files to load
176        # before starting the test
177        # Any existing pre-loaded HID-BPF module will be removed
178        # before the ones in this list will be manually loaded.
179        # Each Element is a HidBpf object, for example
180        # 'HidBpf("xppen-ArtistPro16Gen2.bpf.o", True)'
181        # If 'has_rdesc_fixup' is True, the test needs to wait
182        # for one unbind and rebind before it can be sure the kernel is
183        # ready
184        hid_bpfs: List[HidBpf] = []
185
186        def assertInputEventsIn(self, expected_events, effective_events):
187            effective_events = effective_events.copy()
188            for ev in expected_events:
189                assert ev in effective_events
190                effective_events.remove(ev)
191            return effective_events
192
193        def assertInputEvents(self, expected_events, effective_events):
194            remaining = self.assertInputEventsIn(expected_events, effective_events)
195            assert remaining == []
196
197        @classmethod
198        def debug_reports(cls, reports, uhdev=None, events=None):
199            data = [" ".join([f"{v:02x}" for v in r]) for r in reports]
200
201            if uhdev is not None:
202                human_data = [
203                    uhdev.parsed_rdesc.format_report(r, split_lines=True)
204                    for r in reports
205                ]
206                try:
207                    human_data = [
208                        f'\n\t       {" " * h.index("/")}'.join(h.split("\n"))
209                        for h in human_data
210                    ]
211                except ValueError:
212                    # '/' not found: not a numbered report
213                    human_data = ["\n\t      ".join(h.split("\n")) for h in human_data]
214                data = [f"{d}\n\t ====> {h}" for d, h in zip(data, human_data)]
215
216            reports = data
217
218            if len(reports) == 1:
219                print("sending 1 report:")
220            else:
221                print(f"sending {len(reports)} reports:")
222            for report in reports:
223                print("\t", report)
224
225            if events is not None:
226                print("events received:", events)
227
228        def create_device(self):
229            raise Exception("please reimplement me in subclasses")
230
231        def _load_kernel_module(self, kernel_driver, kernel_module):
232            sysfs_path = Path("/sys/bus/hid/drivers")
233            if kernel_driver is not None:
234                sysfs_path /= kernel_driver
235            else:
236                # special case for when testing all available modules:
237                # we don't know beforehand the name of the module from modinfo
238                sysfs_path = Path("/sys/module") / kernel_module.replace("-", "_")
239            if not sysfs_path.exists():
240                ret = subprocess.run(["/usr/sbin/modprobe", kernel_module])
241                if ret.returncode != 0:
242                    pytest.skip(
243                        f"module {kernel_module} could not be loaded, skipping the test"
244                    )
245
246        @pytest.fixture()
247        def load_kernel_module(self):
248            for k in self.kernel_modules:
249                self._load_kernel_module(k.driver_name, k.module_name)
250            yield
251
252        def load_hid_bpfs(self):
253            # this function will only work when run in the kernel tree
254            script_dir = Path(os.path.dirname(os.path.realpath(__file__)))
255            root_dir = (script_dir / "../../../../..").resolve()
256            bpf_dir = root_dir / "drivers/hid/bpf/progs"
257
258            if not bpf_dir.exists():
259                pytest.skip("looks like we are not in the kernel tree, skipping")
260
261            udev_hid_bpf = shutil.which("udev-hid-bpf")
262            if not udev_hid_bpf:
263                pytest.skip("udev-hid-bpf not found in $PATH, skipping")
264
265            wait = any(b.has_rdesc_fixup for b in self.hid_bpfs)
266
267            for hid_bpf in self.hid_bpfs:
268                # We need to start `udev-hid-bpf` in the background
269                # and dispatch uhid events in case the kernel needs
270                # to fetch features on the device
271                process = subprocess.Popen(
272                    [
273                        "udev-hid-bpf",
274                        "--verbose",
275                        "add",
276                        str(self.uhdev.sys_path),
277                        str(bpf_dir / hid_bpf.object_name),
278                    ],
279                )
280                while process.poll() is None:
281                    self.uhdev.dispatch(1)
282
283                if process.returncode != 0:
284                    pytest.fail(
285                        f"Couldn't insert hid-bpf program '{hid_bpf}', marking the test as failed"
286                    )
287
288            if wait:
289                # the HID-BPF program exports a rdesc fixup, so it needs to be
290                # unbound by the kernel and then rebound.
291                # Ensure we get the bound event exactly 2 times (one for the normal
292                # uhid loading, and then the reload from HID-BPF)
293                now = time.time()
294                while self.uhdev.kernel_ready_count < 2 and time.time() - now < 2:
295                    self.uhdev.dispatch(1)
296
297                if self.uhdev.kernel_ready_count < 2:
298                    pytest.fail(
299                        f"Couldn't insert hid-bpf programs, marking the test as failed"
300                    )
301
302        def unload_hid_bpfs(self):
303            ret = subprocess.run(
304                ["udev-hid-bpf", "--verbose", "remove", str(self.uhdev.sys_path)],
305            )
306            if ret.returncode != 0:
307                pytest.fail(
308                    f"Couldn't unload hid-bpf programs, marking the test as failed"
309                )
310
311        @pytest.fixture()
312        def new_uhdev(self, load_kernel_module):
313            return self.create_device()
314
315        def assertName(self, uhdev):
316            evdev = uhdev.get_evdev()
317            assert uhdev.name in evdev.name
318
319        @pytest.fixture(autouse=True)
320        def context(self, new_uhdev, request):
321            try:
322                with HIDTestUdevRule.instance():
323                    with new_uhdev as self.uhdev:
324                        for skip_cond in request.node.iter_markers("skip_if_uhdev"):
325                            test, message, *rest = skip_cond.args
326
327                            if test(self.uhdev):
328                                pytest.skip(message)
329
330                        self.uhdev.create_kernel_device()
331                        now = time.time()
332                        while not self.uhdev.is_ready() and time.time() - now < 5:
333                            self.uhdev.dispatch(1)
334
335                        if self.hid_bpfs:
336                            self.load_hid_bpfs()
337
338                        if self.uhdev.get_evdev() is None:
339                            logger.warning(
340                                f"available list of input nodes: (default application is '{self.uhdev.application}')"
341                            )
342                            logger.warning(self.uhdev.input_nodes)
343                        yield
344                        if self.hid_bpfs:
345                            self.unload_hid_bpfs()
346                        self.uhdev = None
347            except PermissionError:
348                pytest.skip("Insufficient permissions, run me as root")
349
350        @pytest.fixture(autouse=True)
351        def check_taint(self):
352            # we are abusing SysfsFile here, it's in /proc, but meh
353            taint_file = SysfsFile("/proc/sys/kernel/tainted")
354            taint = taint_file.int_value
355
356            yield
357
358            assert taint_file.int_value == taint
359
360        def test_creation(self):
361            """Make sure the device gets processed by the kernel and creates
362            the expected application input node.
363
364            If this fail, there is something wrong in the device report
365            descriptors."""
366            uhdev = self.uhdev
367            assert uhdev is not None
368            assert uhdev.get_evdev() is not None
369            self.assertName(uhdev)
370            assert len(uhdev.next_sync_events()) == 0
371            assert uhdev.get_evdev() is not None
372
373
374class HIDTestUdevRule(object):
375    _instance = None
376    """
377    A context-manager compatible class that sets up our udev rules file and
378    deletes it on context exit.
379
380    This class is tailored to our test setup: it only sets up the udev rule
381    on the **second** context and it cleans it up again on the last context
382    removed. This matches the expected pytest setup: we enter a context for
383    the session once, then once for each test (the first of which will
384    trigger the udev rule) and once the last test exited and the session
385    exited, we clean up after ourselves.
386    """
387
388    def __init__(self):
389        self.refs = 0
390        self.rulesfile = None
391
392    def __enter__(self):
393        self.refs += 1
394        if self.refs == 2 and self.rulesfile is None:
395            self.create_udev_rule()
396            self.reload_udev_rules()
397
398    def __exit__(self, exc_type, exc_value, traceback):
399        self.refs -= 1
400        if self.refs == 0 and self.rulesfile:
401            os.remove(self.rulesfile.name)
402            self.reload_udev_rules()
403
404    def reload_udev_rules(self):
405        subprocess.run("udevadm control --reload-rules".split())
406        subprocess.run("systemd-hwdb update".split())
407
408    def create_udev_rule(self):
409        import tempfile
410
411        os.makedirs("/run/udev/rules.d", exist_ok=True)
412        with tempfile.NamedTemporaryFile(
413            prefix="91-uhid-test-device-REMOVEME-",
414            suffix=".rules",
415            mode="w+",
416            dir="/run/udev/rules.d",
417            delete=False,
418        ) as f:
419            f.write(
420                """
421KERNELS=="*input*", ATTRS{name}=="*uhid test *", ENV{LIBINPUT_IGNORE_DEVICE}="1"
422KERNELS=="*hid*", ENV{HID_NAME}=="*uhid test *", ENV{HID_BPF_IGNORE_DEVICE}="1"
423KERNELS=="*input*", ATTRS{name}=="*uhid test * System Multi Axis", ENV{ID_INPUT_TOUCHSCREEN}="", ENV{ID_INPUT_SYSTEM_MULTIAXIS}="1"
424"""
425            )
426            self.rulesfile = f
427
428    @classmethod
429    def instance(cls):
430        if not cls._instance:
431            cls._instance = HIDTestUdevRule()
432        return cls._instance
433