Add Integrationtests for threaded listeners

bin/solaar

@@ -24,20 +24,7 @@ def init_paths():
     import os.path
     import sys
 
-    # Python 3 might have problems converting back to UTF-8 in case of Unicode surrogates
-    decoded_path = None
-    try:
-        decoded_path = sys.path[0]
-        sys.path[0].encode(sys.getfilesystemencoding())
-
-    except UnicodeError:
-        sys.stderr.write(
-            "ERROR: Solaar cannot recognize encoding of filesystem path, "
-            "this may happen due to non UTF-8 characters in the pathname.\n"
-        )
-        sys.exit(1)
-
-    root = os.path.join(os.path.realpath(decoded_path), "..")
+    root = os.path.join(os.path.realpath(sys.path[0]), "..")
     prefix = os.path.normpath(root)
     src_lib = os.path.join(prefix, "lib")
     share_lib = os.path.join(prefix, "share", "solaar", "lib")

lib/hidapi/hidapi_impl.py

@@ -171,7 +171,7 @@ class HIDError(Exception):
     pass
 
 
-def _enumerate_devices():
+def _enumerate_devices() -> list:
     """Returns all HID devices which are potentially useful to us"""
     devices = []
     c_devices = _hidapi.hid_enumerate(0, 0)
@@ -201,20 +201,23 @@ def _enumerate_devices():
 
 
 # Use a separate thread to check if devices have been removed or connected
-class _DeviceMonitor(Thread):
+class DeviceMonitor(Thread):
     def __init__(self, device_callback, polling_delay=5.0):
         self.device_callback = device_callback
         self.polling_delay = polling_delay
         self.prev_devices = None
+        self.alive = False
+        self.abort_triggered = False
         # daemon threads are automatically killed when main thread exits
         super().__init__(daemon=True)
 
     def run(self):
+        self.alive = True
         # Populate initial set of devices so startup doesn't cause any callbacks
         self.prev_devices = {tuple(dev.items()): dev for dev in _enumerate_devices()}
 
         # Continously enumerate devices and raise callback for changes
-        while True:
+        while not self.abort_triggered:
             current_devices = {tuple(dev.items()): dev for dev in _enumerate_devices()}
             for key, device in self.prev_devices.items():
                 if key not in current_devices:
@@ -225,6 +228,11 @@ def run(self):
             self.prev_devices = current_devices
             sleep(self.polling_delay)
 
+        self.alive = False
+
+    def stop(self):
+        self.abort_triggered = True
+
 
 def _match(
     action: str,
@@ -359,11 +367,11 @@ def device_callback(action: str, device):
             # Removed devices will be detected by Solaar directly
             pass
 
-    monitor = _DeviceMonitor(device_callback=device_callback)
+    monitor = DeviceMonitor(device_callback=device_callback)
     monitor.start()
 
 
-def enumerate(filter_func) -> DeviceInfo:
+def enumerate(filter_func: Callable) -> DeviceInfo:
     """Enumerate the HID Devices.
 
     List all the HID devices attached to the system, optionally filtering by

lib/logitech_receiver/listener.py

@@ -19,25 +19,42 @@
 import queue
 import threading
 
-from . import base
+from typing import Any
+from typing import Protocol
+
 from . import exceptions
 
 logger = logging.getLogger(__name__)
 
 
+class LowLevelInterface(Protocol):
+    def open_path(self, path):
+        ...
+
+    def ping(self, handle, number, long_message=False):
+        ...
+
+    def make_notification(self, report_id: int, devnumber: int, data: bytes) -> Any:
+        ...
+
+    def close(self, handle):
+        ...
+
+
 class _ThreadedHandle:
     """A thread-local wrapper with different open handles for each thread.
     Closing a ThreadedHandle will close all handles.
     """
 
-    __slots__ = ("path", "_local", "_handles", "_listener")
+    __slots__ = ("path", "_local", "_handles", "_listener", "_base")
 
-    def __init__(self, listener, path, handle):
+    def __init__(self, listener, path, handle, low_level_api: LowLevelInterface):
         assert listener is not None
         assert path is not None
         assert handle is not None
         assert isinstance(handle, int)
 
+        self._base = low_level_api
         self._listener = listener
         self.path = path
         self._local = threading.local()
@@ -46,7 +63,7 @@ def __init__(self, listener, path, handle):
         self._handles = [handle]
 
     def _open(self):
-        handle = base.open_path(self.path)
+        handle = self._base.open_path(self.path)
         if handle is None:
             logger.error("%r failed to open new handle", self)
         else:
@@ -63,7 +80,7 @@ def close(self):
             if logger.isEnabledFor(logging.DEBUG):
                 logger.debug("%r closing %s", self, handles)
             for h in handles:
-                base.close(h)
+                self._base.close(h)
 
     @property
     def notifications_hook(self):
@@ -112,12 +129,13 @@ class EventsListener(threading.Thread):
     Incoming packets will be passed to the callback function in sequence.
     """
 
-    def __init__(self, receiver, notifications_callback):
+    def __init__(self, receiver, notifications_callback, low_level: LowLevelInterface):
         try:
             path_name = receiver.path.split("/")[2]
         except IndexError:
             path_name = receiver.path
         super().__init__(name=self.__class__.__name__ + ":" + path_name)
+        self._base = low_level
         self.daemon = True
         self._active = False
         self.receiver = receiver
@@ -127,7 +145,7 @@ def __init__(self, receiver, notifications_callback):
     def run(self):
         self._active = True
         # replace the handle with a threaded one
-        self.receiver.handle = _ThreadedHandle(self, self.receiver.path, self.receiver.handle)
+        self.receiver.handle = _ThreadedHandle(self, self.receiver.path, self.receiver.handle, self._base)
         if logger.isEnabledFor(logging.INFO):
             logger.info("started with %s (%d)", self.receiver, int(self.receiver.handle))
         self.has_started()
@@ -139,13 +157,13 @@ def run(self):
         while self._active:
             if self._queued_notifications.empty():
                 try:
-                    n = base.read(self.receiver.handle, _EVENT_READ_TIMEOUT)
+                    n = self._base.read(self.receiver.handle, _EVENT_READ_TIMEOUT)
                 except exceptions.NoReceiver:
                     logger.warning("%s disconnected", self.receiver.name)
                     self.receiver.close()
                     break
                 if n:
-                    n = base.make_notification(*n)
+                    n = self._base.make_notification(*n)
             else:
                 n = self._queued_notifications.get()  # deliver any queued notifications
             if n:

lib/solaar/listener.py

@@ -22,6 +22,8 @@
 
 from collections import namedtuple
 from functools import partial
+from typing import Any
+from typing import Protocol
 
 import gi
 import logitech_receiver
@@ -58,12 +60,26 @@ def _ghost(device):
     )
 
 
