Merge d77838282c into 783bd5e4da

Introduce a higher level storage provider interface to decouple storage
implementation from diversion dialog and use composition to pass an
implementation.

This simplifies testing of the diversion dialog and enables clean unit
testing without I/O.

Related #2675

lib/logitech_receiver/diversion.py

@@ -29,16 +29,18 @@ import sys
 import time
 import typing
 
+from pathlib import Path
 from typing import Any
 from typing import Dict
 from typing import Tuple
 
 import gi
 import psutil
-import yaml
 
 from keysyms import keysymdef
 
+from . import rule_storage
+
 # There is no evdev on macOS or Windows. Diversion will not work without
 # it but other Solaar functionality is available.
 if platform.system() in ("Darwin", "Windows"):
@@ -58,6 +60,14 @@ if typing.TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+if os.environ.get("XDG_CONFIG_HOME"):
+    xdg_config_home = Path(os.environ.get("XDG_CONFIG_HOME"))
+else:
+    xdg_config_home = Path("~/.config").expanduser()
+
+RULES_CONFIG = xdg_config_home / "solaar" / "rules.yaml"
+
+
 #
 # See docs/rules.md for documentation
 #
@@ -146,6 +156,17 @@ thumb_wheel_displacement = 0
 _dbus_interface = None
 
 
+class AbstractRepository(typing.Protocol):
+    def save(self, rules: Dict[str, str]) -> None:
+        ...
+
+    def load(self) -> list:
+        ...
+
+
+storage: AbstractRepository = rule_storage.YmlRuleStorage(RULES_CONFIG)
+
+
 class XkbDisplay(ctypes.Structure):
     """opaque struct"""
 
@@ -1553,83 +1574,64 @@ def process_notification(device, notification: HIDPPNotification, feature) -> No
     GLib.idle_add(evaluate_rules, feature, notification, device)
 
 
-_XDG_CONFIG_HOME = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser(os.path.join("~", ".config"))
-_file_path = os.path.join(_XDG_CONFIG_HOME, "solaar", "rules.yaml")
+class Persister:
+    @staticmethod
+    def save_config_rule_file() -> None:
+        """Saves user configured rules."""
 
-rules = built_in_rules
+        # This is a trick to show str/float/int lists in-line (inspired by https://stackoverflow.com/a/14001707)
+        class inline_list(list):
+            pass
 
+        def convert(elem):
+            if isinstance(elem, list):
+                if len(elem) == 1 and isinstance(elem[0], (int, str, float)):
+                    # All diversion classes that expect a list of scalars also support a single scalar without a list
+                    return elem[0]
+                if all(isinstance(c, (int, str, float)) for c in elem):
+                    return inline_list([convert(c) for c in elem])
+                return [convert(c) for c in elem]
+            if isinstance(elem, dict):
+                return {k: convert(v) for k, v in elem.items()}
+            if isinstance(elem, NamedInt):
+                return int(elem)
+            return elem
 
-def _save_config_rule_file(file_name: str = _file_path):
-    # This is a trick to show str/float/int lists in-line (inspired by https://stackoverflow.com/a/14001707)
-    class inline_list(list):
-        pass
+        global rules
 
-    def blockseq_rep(dumper, data):
-        return dumper.represent_sequence("tag:yaml.org,2002:seq", data, flow_style=True)
+        # Save only user-defined rules
+        rules_to_save = sum((r.data()["Rule"] for r in rules.components if r.source == str(RULES_CONFIG)), [])
+        if logger.isEnabledFor(logging.INFO):
+            logger.info(f"saving {len(rules_to_save)} rule(s) to {str(RULES_CONFIG)}")
+        dump_data = [r["Rule"] for r in rules_to_save]
+        try:
+            data = convert(dump_data)
+            storage.save(data)
+        except Exception:
+            logger.error("failed to save to rules config")
 
-    yaml.add_representer(inline_list, blockseq_rep)
+    @staticmethod
+    def load_rule_config() -> Rule:
+        """Loads user configured rules."""
+        global rules
 
