path: root/python/vyos/xml_ref/definition.py

# Copyright 2024 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library.  If not, see <http://www.gnu.org/licenses/>.

from typing import Tuple, Optional, Union, Any, TYPE_CHECKING

# https://peps.python.org/pep-0484/#forward-references
# for type 'ConfigDict'
if TYPE_CHECKING:
    from vyos.config import ConfigDict

def set_source_recursive(o: Union[dict, str, list], b: bool):
    d = {}
    if not isinstance(o, dict):
        d = {'_source': b}
    else:
        for k, v in o.items():
            d[k] = set_source_recursive(v, b)
        d |= {'_source': b}
    return d

def source_dict_merge(src: dict, dest: dict):
    from copy import deepcopy
    dst = deepcopy(dest)
    from_src = {}

    for key, value in src.items():
        if key not in dst:
            dst[key] = value
            from_src[key] = set_source_recursive(value, True)
        elif isinstance(src[key], dict):
            dst[key], f = source_dict_merge(src[key], dst[key])
            f |= {'_source': False}
            from_src[key] = f

    return dst, from_src

def ext_dict_merge(src: dict, dest: Union[dict, 'ConfigDict']):
    d, f = source_dict_merge(src, dest)
    if hasattr(d, '_from_defaults'):
        setattr(d, '_from_defaults', f)
    return d

def from_source(d: dict, path: list) -> bool:
    for key in path:
        d  = d[key] if key in d else {}
        if not d or not isinstance(d, dict):
            return False
    return d.get('_source', False)