+class LowLevelInterface(Protocol):
+    def open_path(self, path):
+        ...
+
+    def ping(self, handle, number, long_message=False):
+        ...
+
+    def make_notification(self, report_id: int, devnumber: int, data: bytes) -> Any:
+        ...
+
+    def close(self, handle):
+        ...
+
+
 class SolaarListener(listener.EventsListener):
     """Keeps the status of a Receiver or Device (member name is receiver but it can also be a device)."""
 
-    def __init__(self, receiver, status_changed_callback):
+    def __init__(self, receiver, status_changed_callback, low_level):
         assert status_changed_callback
-        super().__init__(receiver, self._notifications_handler)
+        super().__init__(receiver, self._notifications_handler, low_level)
         self.status_changed_callback = status_changed_callback
         receiver.status_callback = self._status_changed
 
@@ -275,7 +291,7 @@ def _start(device_info):
                 receiver_.cleanups.append(_cleanup_bluez_dbus)
 
     if receiver_:
-        rl = SolaarListener(receiver_, _status_callback)
+        rl = SolaarListener(receiver_, _status_callback, base)
         rl.start()
         _all_listeners[device_info.path] = rl
         return rl

lib/solaar/tasks.py

@@ -18,15 +18,11 @@
 
 import logging
 
+from queue import Queue
 from threading import Thread
 
 logger = logging.getLogger(__name__)
 
-try:
-    from Queue import Queue
-except ImportError:
-    from queue import Queue
-
 
 class TaskRunner(Thread):
     def __init__(self, name):

tests/integrationtests/test_device_monitor.py

@@ -0,0 +1,25 @@
+import platform
+import time
+
+import pytest
+
+
+@pytest.mark.skipif(platform.system() == "Linux", reason="Test for non Linux platforms")
+def test_device_monitor(mocker):
+    from hidapi.hidapi_impl import DeviceMonitor
+
+    mock_callback = mocker.Mock()
+    monitor = DeviceMonitor(device_callback=mock_callback, polling_delay=0.1)
+    monitor.start()
+
+    while not monitor.alive:
+        time.sleep(0.01)
+
+    assert monitor.alive
+
+    monitor.stop()
+
+    while monitor.alive:
+        time.sleep(0.01)
+
+    assert not monitor.alive

tests/integrationtests/test_events_listener.py

@@ -0,0 +1,16 @@
+from logitech_receiver.listener import EventsListener
+
+
+def test_events_listener(mocker):
+    receiver = mocker.MagicMock()
+    receiver.handle = 1
+    receiver.path = "pathname"
+    status_callback = mocker.MagicMock()
+    low_level_mock = mocker.MagicMock()
+
+    e = EventsListener(receiver, status_callback, low_level_mock)
+    e.start()
+
+    assert bool(e)
+
+    e.stop()

tests/integrationtests/test_solaar_listener.py

@@ -0,0 +1,19 @@
+from solaar.listener import SolaarListener
+
+
+# @pytest.mark.skip(reason="Unstable")
+def test_solaar_listener(mocker):
+    receiver = mocker.MagicMock()
+    receiver.handle = mocker.MagicMock()
+    receiver.path = "dsda"
+    status_callback = mocker.MagicMock()
+    low_level_mock = mocker.MagicMock()
+
+    rl = SolaarListener(receiver, status_callback, low_level_mock)
+    rl.start()
+    rl.stop()
+
+    rl.join()
+
+    assert not rl.is_alive()
+    assert status_callback.call_count == 0

tests/integrationtests/test_task_runner.py

@@ -0,0 +1,16 @@
+from solaar import tasks
+
+
+def run_task():
+    print("Hi!")
+
+
+def test_task_runner(mocker):
+    tr = tasks.TaskRunner(name="Testrunner")
+    tr.start()
+    assert tr.alive
+
+    tr(run_task)
+
+    tr.stop()
+    assert not tr.alive

tests/logitech_receiver/test_notifications.py → tests/unittests/logitech_receiver/test_notifications.py

@@ -22,6 +22,9 @@ def ping(self, handle, number, long_message=False):
     def request(self, handle, devnumber, request_id, *params, **kwargs):
         pass
 
+    def close(self):
+        pass
+
 
 @pytest.mark.parametrize(
     "sub_id, notification_data, expected_error, expected_new_device",