-    def convert(elem):
-        if isinstance(elem, list):
-            if len(elem) == 1 and isinstance(elem[0], (int, str, float)):
-                # All diversion classes that expect a list of scalars also support a single scalar without a list
-                return elem[0]
-            if all(isinstance(c, (int, str, float)) for c in elem):
-                return inline_list([convert(c) for c in elem])
-            return [convert(c) for c in elem]
-        if isinstance(elem, dict):
-            return {k: convert(v) for k, v in elem.items()}
-        if isinstance(elem, NamedInt):
-            return int(elem)
-        return elem
-
-    # YAML format settings
-    dump_settings = {
-        "encoding": "utf-8",
-        "explicit_start": True,
-        "explicit_end": True,
-        "default_flow_style": False,
-        # 'version': (1, 3),  # it would be printed for every rule
-    }
-    # Save only user-defined rules
-    rules_to_save = sum((r.data()["Rule"] for r in rules.components if r.source == file_name), [])
-    if logger.isEnabledFor(logging.INFO):
-        logger.info("saving %d rule(s) to %s", len(rules_to_save), file_name)
-    try:
-        with open(file_name, "w") as f:
-            if rules_to_save:
-                f.write("%YAML 1.3\n")  # Write version manually
-            dump_data = [r["Rule"] for r in rules_to_save]
-            yaml.dump_all(convert(dump_data), f, **dump_settings)
-    except Exception as e:
-        logger.error("failed to save to %s\n%s", file_name, e)
-        return False
-    return True
-
-
-def load_config_rule_file():
-    """Loads user configured rules."""
-    global rules
-
-    if os.path.isfile(_file_path):
-        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)
+        loaded_rules = []
+        try:
+            plain_rules = storage.load()
+            for loaded_rule in plain_rules:
+                rule = Rule(loaded_rule, source=str(RULES_CONFIG))
                 if logger.isEnabledFor(logging.DEBUG):