class Xml:
    def __init__(self):
        self.ref = {}

    def define(self, ref: dict):
        self.ref = ref

    def _get_ref_node_data(self, node: dict, data: str) -> Union[bool, str]:
        res = node.get('node_data', {})
        if not res:
            raise ValueError("non-existent node data")
        if data not in res:
            raise ValueError("non-existent data field")

        return res.get(data)

    def _get_ref_path(self, path: list) -> dict:
        ref_path = path.copy()
        d = self.ref
        while ref_path and d:
            d = d.get(ref_path[0], {})
            ref_path.pop(0)
            if self._is_tag_node(d) and ref_path:
                ref_path.pop(0)

        return d

    def _is_tag_node(self, node: dict) -> bool:
        res = self._get_ref_node_data(node, 'node_type')
        return res == 'tag'

    def exists(self, path: list) -> bool:
        try:
            _ = self._get_ref_path(path)
            return True
        except ValueError:
            return False

    def split_path(self, path: list) -> Tuple[list, Optional[str]]:
        """ Splits a list into config path and value components """

        # First, check if the complete path is valid by itself
        if self.exists(path):
            if self.is_valueless(path) or not self.is_leaf(path):
                # It's a complete path for a valueless node
                # or a path to an empy non-leaf node
                return (path, None)
            else:
                raise ValueError(f'Path "{path}" needs a value or children')
        else:
            # If the complete path doesn't exist, it's probably a path with a value
            if self.exists(path[0:-1]):
                return (path[0:-1], path[-1])
            else:
                # Or not a valid path at all
                raise ValueError(f'Path "{path}" is incorrect')

    def is_tag(self, path: list) -> bool:
        ref_path = path.copy()
        d = self.ref
        while ref_path and d:
            d = d.get(ref_path[0], {})
            ref_path.pop(0)
            if self._is_tag_node(d) and ref_path:
                if len(ref_path) == 1:
                    return False
                ref_path.pop(0)

        return self._is_tag_node(d)

    def is_tag_value(self, path: list) -> bool:
        if len(path) < 2:
            return False

        return self.is_tag(path[:-1])

    def _is_multi_node(self, node: dict) -> bool:
        b = self._get_ref_node_data(node, 'multi')
        assert isinstance(b, bool)
        return b

    def is_multi(self, path: list) -> bool:
        d = self._get_ref_path(path)
        return  self._is_multi_node(d)

    def _is_valueless_node(self, node: dict) -> bool:
        b = self._get_ref_node_data(node, 'valueless')
        assert isinstance(b, bool)
        return b

    def is_valueless(self, path: list) -> bool:
        d = self._get_ref_path(path)
        return  self._is_valueless_node(d)

    def _is_leaf_node(self, node: dict) -> bool:
        res = self._get_ref_node_data(node, 'node_type')
        return res == 'leaf'

    def is_leaf(self, path: list) -> bool:
        d = self._get_ref_path(path)
        return self._is_leaf_node(d)

    def _least_upper_data(self, path: list, name: str) -> str:
        ref_path = path.copy()
        d = self.ref
        data = ''
        tag = ''
        while ref_path and d:
            tag_val = ''
            d = d.get(ref_path[0], {})
            ref_path.pop(0)
            if self._is_tag_node(d) and ref_path:
                tag_val = ref_path[0]
                ref_path.pop(0)
            if self._is_leaf_node(d) and ref_path:
                ref_path.pop(0)
            res = self._get_ref_node_data(d, name)
            if res is not None:
                data = res
                tag = tag_val

        return data, tag

    def owner(self, path: list, with_tag=False) -> str:
        from pathlib import Path
        data, tag = self._least_upper_data(path, 'owner')
        tag_ext = f'_{tag}' if tag else ''
        if data:
            if with_tag:
                data = Path(data.split()[0]).stem
                data = f'{data}{tag_ext}'
            else:
                data = Path(data.split()[0]).name
        return data

    def priority(self, path: list) -> str:
        data, _ = self._least_upper_data(path, 'priority')
        return data

    @staticmethod
    def _dict_get(d: dict, path: list) -> dict:
        for i in path:
            d = d.get(i, {})
            if not isinstance(d, dict):
                return {}
            if not d:
                break
        return d

    def _dict_find(self, d: dict, key: str, non_local=False) -> bool:
        for k in list(d):
            if k in ('node_data', 'component_version'):
                continue
            if k == key:
                return True
            if non_local and isinstance(d[k], dict):
                if self._dict_find(d[k], key):
                    return True
        return False

    def cli_defined(self, path: list, node: str, non_local=False) -> bool:
        d = self._dict_get(self.ref, path)
        return self._dict_find(d, node, non_local=non_local)

    def component_version(self) -> dict:
        d = {}
        for k, v in self.ref['component_version'].items():
            d[k] = int(v)
        return d

    def multi_to_list(self, rpath: list, conf: dict) -> dict:
        res: Any = {}

        for k in list(conf):
            d = self._get_ref_path(rpath + [k])
            if self._is_leaf_node(d):
                if self._is_multi_node(d) and not isinstance(conf[k], list):
                    res[k] = [conf[k]]
                else:
                    res[k] = conf[k]
            else:
                res[k] = self.multi_to_list(rpath + [k], conf[k])

        return res

    def _get_default_value(self, node: dict) -> Optional[str]:
        return self._get_ref_node_data(node, "default_value")

    def _get_default(self, node: dict) -> Optional[Union[str, list]]:
        default = self._get_default_value(node)
        if default is None:
            return None
        if self._is_multi_node(node):
            return default.split()
        return default

    def default_value(self, path: list) -> Optional[Union[str, list]]:
        d = self._get_ref_path(path)
        default = self._get_default_value(d)
        if default is None:
            return None
        if self._is_multi_node(d) or self._is_tag_node(d):
            return default.split()
        return default

    def get_defaults(self, path: list, get_first_key=False, recursive=False) -> dict:
        """Return dict containing default values below path

        Note that descent below path will not proceed beyond an encountered
        tag node, as no tag node value is known. For a default dict relative
        to an existing config dict containing tag node values, see function:
        'relative_defaults'
        """
        res: dict = {}
        if self.is_tag(path):
            return res

        d = self._get_ref_path(path)

        if self._is_leaf_node(d):
            default_value = self._get_default(d)
            if default_value is not None:
                return {path[-1]: default_value} if path else {}

        for k in list(d):
            if k in ('node_data', 'component_version') :
                continue
            if self._is_leaf_node(d[k]):
                default_value = self._get_default(d[k])
                if default_value is not None:
                    res |= {k: default_value}
            elif self.is_tag(path + [k]):
                # tag node defaults are used as suggestion, not default value;
                # should this change, append to path and continue if recursive
                pass
            else:
                if recursive:
                    pos = self.get_defaults(path + [k], recursive=True)
                    res |= pos
        if res:
            if get_first_key or not path:
                return res
            return {path[-1]: res}

        return {}

    def _well_defined(self, path: list, conf: dict) -> bool:
        # test disjoint path + conf for sensible config paths
        def step(c):
            return [next(iter(c.keys()))] if c else []
        try:
            tmp = step(conf)
            if tmp and self.is_tag_value(path + tmp):
                c = conf[tmp[0]]
                if not isinstance(c, dict):
                    raise ValueError
                tmp = tmp + step(c)
                self._get_ref_path(path + tmp)
            else:
                self._get_ref_path(path + tmp)
        except ValueError:
            return False
        return True

    def _relative_defaults(self, rpath: list, conf: dict, recursive=False) -> dict:
        res: dict = {}
        res = self.get_defaults(rpath, recursive=recursive,
                                get_first_key=True)
        for k in list(conf):
            if isinstance(conf[k], dict):
                step = self._relative_defaults(rpath + [k], conf=conf[k],
                                               recursive=recursive)
                res |= step

        if res:
            return {rpath[-1]: res} if rpath else res

        return {}

    def relative_defaults(self, path: list, conf: dict, get_first_key=False,
                          recursive=False) -> dict:
        """Return dict containing defaults along paths of a config dict
        """
        if not conf:
            return self.get_defaults(path, get_first_key=get_first_key,
                                     recursive=recursive)
        if not self._well_defined(path, conf):
            # adjust for possible overlap:
            if path and path[-1] in list(conf):
                conf = conf[path[-1]]
                conf = {} if not isinstance(conf, dict) else conf
            if not self._well_defined(path, conf):
                print('path to config dict does not define full config paths')
                return {}

        res = self._relative_defaults(path, conf, recursive=recursive)

        if get_first_key and path:
            if res.values():
                res = next(iter(res.values()))
            else:
                res = {}

        return res