From 0a9bad3fb5b15cfef066a2b3bb2668b484cbceaf Mon Sep 17 00:00:00 2001 From: MattHag <16444067+MattHag@users.noreply.github.com> Date: Sun, 19 May 2024 17:07:30 +0200 Subject: [PATCH] Refine interfaces for testability --- lib/logitech_receiver/device.py | 47 ++++++-- lib/logitech_receiver/receiver.py | 21 +++- lib/solaar/cli/__init__.py | 9 +- lib/solaar/listener.py | 2 +- tests/logitech_receiver/test_device.py | 148 ++++++++++++------------- 5 files changed, 135 insertions(+), 92 deletions(-) diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index 8624aca1dd..8d14e60681 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -21,6 +21,8 @@ from typing import Callable from typing import Optional +from typing import Protocol +from typing import cast import hidapi @@ -46,23 +48,49 @@ _IR = hidpp10_constants.INFO_SUBREGISTERS +class LowLevelInterface(Protocol): + def open_path(self, path): + ... + + def ping(self, handle, number, long_message: bool): + ... + + def request(self, handle, devnumber, request_id, *params, **kwargs): + ... + + def close(self, handle, *args, **kwargs) -> bool: + ... + + +low_level_interface = cast(LowLevelInterface, base) + + class DeviceFactory: @staticmethod - def create_device(device_info, setting_callback=None): + def create_device(low_level: LowLevelInterface, device_info, setting_callback=None): """Opens a Logitech Device found attached to the machine, by Linux device path. :returns: An open file handle for the found receiver, or None. """ try: - handle = base.open_path(device_info.path) + handle = low_level.open_path(device_info.path) if handle: # a direct connected device might not be online (as reported by user) - return Device(None, None, None, handle=handle, device_info=device_info, setting_callback=setting_callback) + return Device( + low_level, + None, + None, + None, + handle=handle, + device_info=device_info, + setting_callback=setting_callback, + ) except OSError as e: logger.exception("open %s", device_info) if e.errno == errno.EACCES: raise except Exception: logger.exception("open %s", device_info) + raise class Device: @@ -72,6 +100,7 @@ class Device: def __init__( self, + low_level: LowLevelInterface, receiver, number, online, @@ -83,6 +112,7 @@ def __init__( assert receiver or device_info if receiver: assert 0 < number <= 15 # some receivers have devices past their max # of devices + self.low_level = low_level self.number = number # will be None at this point for directly connected devices self.online = online # is the device online? - gates many atempts to contact the device self.descriptor = None @@ -129,11 +159,11 @@ def __init__( self.path = hidapi.find_paired_node(receiver.path, number, 1) if receiver else None if not self.handle: try: - self.handle = base.open_path(self.path) if self.path else None + self.handle = self.low_level.open_path(self.path) if self.path else None except Exception: # maybe the device wasn't set up try: time.sleep(1) - self.handle = base.open_path(self.path) if self.path else None + self.handle = self.low_level.open_path(self.path) if self.path else None except Exception: # give up self.handle = None # should this give up completely? @@ -484,7 +514,7 @@ def request(self, request_id, *params, no_reply=False): long = self.hidpp_long is True or ( self.hidpp_long is None and (self.bluetooth or self._protocol is not None and self._protocol >= 2.0) ) - return base.request( + return self.low_level.request( self.handle or self.receiver.handle, self.number, request_id, @@ -503,7 +533,8 @@ def ping(self): long = self.hidpp_long is True or ( self.hidpp_long is None and (self.bluetooth or self._protocol is not None and self._protocol >= 2.0) ) - protocol = base.ping(self.handle or self.receiver.handle, self.number, long_message=long) + handle = self.handle or self.receiver.handle + protocol = self.low_level.ping(handle, self.number, long_message=long) self.online = protocol is not None if protocol: self._protocol = protocol @@ -519,7 +550,7 @@ def close(self): if hasattr(self, "cleanups"): for cleanup in self.cleanups: cleanup(self) - return handle and base.close(handle) + return handle and self.low_level.close(handle) def __index__(self): return self.number diff --git a/lib/logitech_receiver/receiver.py b/lib/logitech_receiver/receiver.py index 753c8687d8..b5b91ad60e 100644 --- a/lib/logitech_receiver/receiver.py +++ b/lib/logitech_receiver/receiver.py @@ -22,6 +22,8 @@ from dataclasses import dataclass from typing import Callable from typing import Optional +from typing import Protocol +from typing import cast import hidapi @@ -42,6 +44,23 @@ _IR = hidpp10_constants.INFO_SUBREGISTERS +class LowLevelInterface(Protocol): + def open_path(self, path): + ... + + def ping(self, handle, number, long_message=False): + ... + + def request(self, handle, devnumber, request_id, *params, **kwargs): + ... + + def close(self, handle): + ... + + +low_level_interface = cast(LowLevelInterface, base) + + @dataclass class Pairing: """Information about the current or most recent pairing""" @@ -228,7 +247,7 @@ def register_new_device(self, number, notification=None): logger.warning("mismatch on device kind %s %s", info["kind"], nkind) else: online = True - dev = Device(self, number, online, pairing_info=info, setting_callback=self.setting_callback) + dev = Device(low_level_interface, self, number, online, pairing_info=info, setting_callback=self.setting_callback) if logger.isEnabledFor(logging.INFO): logger.info("%s: found new device %d (%s)", self, number, dev.wpid) self._devices[number] = dev diff --git a/lib/solaar/cli/__init__.py b/lib/solaar/cli/__init__.py index 9da8ab14c2..bd4a158a95 100644 --- a/lib/solaar/cli/__init__.py +++ b/lib/solaar/cli/__init__.py @@ -25,8 +25,7 @@ import logitech_receiver.device as _device import logitech_receiver.receiver as _receiver -from logitech_receiver.base import receivers -from logitech_receiver.base import receivers_and_devices +from logitech_receiver import base from solaar import NAME @@ -108,7 +107,7 @@ def _create_parser(): def _receivers(dev_path=None): - for dev_info in receivers(): + for dev_info in base.receivers(): if dev_path is not None and dev_path != dev_info.path: continue try: @@ -123,12 +122,12 @@ def _receivers(dev_path=None): def _receivers_and_devices(dev_path=None): - for dev_info in receivers_and_devices(): + for dev_info in base.receivers_and_devices(): if dev_path is not None and dev_path != dev_info.path: continue try: if dev_info.isDevice: - d = _device.DeviceFactory.create_device(dev_info) + d = _device.DeviceFactory.create_device(base, dev_info) else: d = _receiver.ReceiverFactory.create_receiver(dev_info) diff --git a/lib/solaar/listener.py b/lib/solaar/listener.py index f0088fceac..a8dfd7131a 100644 --- a/lib/solaar/listener.py +++ b/lib/solaar/listener.py @@ -258,7 +258,7 @@ def _start(device_info): if not isDevice: receiver = _receiver.ReceiverFactory.create_receiver(device_info, _setting_callback) else: - receiver = _device.DeviceFactory.create_device(device_info, _setting_callback) + receiver = _device.DeviceFactory.create_device(_base, device_info, _setting_callback) if receiver: configuration.attach_to(receiver) if receiver.bluetooth and receiver.hid_serial: diff --git a/tests/logitech_receiver/test_device.py b/tests/logitech_receiver/test_device.py index 69d68183cf..6979476b08 100644 --- a/tests/logitech_receiver/test_device.py +++ b/tests/logitech_receiver/test_device.py @@ -29,72 +29,80 @@ from . import hidpp -@pytest.fixture -def mock_base(): # allow override of base functions - with mock.patch("logitech_receiver.base.open_path", return_value=None) as mock_open_path: - with mock.patch("logitech_receiver.base.request", return_value=None) as mock_request: - with mock.patch("logitech_receiver.base.ping", return_value=None) as mock_ping: - yield mock_open_path, mock_request, mock_ping +class LowLevelInterfaceFake: + def __init__(self, responses=None): + self.responses = responses + + def open_path(self, path): + return hidpp.open_path(path) + + def request(self, response, *args, **kwargs): + func = partial(hidpp.request, self.responses) + return func(response, *args, **kwargs) + + def ping(self, response, *args, **kwargs): + func = partial(hidpp.ping, self.responses) + return func(response, *args, **kwargs) + + def close(self, *args, **kwargs): + pass @dataclass -class DeviceInfo: +class DeviceInfoStub: path: str + product_id: str vendor_id: int = 1133 - product_id: int = 4066 hidpp_short: bool = False hidpp_long: bool = True bus_id: int = 0x0003 # USB serial: str = "aa:aa:aa;aa" -di_bad_handle = DeviceInfo(None, product_id="CCCC") -di_error = DeviceInfo(11, product_id="CCCC") -di_CCCC = DeviceInfo("11", product_id="CCCC") -di_C318 = DeviceInfo("11", product_id="C318") -di_B530 = DeviceInfo("11", product_id="B350", bus_id=0x0005) -di_C068 = DeviceInfo("11", product_id="C06B") -di_C08A = DeviceInfo("11", product_id="C08A") -di_DDDD = DeviceInfo("11", product_id="DDDD") +di_bad_handle = DeviceInfoStub(None, product_id="CCCC") +di_error = DeviceInfoStub(11, product_id="CCCC") +di_CCCC = DeviceInfoStub("11", product_id="CCCC") +di_C318 = DeviceInfoStub("11", product_id="C318") +di_B530 = DeviceInfoStub("11", product_id="B350", bus_id=0x0005) +di_C068 = DeviceInfoStub("11", product_id="C06B") +di_C08A = DeviceInfoStub("11", product_id="C08A") +di_DDDD = DeviceInfoStub("11", product_id="DDDD") @pytest.mark.parametrize( - "device_info, responses, success", + "device_info, responses, expected_success", [(di_bad_handle, hidpp.r_empty, None), (di_error, hidpp.r_empty, False), (di_CCCC, hidpp.r_empty, True)], ) -def test_DeviceFactory(device_info, responses, success, mock_base): - mock_base[0].side_effect = hidpp.open_path - mock_base[1].side_effect = partial(hidpp.request, responses) - mock_base[2].side_effect = partial(hidpp.ping, responses) - - if success is None: - with pytest.raises(Exception): # noqa: B017 - test_device = device.DeviceFactory.create_device(device_info) +def test_create_device(device_info, responses, expected_success): + low_level_mock = LowLevelInterfaceFake(responses) + if expected_success is None: + with pytest.raises(PermissionError): + device.DeviceFactory.create_device(low_level_mock, device_info) + elif not expected_success: + with pytest.raises(TypeError): + device.DeviceFactory.create_device(low_level_mock, device_info) else: - test_device = device.DeviceFactory.create_device(device_info) - assert bool(test_device) == success + test_device = device.DeviceFactory.create_device(low_level_mock, device_info) + assert bool(test_device) == expected_success +@pytest.mark.skipif(platform.system() == "Darwin", reason="Cleanup fails on macOS") @pytest.mark.parametrize( - "device_info, responses, codename, name, kind", + "device_info, responses, expected_codename, expected_name, expected_kind", [ (di_CCCC, hidpp.r_empty, "?? (CCCC)", "Unknown device CCCC", "?"), (di_C318, hidpp.r_keyboard_1, "?? (C318)", "Unknown device C318", "?"), (di_B530, hidpp.r_keyboard_2, "ABCDEFGHIJKLMNOPQR", "ABCDEFGHIJKLMNOPQR", common.NamedInt(1, "keyboard")), ], ) -def test_Device_name(device_info, responses, codename, name, kind, mock_base): - mock_base[0].side_effect = hidpp.open_path - mock_base[1].side_effect = partial(hidpp.request, responses) - mock_base[2].side_effect = partial(hidpp.ping, responses) - test_device = device.DeviceFactory.create_device(device_info) - test_device._codename = None - test_device._name = None - test_device._kind = None +def test_device_name(device_info, responses, expected_codename, expected_name, expected_kind): + low_level = LowLevelInterfaceFake(responses) - assert test_device.codename == codename - assert test_device.name == name - assert test_device.kind == kind + test_device = device.DeviceFactory.create_device(low_level, device_info) + + assert test_device.codename == expected_codename + assert test_device.name == expected_name + assert test_device.kind == expected_kind @pytest.mark.skipif(platform.system() == "Darwin", reason="Cleanup fails on macOS") @@ -111,12 +119,8 @@ def test_Device_name(device_info, responses, codename, name, kind, mock_base): [[], [], [], (common.NamedInt(7, "battery status"), common.NamedInt(81, "three leds")), [], []], ), ) -def test_Device_info(device_info, responses, handle, _name, _codename, number, protocol, registers, mock_base): - mock_base[0].side_effect = hidpp.open_path - mock_base[1].side_effect = partial(hidpp.request, responses) - mock_base[2].side_effect = partial(hidpp.ping, responses) - - test_device = device.Device(None, None, None, handle=handle, device_info=device_info) +def test_device_info(device_info, responses, handle, _name, _codename, number, protocol, registers): + test_device = device.Device(LowLevelInterfaceFake(responses), None, None, None, handle=handle, device_info=device_info) assert test_device.handle == handle assert test_device._name == _name @@ -132,7 +136,7 @@ def test_Device_info(device_info, responses, handle, _name, _codename, number, p @dataclass -class Receiver: +class FakeReceiver: path: str = "11" handle: int = 0x11 codename: Optional[str] = None @@ -180,13 +184,14 @@ def mock_hid(): ], ), ) -def test_Device_receiver(number, pairing_info, responses, handle, _name, codename, p, p2, name, mock_base, mock_hid): - mock_base[0].side_effect = hidpp.open_path - mock_base[1].side_effect = partial(hidpp.request, hidpp.replace_number(responses, number)) - mock_base[2].side_effect = partial(hidpp.ping, hidpp.replace_number(responses, number)) +def test_device_receiver(number, pairing_info, responses, handle, _name, codename, p, p2, name, mock_hid): mock_hid.side_effect = lambda x, y, z: x - test_device = device.Device(Receiver(codename="CODE"), number, True, pairing_info, handle=handle) + low_level = LowLevelInterfaceFake(responses) + low_level.request = partial(hidpp.request, hidpp.replace_number(responses, number)) + low_level.ping = partial(hidpp.ping, hidpp.replace_number(responses, number)) + + test_device = device.Device(low_level, FakeReceiver(codename="CODE"), number, True, pairing_info, handle=handle) test_device.receiver.device = test_device assert test_device.handle == handle @@ -197,6 +202,7 @@ def test_Device_receiver(number, pairing_info, responses, handle, _name, codenam assert test_device.protocol == p2 assert test_device.codename == codename assert test_device.name == name + assert test_device == test_device assert not (test_device != test_device) assert bool(test_device) @@ -222,15 +228,14 @@ def test_Device_receiver(number, pairing_info, responses, handle, _name, codenam ["1ms", "2ms", "4ms", "8ms", "1ms", "9ms"], # polling rate ), ) -def test_Device_ids( - number, info, responses, handle, unitId, modelId, tid, kind, firmware, serial, id, psl, rate, mock_base, mock_hid -): - mock_base[0].side_effect = hidpp.open_path - mock_base[1].side_effect = partial(hidpp.request, hidpp.replace_number(responses, number)) - mock_base[2].side_effect = partial(hidpp.ping, hidpp.replace_number(responses, number)) +def test_device_ids(number, info, responses, handle, unitId, modelId, tid, kind, firmware, serial, id, psl, rate, mock_hid): mock_hid.side_effect = lambda x, y, z: x - test_device = device.Device(Receiver(), number, True, info, handle=handle) + low_level = LowLevelInterfaceFake(responses) + low_level.request = partial(hidpp.request, hidpp.replace_number(responses, number)) + low_level.ping = partial(hidpp.ping, hidpp.replace_number(responses, number)) + + test_device = device.Device(low_level, FakeReceiver(), number, True, info, handle=handle) assert test_device.unitId == unitId assert test_device.modelId == modelId @@ -242,10 +247,10 @@ def test_Device_ids( assert test_device.polling_rate == rate -class TestDevice(device.Device): # a fully functional Device but its HID++ functions look at local data +class FakeDevice(device.Device): # a fully functional Device but its HID++ functions look at local data def __init__(self, responses, *args, **kwargs): self.responses = responses - super().__init__(*args, **kwargs) + super().__init__(LowLevelInterfaceFake(responses), *args, **kwargs) request = hidpp.Device.request ping = hidpp.Device.ping @@ -262,8 +267,8 @@ def __init__(self, responses, *args, **kwargs): (di_B530, hidpp.complex_responses_2, 4.5, hidpp20.RGBEffectsInfo, 8, 3, 1, True, True), ], ) -def test_Device_complex(device_info, responses, protocol, led, keys, remap, gestures, backlight, profiles, mocker): - test_device = TestDevice(responses, None, None, True, device_info=device_info) +def test_device_complex(device_info, responses, protocol, led, keys, remap, gestures, backlight, profiles, mocker): + test_device = FakeDevice(responses, None, None, True, device_info=device_info) test_device._name = "TestDevice" test_device._protocol = protocol spy_request = mocker.spy(test_device, "request") @@ -300,9 +305,9 @@ def test_Device_complex(device_info, responses, protocol, led, keys, remap, gest (di_C08A, hidpp.r_mouse_2, 4.5, {"p": "p"}, {"p": "p"}, 0), ], ) -def test_Device_settings(device_info, responses, protocol, p, persister, settings, mocker): +def test_device_settings(device_info, responses, protocol, p, persister, settings, mocker): mocker.patch("solaar.configuration.persister", return_value=p) - test_device = TestDevice(responses, None, None, True, device_info=device_info) + test_device = FakeDevice(responses, None, None, True, device_info=device_info) test_device._name = "TestDevice" test_device._protocol = protocol @@ -318,8 +323,8 @@ def test_Device_settings(device_info, responses, protocol, p, persister, setting (di_B530, hidpp.r_keyboard_2, 4.5, common.Battery(18, 52, None, None), {"active": True, "alert": 0, "reason": None}), ], ) -def test_Device_battery(device_info, responses, protocol, battery, changed, mocker): - test_device = TestDevice(responses, None, None, online=True, device_info=device_info) +def test_device_battery(device_info, responses, protocol, battery, changed, mocker): + test_device = FakeDevice(responses, None, None, online=True, device_info=device_info) test_device._name = "TestDevice" test_device._protocol = protocol spy_changed = mocker.spy(test_device, "changed") @@ -327,14 +332,3 @@ def test_Device_battery(device_info, responses, protocol, battery, changed, mock assert test_device.battery() == battery test_device.read_battery() spy_changed.assert_called_with(**changed) - - -""" TODO - changed - enable_connection_notifications - add_notification_handler - remove_notification_handler - handle_notification -""" - -# IMPORTANT TODO - battery