Refactor rule loading for testability

Introduce tests for YAML rule loading functionality.

lib/logitech_receiver/diversion.py

@@ -176,7 +176,8 @@ def gnome_dbus_interface_setup():
         remote_object = bus.get_object("org.gnome.Shell", "/io/github/pwr_solaar/solaar")
         _dbus_interface = dbus.Interface(remote_object, "io.github.pwr_solaar.solaar")
     except dbus.exceptions.DBusException:
-        logger.warning("Solaar Gnome extension not installed - some rule capabilities inoperable", exc_info=_sys.exc_info())
+        logger.warning("Solaar Gnome extension not installed - some rule capabilities inoperable",
+                       exc_info=_sys.exc_info())
         _dbus_interface = False
     return _dbus_interface
 
@@ -217,7 +218,8 @@ def xkb_setup():
     for _, evcode in buttons.values():
         if evcode:
             key_events.append(evcode)
-    devicecap = {evdev.ecodes.EV_KEY: key_events, evdev.ecodes.EV_REL: [evdev.ecodes.REL_WHEEL, evdev.ecodes.REL_HWHEEL]}
+    devicecap = {evdev.ecodes.EV_KEY: key_events,
+                 evdev.ecodes.EV_REL: [evdev.ecodes.REL_WHEEL, evdev.ecodes.REL_HWHEEL]}
 else:
     # Just mock these since they won't be useful without evdev anyway
     buttons = {}
@@ -434,9 +436,9 @@ def thumb_wheel_down(f, r, d, a):
 
 def charging(f, r, d, a):
     if (
-        (f == _F.BATTERY_STATUS and r == 0 and 1 <= d[2] <= 4)
-        or (f == _F.BATTERY_VOLTAGE and r == 0 and d[2] & (1 << 7))
-        or (f == _F.UNIFIED_BATTERY and r == 0 and 1 <= d[2] <= 3)
+            (f == _F.BATTERY_STATUS and r == 0 and 1 <= d[2] <= 4)
+            or (f == _F.BATTERY_VOLTAGE and r == 0 and d[2] & (1 << 7))
+            or (f == _F.UNIFIED_BATTERY and r == 0 and 1 <= d[2] <= 3)
     ):
         return 1
     else:
@@ -454,10 +456,14 @@ def charging(f, r, d, a):
     "crown_pressed": [lambda f, r, d, a: f == _F.CROWN and r == 0 and d[6] >= 0x01 and d[6] <= 0x04 and d[6], False],
     "thumb_wheel_up": [thumb_wheel_up, True],
     "thumb_wheel_down": [thumb_wheel_down, True],
-    "lowres_wheel_up": [lambda f, r, d, a: f == _F.LOWRES_WHEEL and r == 0 and signed(d[0:1]) > 0 and signed(d[0:1]), False],
-    "lowres_wheel_down": [lambda f, r, d, a: f == _F.LOWRES_WHEEL and r == 0 and signed(d[0:1]) < 0 and signed(d[0:1]), False],
-    "hires_wheel_up": [lambda f, r, d, a: f == _F.HIRES_WHEEL and r == 0 and signed(d[1:3]) > 0 and signed(d[1:3]), False],
-    "hires_wheel_down": [lambda f, r, d, a: f == _F.HIRES_WHEEL and r == 0 and signed(d[1:3]) < 0 and signed(d[1:3]), False],
+    "lowres_wheel_up": [lambda f, r, d, a: f == _F.LOWRES_WHEEL and r == 0 and signed(d[0:1]) > 0 and signed(d[0:1]),
+                        False],
+    "lowres_wheel_down": [lambda f, r, d, a: f == _F.LOWRES_WHEEL and r == 0 and signed(d[0:1]) < 0 and signed(d[0:1]),
+                          False],
+    "hires_wheel_up": [lambda f, r, d, a: f == _F.HIRES_WHEEL and r == 0 and signed(d[1:3]) > 0 and signed(d[1:3]),
+                       False],
+    "hires_wheel_down": [lambda f, r, d, a: f == _F.HIRES_WHEEL and r == 0 and signed(d[1:3]) < 0 and signed(d[1:3]),
+                         False],
     "charging": [charging, False],
     "False": [lambda f, r, d, a: False, False],
     "True": [lambda f, r, d, a: True, False],
@@ -962,14 +968,14 @@ class TestBytes(Condition):
     def __init__(self, test, warn=True):
         self.test = test
         if (
-            isinstance(test, list)
-            and 2 < len(test) <= 4
-            and all(isinstance(t, int) for t in test)
-            and test[0] >= 0
-            and test[0] <= 16
-            and test[1] >= 0
-            and test[1] <= 16
-            and test[0] < test[1]
+                isinstance(test, list)
+                and 2 < len(test) <= 4
+                and all(isinstance(t, int) for t in test)
+                and test[0] >= 0
+                and test[0] <= 16
+                and test[1] >= 0
+                and test[1] <= 16
+                and test[0] < test[1]
         ):
             self.function = bit_test(*test) if len(test) == 3 else range_test(*test)
         else:
@@ -1193,7 +1199,8 @@ def evaluate(self, feature, notification, device, last_result):
         if gkeymap:
             current = gkeymap.get_modifier_state()
             if logger.isEnabledFor(logging.INFO):
-                logger.info("KeyPress action: %s %s, group %s, modifiers %s", self.key_names, self.action, kbdgroup(), current)
+                logger.info("KeyPress action: %s %s, group %s, modifiers %s", self.key_names, self.action, kbdgroup(),
+                            current)
             if self.action != RELEASE:
                 self.keyDown(self.key_symbols, current)
             if self.action != DEPRESS:
