[#114] Add yaml configuration controllers

Signed-off-by: Andrey Berezin <a.berezin@yadro.com>

src/frostfs_testlib/storage/configuration/interfaces.py

@@ -0,0 +1,65 @@
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class ServiceConfigurationYml(ABC):
+    """
+    Class to manipulate yml configuration for service
+    """
+
+    def _find_option(self, key: str, data: dict):
+        tree = key.split(":")
+        current = data
+        for node in tree:
+            if isinstance(current, list) and len(current) - 1 >= int(node):
+                current = current[int(node)]
+                continue
+
+            if node not in current:
+                return None
+
+            current = current[node]
+
+        return current
+
+    def _set_option(self, key: str, value: Any, data: dict):
+        tree = key.split(":")
+        current = data
+        for node in tree[:-1]:
+            if isinstance(current, list) and len(current) - 1 >= int(node):
+                current = current[int(node)]
+                continue
+
+            if node not in current:
+                current[node] = {}
+
+            current = current[node]
+
+        current[tree[-1]] = value
+
+    @abstractmethod
+    def get(self, key: str) -> str:
+        """
+        Get parameter value from current configuration
+
+        Args:
+            key: key of the parameter in yaml format like 'storage:shard:default:resync_metabase'
+
+        Returns:
+            value of the parameter
+        """
+
+    @abstractmethod
+    def set(self, values: dict[str, Any]):
+        """
+        Sets parameters to configuration
+
+        Args:
+            values: dict where key is the key of the parameter in yaml format like 'storage:shard:default:resync_metabase' and value is the value of the option to set
+        """
+
+    @abstractmethod
+    def revert(self):
+        """
+        Revert changes
+        """

src/frostfs_testlib/storage/configuration/service_configuration.py

@@ -0,0 +1,67 @@
+import os
+import re
+from typing import Any
+
+import yaml
+
+from frostfs_testlib.reporter import get_reporter
+from frostfs_testlib.shell.interfaces import CommandOptions
+from frostfs_testlib.storage.configuration.interfaces import ServiceConfigurationYml
+from frostfs_testlib.storage.dataclasses.node_base import ServiceClass
+
+reporter = get_reporter()
+
+
+class ServiceConfiguration(ServiceConfigurationYml):
+    def __init__(self, service: "ServiceClass") -> None:
+        self.service = service
+        self.shell = self.service.host.get_shell()
+        self.confd_path = os.path.join(self.service.config_dir, "conf.d")
+        self.custom_file = os.path.join(self.confd_path, "99_changes.yml")
+
+    def _path_exists(self, path: str) -> bool:
+        return not self.shell.exec(f"test -e {path}", options=CommandOptions(check=False)).return_code
+
+    def _get_data_from_file(self, path: str) -> dict:
+        content = self.shell.exec(f"cat {path}").stdout
+        data = yaml.safe_load(content)
+        return data
+
+    def get(self, key: str) -> str:
+        with reporter.step(f"Get {key} configuration value for {self.service}"):
+            config_files = [self.service.main_config_path]
+
+            if self._path_exists(self.confd_path):
+                files = self.shell.exec(f"find {self.confd_path} -type f").stdout.strip().split()
+                # Sorting files in backwards order from latest to first one
+                config_files.extend(sorted(files, key=lambda x: -int(re.findall("^\d+", os.path.basename(x))[0])))
+
+            result = None
+            for file in files:
+                data = self._get_data_from_file(file)
+                result = self._find_option(key, data)
+                if result is not None:
+                    break
+
+        return result
+
+    def set(self, values: dict[str, Any]):
+        with reporter.step(f"Change configuration for {self.service}"):
+            if not self._path_exists(self.confd_path):
+                self.shell.exec(f"mkdir {self.confd_path}")
+
+            if self._path_exists(self.custom_file):
+                data = self._get_data_from_file(self.custom_file)
+            else:
+                data = {}
+
+            for key, value in values.items():
+                self._set_option(key, value, data)
+
+            content = yaml.dump(data)
+            self.shell.exec(f"echo '{content}' | sudo tee {self.custom_file}")
+            self.shell.exec(f"chmod 777 {self.custom_file}")
+
+    def revert(self):
+        with reporter.step(f"Revert changed options for {self.service}"):
+            self.shell.exec(f"rm -rf {self.custom_file}")