summaryrefslogtreecommitdiff
path: root/python
diff options
context:
space:
mode:
Diffstat (limited to 'python')
-rw-r--r--python/vyos/configdep.py61
1 files changed, 54 insertions, 7 deletions
diff --git a/python/vyos/configdep.py b/python/vyos/configdep.py
index 7a8559839..05d9a3fa3 100644
--- a/python/vyos/configdep.py
+++ b/python/vyos/configdep.py
@@ -1,4 +1,4 @@
-# Copyright 2022 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2023 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
@@ -17,8 +17,10 @@ import os
import json
import typing
from inspect import stack
+from graphlib import TopologicalSorter, CycleError
from vyos.utils.system import load_as_module
+from vyos.configdict import dict_merge
from vyos.defaults import directories
from vyos.configsource import VyOSError
from vyos import ConfigError
@@ -28,6 +30,9 @@ from vyos import ConfigError
if typing.TYPE_CHECKING:
from vyos.config import Config
+dependency_dir = os.path.join(directories['data'],
+ 'config-mode-dependencies')
+
dependent_func: dict[str, list[typing.Callable]] = {}
def canon_name(name: str) -> str:
@@ -40,12 +45,20 @@ def canon_name_of_path(path: str) -> str:
def caller_name() -> str:
return stack()[-1].filename
-def read_dependency_dict() -> dict:
- path = os.path.join(directories['data'],
- 'config-mode-dependencies.json')
- with open(path) as f:
- d = json.load(f)
- return d
+def read_dependency_dict(dependency_dir: str = dependency_dir) -> dict:
+ res = {}
+ for dep_file in os.listdir(dependency_dir):
+ if not dep_file.endswith('.json'):
+ continue
+ path = os.path.join(dependency_dir, dep_file)
+ with open(path) as f:
+ d = json.load(f)
+ if dep_file == 'vyos-1x.json':
+ res = dict_merge(res, d)
+ else:
+ res = dict_merge(d, res)
+
+ return res
def get_dependency_dict(config: 'Config') -> dict:
if hasattr(config, 'cached_dependency_dict'):
@@ -93,3 +106,37 @@ def call_dependents():
while l:
f = l.pop(0)
f()
+
+def graph_from_dependency_dict(d: dict) -> dict:
+ g = {}
+ for k in list(d):
+ g[k] = set()
+ # add the dependencies for every sub-case; should there be cases
+ # that are mutally exclusive in the future, the graphs will be
+ # distinguished
+ for el in list(d[k]):
+ g[k] |= set(d[k][el])
+
+ return g
+
+def is_acyclic(d: dict) -> bool:
+ g = graph_from_dependency_dict(d)
+ ts = TopologicalSorter(g)
+ try:
+ # get node iterator
+ order = ts.static_order()
+ # try iteration
+ _ = [*order]
+ except CycleError:
+ return False
+
+ return True
+
+def check_dependency_graph(dependency_dir: str = dependency_dir,
+ supplement: str = None) -> bool:
+ d = read_dependency_dict(dependency_dir=dependency_dir)
+ if supplement is not None:
+ with open(supplement) as f:
+ d = dict_merge(json.load(f), d)
+
+ return is_acyclic(d)