-                    logger.debug("load rule: %s", rule)
+                    logger.debug(f"load rule: {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])
+                logger.info(
+                    f"loaded {len(loaded_rules)} rules from config file",
+                )
+        except Exception as e:
+            logger.error(f"failed to load from {RULES_CONFIG}\n{e}")
+        user_rules = Rule(loaded_rules, source=str(RULES_CONFIG))
+        rules = Rule([user_rules, built_in_rules])
+        return rules
 
 
-load_config_rule_file()
+Persister.load_rule_config()

lib/logitech_receiver/hidpp20.py

@@ -296,7 +296,10 @@ class ReprogrammableKeyV4(ReprogrammableKey):
         if self._mapped_to is None:
             self._getCidReporting()
         self._device.keys._ensure_all_keys_queried()
-        task = str(special_keys.Task(self._device.keys.cid_to_tid[self._mapped_to]))
+        try:
+            task = str(special_keys.Task(self._device.keys.cid_to_tid[self._mapped_to]))
+        except ValueError:
+            task = f"Unknown_{self._mapped_to:x}"
         return NamedInt(self._mapped_to, task)
 
     @property

lib/logitech_receiver/rule_storage.py

@@ -0,0 +1,47 @@
+from pathlib import Path
+from typing import Dict
+
+import yaml
+
+
+class YmlRuleStorage:
+    def __init__(self, path: Path):
+        self._config_path = path
+
+    def save(self, rules: Dict[str, str]) -> None:
+        # This is a trick to show str/float/int lists in-line (inspired by https://stackoverflow.com/a/14001707)
+        class inline_list(list):
+            pass
+
+        def blockseq_rep(dumper, data):
+            return dumper.represent_sequence("tag:yaml.org,2002:seq", data, flow_style=True)
+
+        yaml.add_representer(inline_list, blockseq_rep)
+        format_settings = {
+            "encoding": "utf-8",
+            "explicit_start": True,
+            "explicit_end": True,
+            "default_flow_style": False,
+        }
+        with open(self._config_path, "w") as f:
+            f.write("%YAML 1.3\n")  # Write version manually
+            yaml.dump_all(rules, f, **format_settings)
+
+    def load(self) -> list:
+        with open(self._config_path) as config_file:
+            plain_rules = list(yaml.safe_load_all(config_file))
+        return plain_rules
+
+
+class FakeRuleStorage:
+    def __init__(self, rules=None):
+        if rules is None:
+            self._rules = {}
+        else:
+            self._rules = rules
+
+    def save(self, rules: dict) -> None:
+        self._rules = rules
+
+    def load(self) -> dict:
+        return self._rules

lib/solaar/ui/diversion_rules.py

@@ -31,6 +31,7 @@ from typing import Any
 from typing import Callable
 from typing import Dict
 from typing import Optional
+from typing import Protocol
 
 from gi.repository import Gdk
 from gi.repository import GObject
@@ -359,7 +360,7 @@ class ActionMenu:
         else:
             idx = parent_c.components.index(c)
         if isinstance(new_c, diversion.Rule) and wrapped.level == 1:
-            new_c.source = diversion._file_path  # new rules will be saved to the YAML file
+            new_c.source = str(diversion.RULES_CONFIG)  # new rules will be saved to the YAML file
         idx += int(below)
         parent_c.components.insert(idx, new_c)
         self._populate_model_func(m, parent_it, new_c, level=wrapped.level, pos=idx)
@@ -562,8 +563,14 @@ class ActionMenu:
         return menu_copy
 
 
+class RulePersister(Protocol):
+    def load_rule_config(self) -> _DIV.Rule: ...
+
+    def save_config_rule_file(self) -> None: ...
+
+
 class DiversionDialog:
-    def __init__(self, action_menu):
+    def __init__(self, action_menu, rule_persister: RulePersister):
         window = Gtk.Window()
         window.set_title(_("Solaar Rule Editor"))
         window.connect(GtkSignal.DELETE_EVENT.value, self._closing)
@@ -580,6 +587,7 @@ class DiversionDialog:
             populate_model_func=_populate_model,
             on_update=self.on_update,
         )
+        self._ruler_persister = rule_persister
 
         self.dirty = False  # if dirty, there are pending changes to be saved
 
@@ -638,16 +646,20 @@ class DiversionDialog:
         self.dirty = False
         for c in self.selected_rule_edit_panel.get_children():
             self.selected_rule_edit_panel.remove(c)
+        self._ruler_persister.load_rule_config()
         diversion.load_config_rule_file()
         self.model = self._create_model()
         self.view.set_model(self.model)
         self.view.expand_all()
 
     def _save_yaml_file(self):
-        if diversion._save_config_rule_file():
+        try:
+            self._ruler_persister.save_config_rule_file()
             self.dirty = False
             self.save_btn.set_sensitive(False)
             self.discard_btn.set_sensitive(False)
+        except Exception:
+            pass
 
     def _create_top_panel(self):
         sw = Gtk.ScrolledWindow()
@@ -1879,6 +1891,6 @@ def show_window(model: Gtk.TreeStore):
     global _dev_model
     _dev_model = model
     if _diversion_dialog is None:
-        _diversion_dialog = DiversionDialog(ActionMenu)
+        _diversion_dialog = DiversionDialog(action_menu=ActionMenu, rule_persister=diversion.Persister())
     update_devices()
     _diversion_dialog.window.present()

tests/logitech_receiver/test_diversion.py

@@ -48,7 +48,7 @@ def test_load_rule_config(rule_config):
     ]
 
     with mock.patch("builtins.open", new=mock_open(read_data=rule_config)):
-        loaded_rules = diversion._load_rule_config(file_path=mock.Mock())
+        loaded_rules = diversion.Persister.load_rule_config()
 
     assert len(loaded_rules.components) == 2  # predefined and user configured rules
     user_configured_rules = loaded_rules.components[0]