Newer
Older
#pylint: disable=invalid-name, missing-function-docstring, line-too-long, logging-fstring-interpolation, missing-class-docstring, missing-module-docstring
# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
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
from typing import Any, Tuple, Optional, Dict
class InvalidIfnameError(Exception):
def __init__(self, ifname):
# Call the base class constructor with the parameters it needs
super().__init__(f"Invalid interface name {ifname}, expecting format \"MODULENAME|PORTNAME\" or \"MODULENAME|PORTNAME.VLAN\"")
def ifname_to_module_and_aid(ifname: str) -> Tuple[str, str]:
a = ifname.split("|")
if len(a) != 2:
raise InvalidIfnameError(ifname)
return (a[0], a[1])
def virtual_aid_to_aid_and_vlan(ifname: str) -> Tuple[str, Optional[str]]:
a = ifname.split(".")
if len(a) == 1:
return (ifname, None)
if len(a) != 2:
raise InvalidIfnameError(ifname)
return (a[0], a[1])
def ifname_to_module_aid_vlan(ifname: str) -> Tuple[str, str, Optional[str]]:
module, aid = ifname_to_module_and_aid(ifname)
aid, vlan = virtual_aid_to_aid_and_vlan(aid)
return (module, aid, vlan)
# For some reason when writing config, selector has moduleClientIfAid, when reading
# state it has clientIfAid...
def make_selector(mod, aid, _vlan) -> Dict[str, Any]:
selector = {
"ifSelectorByModuleName": {
"moduleName": mod,
"moduleClientIfAid": aid,
}
}
return selector
def get_constellation_module_ifnames(module):
ifnames = []
try:
module_state = module["state"]
module_name = module_state["module"]["moduleName"]
if "endpoints" in module_state:
for endpoint in module_state["endpoints"]:
try:
ifname = endpoint["moduleIf"]["clientIfAid"]
ifnames.append(f"{module_name}|{ifname}")
except KeyError:
pass
except KeyError:
pass
return ifnames
def set_optional_parameter(container: Dict[str, any], key:str, value: Optional[any]):
if value is not None:
container[key] = value