@@ -1262,7 +1269,8 @@ def __init__(self, args, warn=True):
             if count in [CLICK, DEPRESS, RELEASE]:
                 self.count = count
             elif warn:
-                logger.warning("rule MouseClick action: argument %s should be an integer or CLICK, PRESS, or RELEASE", count)
+                logger.warning("rule MouseClick action: argument %s should be an integer or CLICK, PRESS, or RELEASE",
+                               count)
                 self.count = 1
 
     def __str__(self):
@@ -1307,7 +1315,8 @@ def evaluate(self, feature, notification, device, last_result):
             return None
         args = setting.acceptable(self.args[2:], setting.read())
         if args is None:
-            logger.warning("Set Action: invalid args %s for setting %s of %s", self.args[2:], self.args[1], self.args[0])
+            logger.warning("Set Action: invalid args %s for setting %s of %s", self.args[2:], self.args[1],
+                           self.args[0])
             return None
         if len(args) > 1:
             setting.write_key_value(args[0], args[1])
@@ -1369,9 +1378,11 @@ def __str__(self):
     def evaluate(self, feature, notification, device, last_result):
         if self.delay and self.rule:
             if self.delay >= 1:
-                GLib.timeout_add_seconds(int(self.delay), Rule.once, self.rule, feature, notification, device, last_result)
+                GLib.timeout_add_seconds(int(self.delay), Rule.once, self.rule, feature, notification, device,
+                                         last_result)
             else:
-                GLib.timeout_add(int(self.delay * 1000), Rule.once, self.rule, feature, notification, device, last_result)
+                GLib.timeout_add(int(self.delay * 1000), Rule.once, self.rule, feature, notification, device,
+                                 last_result)
         return None
 
     def data(self):
@@ -1547,23 +1558,29 @@ def convert(elem):
     return True
 
 
-def _load_config_rule_file():
+def load_config_rule_file():
+    """Loads user configured rules."""
     global rules
-    loaded_rules = []
+
     if _path.isfile(_file_path):
-        try:
-            with open(_file_path) as config_file:
-                loaded_rules = []
-                for loaded_rule in _yaml_safe_load_all(config_file):
-                    rule = Rule(loaded_rule, source=_file_path)
-                    if logger.isEnabledFor(logging.DEBUG):
-                        logger.debug("load rule: %s", rule)
-                    loaded_rules.append(rule)
-                if logger.isEnabledFor(logging.INFO):
-                    logger.info("loaded %d rules from %s", len(loaded_rules), config_file.name)
-        except Exception as e:
-            logger.error("failed to load from %s\n%s", _file_path, e)
-    rules = Rule([Rule(loaded_rules, source=_file_path), built_in_rules])
+        rules = _load_rule_config(_file_path)
+
+
+def _load_rule_config(file_path: str) -> Rule:
+    loaded_rules = []
+    try:
+        with open(file_path) as config_file:
+            loaded_rules = []
+            for loaded_rule in _yaml_safe_load_all(config_file):
+                rule = Rule(loaded_rule, source=file_path)
+                if logger.isEnabledFor(logging.DEBUG):
+                    logger.debug("load rule: %s", rule)
+                loaded_rules.append(rule)
+            if logger.isEnabledFor(logging.INFO):
+                logger.info("loaded %d rules from %s", len(loaded_rules), config_file.name)
+    except Exception as e:
+        logger.error("failed to load from %s\n%s", file_path, e)
+    return Rule([Rule(loaded_rules, source=file_path), built_in_rules])
 
 
-_load_config_rule_file()
+load_config_rule_file()

lib/solaar/ui/diversion_rules.py

@@ -229,7 +229,7 @@ def _reload_yaml_file(self):
         self.dirty = False
         for c in self.selected_rule_edit_panel.get_children():
             self.selected_rule_edit_panel.remove(c)
-        _DIV._load_config_rule_file()
+        _DIV.load_config_rule_file()
         self.model = self._create_model()
         self.view.set_model(self.model)
         self.view.expand_all()

tests/logitech_receiver/test_diversion.py

@@ -0,0 +1,63 @@
+import textwrap
+
+from unittest import mock
+from unittest.mock import mock_open
+
+import pytest
+
+from logitech_receiver import diversion
+
+
+@pytest.fixture
+def rule_config():
+    rule_content = """
+    %YAML 1.3
+    ---
+    - MouseGesture: Mouse Left
+    - KeyPress:
+      - [Control_L, Alt_L, Left]
+      - click
+    ...
+    ---
+    - MouseGesture: Mouse Up
+    - KeyPress:
+      - [Super_L, Up]
+      - click
+    ...
+    ---
+    - Test: [thumb_wheel_up, 10]
+    - KeyPress:
+      - [Control_L, Page_Down]
+      - click
+    ...
+    ---
+    """
+    return textwrap.dedent(rule_content)
+
+
+def test_load_rule_config(rule_config):
+    expected_rules = [
+        [
+            diversion.MouseGesture,
+            diversion.KeyPress,
+        ],
+        [
+            diversion.MouseGesture,
+            diversion.KeyPress
+        ],
+        [
+            diversion.Test,
+            diversion.KeyPress
+        ],
+    ]
+
+    with mock.patch('builtins.open', new=mock_open(read_data=rule_config)):
+        loaded_rules = diversion._load_rule_config(file_path=mock.Mock())
+
+    assert len(loaded_rules.components) == 2  # predefined and user configured rules
+    user_configured_rules = loaded_rules.components[0]
+    assert isinstance(user_configured_rules, diversion.Rule)
+
+    for components, expected_components in zip(user_configured_rules.components, expected_rules):
+        for component, expected_component in zip(components.components, expected_components):
+            assert isinstance(component, expected_component)