Rule storage with AbstractRepository pattern

lib/logitech_receiver/diversion.py

@@ -29,16 +29,18 @@
 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 @@
 
 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 @@
 _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"""
 
@@ -1548,83 +1569,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")
-
-rules = built_in_rules
-
-
-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
-
-    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)
-
-    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)
+class Persister:
+    @staticmethod
+    def save_config_rule_file() -> None:
+        """Saves user configured 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
+
+        global rules
+
+        # 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")
 
+    @staticmethod
+    def load_rule_config() -> Rule:
+        """Loads user configured rules."""
+        global rules
 
-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/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 Callable
 from typing import Dict
 from typing import Optional
+from typing import Protocol
 
 from gi.repository import Gdk
 from gi.repository import GObject
@@ -347,7 +348,7 @@ def _menu_do_insert(self, _mitem, m, it, new_c, below=False):
         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)
@@ -550,8 +551,14 @@ def _menu_copy(self, m, it):
         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("delete-event", self._closing)
@@ -568,6 +575,7 @@ def __init__(self, action_menu):
             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
 
@@ -626,16 +634,20 @@ 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)
+        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()
@@ -1864,6 +1876,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]