conf-mode: T5412: add support for supplemental dependency definitions

Add support for defining config-mode dependencies in add-on packages.

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)