diff --git a/python/vyos/configdep.py b/python/vyos/configdep.py
index e0fe1ddac..cf7c9d543 100644
--- a/python/vyos/configdep.py
+++ b/python/vyos/configdep.py
@@ -1,191 +1,211 @@
# Copyright 2023-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/>.
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
# https://peps.python.org/pep-0484/#forward-references
# for type 'Config'
if typing.TYPE_CHECKING:
from vyos.config import Config
dependency_dir = os.path.join(directories['data'],
'config-mode-dependencies')
dependency_list: list[typing.Callable] = []
DEBUG = False
def debug_print(s: str):
if DEBUG:
print(s)
def canon_name(name: str) -> str:
return os.path.splitext(name)[0].replace('-', '_')
def canon_name_of_path(path: str) -> str:
script = os.path.basename(path)
return canon_name(script)
def caller_name() -> str:
filename = stack()[2].filename
return canon_name_of_path(filename)
def name_of(f: typing.Callable) -> str:
return f.__name__
def names_of(l: list[typing.Callable]) -> list[str]:
return [name_of(f) for f in l]
def remove_redundant(l: list[typing.Callable]) -> list[typing.Callable]:
names = set()
for e in reversed(l):
_ = l.remove(e) if name_of(e) in names else names.add(name_of(e))
def append_uniq(l: list[typing.Callable], e: typing.Callable):
"""Append an element, removing earlier occurrences
The list of dependencies is generally short and traversing the list on
each append is preferable to the cost of redundant script invocation.
"""
l.append(e)
remove_redundant(l)
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'):
d = getattr(config, 'cached_dependency_dict')
else:
d = read_dependency_dict()
setattr(config, 'cached_dependency_dict', d)
return d
-def run_config_mode_script(script: str, config: 'Config'):
+def run_config_mode_script(target: str, config: 'Config'):
+ script = target + '.py'
path = os.path.join(directories['conf_mode'], script)
name = canon_name(script)
mod = load_as_module(name, path)
config.set_level([])
try:
c = mod.get_config(config)
mod.verify(c)
mod.generate(c)
mod.apply(c)
except (VyOSError, ConfigError) as e:
raise ConfigError(str(e)) from e
+def run_conditionally(target: str, tagnode: str, config: 'Config'):
+ tag_ext = f'_{tagnode}' if tagnode else ''
+ script_name = f'{target}{tag_ext}'
+
+ scripts_called = getattr(config, 'scripts_called', [])
+ commit_scripts = getattr(config, 'commit_scripts', [])
+
+ debug_print(f'scripts_called: {scripts_called}')
+ debug_print(f'commit_scripts: {commit_scripts}')
+
+ if script_name in commit_scripts and script_name not in scripts_called:
+ debug_print(f'dependency {script_name} deferred to priority')
+ return
+
+ run_config_mode_script(target, config)
+
def def_closure(target: str, config: 'Config',
tagnode: typing.Optional[str] = None) -> typing.Callable:
- script = target + '.py'
def func_impl():
+ tag_value = ''
if tagnode is not None:
os.environ['VYOS_TAGNODE_VALUE'] = tagnode
- run_config_mode_script(script, config)
+ tag_value = tagnode
+ run_conditionally(target, tag_value, config)
+
tag_ext = f'_{tagnode}' if tagnode is not None else ''
func_impl.__name__ = f'{target}{tag_ext}'
+
return func_impl
def set_dependents(case: str, config: 'Config',
tagnode: typing.Optional[str] = None):
global dependency_list
dependency_list = config.dependency_list
d = get_dependency_dict(config)
k = caller_name()
l = dependency_list
for target in d[k][case]:
func = def_closure(target, config, tagnode)
append_uniq(l, func)
debug_print(f'set_dependents: caller {k}, current dependents {names_of(l)}')
def call_dependents():
k = caller_name()
l = dependency_list
debug_print(f'call_dependents: caller {k}, remaining dependents {names_of(l)}')
while l:
f = l.pop(0)
debug_print(f'calling: {f.__name__}')
try:
f()
except ConfigError as e:
s = f'dependent {f.__name__}: {str(e)}'
raise ConfigError(s) from e
def called_as_dependent() -> bool:
st = stack()[1:]
for f in st:
if f.filename == __file__:
return True
return False
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)