From 062d866951943e588342be10ed28c2dac6faa2ab Mon Sep 17 00:00:00 2001 From: MattHag <16444067+MattHag@users.noreply.github.com> Date: Mon, 11 Nov 2024 00:00:19 +0100 Subject: [PATCH 1/2] rule storage: Apply AbstractRepository pattern Introduce a AbstractRepository as interface to the rules database and implement a YML storage that implements it. Additionally, a FakeStorage implements that interface to test diversion related code, without I/O calls and side effects. It keeps the rules in-memory for the test duration. Related #2675 --- lib/logitech_receiver/diversion.py | 98 +++++++++++------------ lib/logitech_receiver/rule_storage.py | 47 +++++++++++ lib/solaar/ui/diversion_rules.py | 7 +- tests/logitech_receiver/test_diversion.py | 2 +- 4 files changed, 102 insertions(+), 52 deletions(-) create mode 100644 lib/logitech_receiver/rule_storage.py diff --git a/lib/logitech_receiver/diversion.py b/lib/logitech_receiver/diversion.py index 5ac595dc..e764813d 100644 --- a/lib/logitech_receiver/diversion.py +++ b/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""" @@ -1548,22 +1569,13 @@ 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") +def save_config_rule_file() -> None: + """Saves user configured rules.""" -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)): @@ -1578,53 +1590,41 @@ def _save_config_rule_file(file_name: str = _file_path): 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 - } + global rules + # Save only user-defined rules - rules_to_save = sum((r.data()["Rule"] for r in rules.components if r.source == file_name), []) + 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("saving %d rule(s) to %s", len(rules_to_save), file_name) + 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: - 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 + data = convert(dump_data) + storage.save(data) + except Exception: + logger.error("failed to save to rules config") -def load_config_rule_file(): +def load_rule_config() -> Rule: """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) - 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) + 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(f"load rule: {rule}") + loaded_rules.append(rule) + if logger.isEnabledFor(logging.INFO): + logger.info( + f"loaded {len(loaded_rules)} rules from config file", + ) 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.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() +load_rule_config() diff --git a/lib/logitech_receiver/rule_storage.py b/lib/logitech_receiver/rule_storage.py new file mode 100644 index 00000000..1c38e4c5 --- /dev/null +++ b/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 diff --git a/lib/solaar/ui/diversion_rules.py b/lib/solaar/ui/diversion_rules.py index 788eab24..17f8d02f 100644 --- a/lib/solaar/ui/diversion_rules.py +++ b/lib/solaar/ui/diversion_rules.py @@ -347,7 +347,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) @@ -632,10 +632,13 @@ class DiversionDialog: self.view.expand_all() def _save_yaml_file(self): - if diversion._save_config_rule_file(): + try: + diversion.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() diff --git a/tests/logitech_receiver/test_diversion.py b/tests/logitech_receiver/test_diversion.py index 70aba163..5218a240 100644 --- a/tests/logitech_receiver/test_diversion.py +++ b/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.load_rule_config() assert len(loaded_rules.components) == 2 # predefined and user configured rules user_configured_rules = loaded_rules.components[0] From d77838282cbaabc0f9dc1390208afb60ed7f79eb Mon Sep 17 00:00:00 2001 From: MattHag <16444067+MattHag@users.noreply.github.com> Date: Tue, 12 Nov 2024 02:20:18 +0100 Subject: [PATCH 2/2] rule storage: Use composition for diversion dialog 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 | 106 +++++++++++----------- lib/solaar/ui/diversion_rules.py | 15 ++- tests/logitech_receiver/test_diversion.py | 2 +- 3 files changed, 67 insertions(+), 56 deletions(-) diff --git a/lib/logitech_receiver/diversion.py b/lib/logitech_receiver/diversion.py index e764813d..9c3ab8cc 100644 --- a/lib/logitech_receiver/diversion.py +++ b/lib/logitech_receiver/diversion.py @@ -1569,62 +1569,64 @@ def process_notification(device, notification: HIDPPNotification, feature) -> No GLib.idle_add(evaluate_rules, feature, notification, device) -def save_config_rule_file() -> None: - """Saves user configured rules.""" +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 + # 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 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 + 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") - - -def load_rule_config() -> Rule: - """Loads user configured rules.""" - global rules - - 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(f"load rule: {rule}") - loaded_rules.append(rule) + # 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"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 + 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 + + 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(f"load rule: {rule}") + loaded_rules.append(rule) + if logger.isEnabledFor(logging.INFO): + 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_rule_config() +Persister.load_rule_config() diff --git a/lib/solaar/ui/diversion_rules.py b/lib/solaar/ui/diversion_rules.py index 17f8d02f..73689824 100644 --- a/lib/solaar/ui/diversion_rules.py +++ b/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 @@ -550,8 +551,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("delete-event", self._closing) @@ -568,6 +575,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 @@ -626,6 +634,7 @@ 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) @@ -633,7 +642,7 @@ class DiversionDialog: def _save_yaml_file(self): try: - diversion.save_config_rule_file() + self._ruler_persister.save_config_rule_file() self.dirty = False self.save_btn.set_sensitive(False) self.discard_btn.set_sensitive(False) @@ -1867,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() diff --git a/tests/logitech_receiver/test_diversion.py b/tests/logitech_receiver/test_diversion.py index 5218a240..9e8ae4b2 100644 --- a/tests/logitech_receiver/test_diversion.py +++ b/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() + 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]