1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
# 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'):
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 def_closure(target: str, config: 'Config',
tagnode: typing.Optional[str] = None) -> typing.Callable:
script = target + '.py'
def func_impl():
if tagnode is not None:
os.environ['VYOS_TAGNODE_VALUE'] = tagnode
run_config_mode_script(script, 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)
|
