T8980: refactor functions out of config_mgmt.py

This is no real need to have these in config_mgmt; moving to config_sync
will localize needed modifications to local/remote configs for exclusion
mask.

1 files changed, 92 insertions, 12 deletions

diff --git a/src/op_mode/config_sync.py b/src/op_mode/config_sync.py
index d6eff7cd2..fb344767e 100644
--- a/src/op_mode/config_sync.py
+++ b/src/op_mode/config_sync.py
@@ -23,10 +23,11 @@ from vyos import http_api_client as http
 from vyos.utils.file import read_json
 from vyos.utils.dict import dict_to_paths
 from vyos.utils.list import list_contains_sublist
+from vyos.utils.config import get_saved_config_tree
 from vyos.configtree import ConfigTree
 from vyos.configtree import ConfigTreeError
-from vyos.config_mgmt import ConfigMgmt
-from vyos.config_mgmt import ConfigMgmtError
+from vyos.configtree import DiffTree
+from vyos.config import Config
 
 CONFIG_FILE = Path('/run/config_sync_conf.conf')
 
@@ -60,6 +61,7 @@ def _load_config_sync_sections() -> list:
 
 
 def _load_config_sync_settings() -> dict:
+    # pylint: disable=use-dict-literal
     """Load remote API settings from config-sync service runtime JSON file"""
 
     cfg = _read_json_config()
@@ -82,9 +84,12 @@ class ConfigSyncDiffManager:
         api_settings = _load_config_sync_settings()
         self._client = http.ApiClient(http.ApiClientConfig(**api_settings))
 
-        self._config_mgmt = ConfigMgmt()
+        self._config = Config()
+        self.running_config = self._config.get_config_tree(effective=True)
+        self.session_config = self._config.get_config_tree(effective=False)
 
     def _get_remote_config_tree(self, section_path: list = None) -> ConfigTree:
+        # pylint: disable=redefined-outer-name
         """
         Retrieve remote config (or subtree) as ConfigTree via HTTPS API.
 
@@ -116,6 +121,84 @@ class ConfigSyncDiffManager:
         except ConfigTreeError as e:
             raise opmode.InternalError(f'Unable to build remote ConfigTree: {e}') from e
 
+    def _format_remote_diff(self, diff_tree: DiffTree, path: list, commands: bool):
+        add_tree = diff_tree.add
+        del_tree = diff_tree.delete
+        command_prefix = ' '.join(path)
+
+        result_lines = []
+        if commands:
+            # Process the deleted elements into command format and filter based on prefix (path)
+            for line in del_tree.to_commands(op='delete').splitlines():
+                if line.startswith(f'delete {command_prefix}'):
+                    result_lines.append(line)
+
+            # Process the added elements into command format and filter based on prefix (path)
+            for line in add_tree.to_commands(op='set').splitlines():
+                if line.startswith(f'set {command_prefix}'):
+                    result_lines.append(line)
+        else:
+            with_node = len(path) > 1
+            # Retrieve subtrees for the specified path from both added and deleted trees
+            del_tree = del_tree.get_subtree(path, with_node=with_node)
+            add_tree = add_tree.get_subtree(path, with_node=with_node)
+
+            # Convert the subtrees to string lines for further processing
+            del_tree_lines = str(del_tree).splitlines()
+            add_tree_lines = str(add_tree).splitlines()
+
+            # Format the lines with a prefix ('-', '+') and filter out empty lines
+            del_lines = [f'- {l}' for l in del_tree_lines if l.strip()]  # noqa: E741
+            add_lines = [f'+ {l}' for l in add_tree_lines if l.strip()]  # noqa: E741
+
+            if del_lines or add_lines:
+                # Adjust command prefix if a node is present in the path
+                command_prefix = ' '.join(path[:-1]) if with_node else command_prefix
+                if command_prefix:
+                    result_lines.append(f'[{command_prefix}]')
+
+                # Combine both deleted and added lines and process them
+                result_lines.extend(del_lines + add_lines)
+
+        # Join the result lines into a single string, excluding empty lines
+        return '\n'.join((line for line in result_lines if line.strip()))
+
+    def remote_compare(
+        self,
+        source: str,
+        remote_tree: ConfigTree,
+        path: typing.Optional[list] = None,
+        commands: bool = False,
+    ) -> str:
+        # pylint: disable=redefined-outer-name
+        """
+        Compares a local configuration tree with a remote
+        configuration tree based on the specified source ('running', 'candidate', 'saved').
+        """
+        path = path or []
+
+        # Determine the correct local configuration tree based on the 'source' parameter
+        if source == 'running':
+            local_tree = self.running_config
+        elif source == 'candidate':
+            local_tree = self.session_config
+        elif source == 'saved':
+            try:
+                local_tree = get_saved_config_tree()
+            except ConfigTreeError as e:
+                raise opmode.InternalError(str(e)) from e
+        else:
+            raise opmode.IncorrectValue(
+                'Invalid source, must be one of: running, candidate, saved'
+            )
+
+        try:
+            diff_tree = DiffTree(remote_tree, local_tree)
+        except ConfigTreeError as e:
+            raise opmode.InternalError(str(e)) from e
+
+        return self._format_remote_diff(diff_tree, path, commands)
+
     def get_sync_diff(
         self,
         source: str,
@@ -127,15 +210,12 @@ class ConfigSyncDiffManager:
         results = []
         remote_tree = self._get_remote_config_tree()
         for section_path in sections:
-            try:
-                result = self._config_mgmt.remote_compare(
-                    source,
-                    remote_tree,
-                    path=section_path,
-                    commands=commands,
-                )
-            except ConfigMgmtError as e:
-                raise opmode.InternalError(str(e)) from e
+            result = self.remote_compare(
+                source,
+                remote_tree,
+                path=section_path,
+                commands=commands,
+            )
 
             result = result.strip()
             if result: