diff options
Diffstat (limited to 'src/onapsdk/utils')
| -rw-r--r-- | src/onapsdk/utils/__init__.py | 40 | ||||
| -rw-r--r-- | src/onapsdk/utils/configuration.py | 25 | ||||
| -rw-r--r-- | src/onapsdk/utils/gui.py | 35 | ||||
| -rw-r--r-- | src/onapsdk/utils/headers_creator.py | 245 | ||||
| -rw-r--r-- | src/onapsdk/utils/jinja.py | 50 | ||||
| -rw-r--r-- | src/onapsdk/utils/mixins.py | 99 | ||||
| -rw-r--r-- | src/onapsdk/utils/tosca_file_handler.py | 106 |
7 files changed, 600 insertions, 0 deletions
diff --git a/src/onapsdk/utils/__init__.py b/src/onapsdk/utils/__init__.py new file mode 100644 index 0000000..bd7f9f5 --- /dev/null +++ b/src/onapsdk/utils/__init__.py @@ -0,0 +1,40 @@ +"""ONAP SDK utils package.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. +import json +from datetime import datetime + + +def get_zulu_time_isoformat() -> str: + """Get zulu time in accepted by ONAP modules format. + + Returns: + str: Actual Zulu time. + + """ + return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.%fZ') + + +def load_json_file(path_to_json_file: str) -> str: + """ + Return json as string from selected file. + + Args: + path_to_json_file: (str) path to file with json + Returns: + File content as string (str) + """ + with open(path_to_json_file) as json_file: + data = json.load(json_file) + return json.dumps(data) diff --git a/src/onapsdk/utils/configuration.py b/src/onapsdk/utils/configuration.py new file mode 100644 index 0000000..89bee5a --- /dev/null +++ b/src/onapsdk/utils/configuration.py @@ -0,0 +1,25 @@ +"""Configuration package.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. +from typing import List + + +def tosca_path() -> str: + """Return tosca file paths.""" + return '/tmp/tosca_files/' + + +def components_needing_distribution() -> List[str]: + """Return the list of components needing distribution.""" + return ["SO", "sdnc", "aai"] diff --git a/src/onapsdk/utils/gui.py b/src/onapsdk/utils/gui.py new file mode 100644 index 0000000..421e966 --- /dev/null +++ b/src/onapsdk/utils/gui.py @@ -0,0 +1,35 @@ +"""Definition of GUI objects.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. +from dataclasses import dataclass +from typing import List + +@dataclass +class GuiItem: + """Class for keeping track of a GUI.""" + + url: str + status: int + +@dataclass +class GuiList: + """Class to list all the GUIs.""" + + guilist: List[GuiItem] + + def add(self, element): + """Add a GUi to GUI list.""" + if not isinstance(element, GuiItem): + raise AttributeError + self.guilist.append(element) diff --git a/src/onapsdk/utils/headers_creator.py b/src/onapsdk/utils/headers_creator.py new file mode 100644 index 0000000..adb0609 --- /dev/null +++ b/src/onapsdk/utils/headers_creator.py @@ -0,0 +1,245 @@ +"""Header creator package.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. +from typing import Dict +from uuid import uuid4 +import base64 +import hashlib + +from onapsdk.configuration import settings + + +def headers_sdc_creator(base_header: Dict[str, str], + user: str = "cs0008", + authorization: str = None): + """ + Create the right headers for SDC creator type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to cs0008 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_tester(base_header: Dict[str, str], + user: str = "jm0007", + authorization: str = None): + """ + Create the right headers for SDC tester type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to jm0007 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_governor(base_header: Dict[str, str], + user: str = "gv0001", + authorization: str = None): + """ + Create the right headers for SDC governor type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to gv0001 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_operator(base_header: Dict[str, str], + user: str = "op0001", + authorization: str = None): + """ + Create the right headers for SDC operator type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str, optional): the user to use. Default to op0001 + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + return headers_sdc_generic(base_header, user, authorization=authorization) + + +def headers_sdc_generic(base_header: Dict[str, str], + user: str, + authorization: str = None): + """ + Create the right headers for SDC generic type. + + Args: + base_header (Dict[str, str]): the base header to use + user (str): the user to use. + authorization (str, optional): the basic auth to use. + Default to "classic" one + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["USER_ID"] = user + headers["Authorization"] = authorization or settings.SDC_AUTH + headers["X-ECOMP-InstanceID"] = "onapsdk" + return headers + + +def headers_aai_creator(base_header: Dict[str, str]): + """ + Create the right headers for AAI creator type. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["x-fromappid"] = "AAI" + headers["x-transactionid"] = "0a3f6713-ba96-4971-a6f8-c2da85a3176e" + headers["authorization"] = settings.AAI_AUTH + return headers + + +def headers_so_creator(base_header: Dict[str, str]): + """ + Create the right headers for SO creator type. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["x-fromappid"] = "AAI" + headers["x-transactionid"] = str(uuid4()) + headers["authorization"] = settings.SO_AUTH + headers["cache-control"] = "no-cache" + return headers + +def headers_so_catelog_db_creator(base_header: Dict[str, str]): + """ + Create the right headers for SO creator type. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["x-fromappid"] = "AAI" + headers["x-transactionid"] = str(uuid4()) + headers["authorization"] = settings.SO_CAT_DB_AUTH + headers["cache-control"] = "no-cache" + return headers + +def headers_msb_creator(base_header: Dict[str, str]): + """ + Create the right headers for MSB. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["cache-control"] = "no-cache" + return headers + + +def headers_sdnc_creator(base_header: Dict[str, str]): + """ + Create the right headers for SDNC. + + Args: + base_header (Dict[str, str]): the base header to use + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["authorization"] = settings.SDNC_AUTH + headers["x-transactionid"] = str(uuid4()) + headers["x-fromappid"] = "API client" + return headers + + +def headers_sdc_artifact_upload(base_header: Dict[str, str], data: str): + """ + Create the right headers for sdc artifact upload. + + Args: + base_header (Dict[str, str]): the base header to use + data (str): payload data used to create an md5 content header + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["Accept"] = "application/json, text/plain, */*" + headers["Accept-Encoding"] = "gzip, deflate, br" + headers["Content-Type"] = "application/json; charset=UTF-8" + md5_content = hashlib.md5(data.encode('UTF-8')).hexdigest() + content = base64.b64encode(md5_content.encode('ascii')).decode('UTF-8') + headers["Content-MD5"] = content + return headers + +def headers_clamp_creator(base_header: Dict[str, str]): + """ + Create the right headers for CLAMP generic type. + + base_header (Dict[str, str]): the base header to use + data (str): payload data used to create an md5 content header + + Returns: + Dict[str, str]: the needed headers + + """ + headers = base_header.copy() + headers["Authorization"] = settings.CLAMP_AUTH + headers["X-ECOMP-InstanceID"] = "onapsdk" + return headers diff --git a/src/onapsdk/utils/jinja.py b/src/onapsdk/utils/jinja.py new file mode 100644 index 0000000..fe59eae --- /dev/null +++ b/src/onapsdk/utils/jinja.py @@ -0,0 +1,50 @@ +"""Jinja module.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. + +from jinja2 import Environment, PackageLoader, select_autoescape, ChoiceLoader + + +def jinja_env() -> Environment: + """Create Jinja environment. + + jinja_env allow to fetch simply jinja templates where they are. + by default jinja engine will look for templates in `templates` directory of + the package. So to load a template, you just have to do: + + Example: + >>> template = jinja_env().get_template('vendor_create.json.j2') + >>> data = template.render(name="vendor") + + See also: + SdcElement.create() for real use + + Returns: + Environment: the Jinja environment to use + + """ + return Environment(autoescape=select_autoescape(['html', 'htm', 'xml']), + loader=ChoiceLoader([ + PackageLoader("onapsdk.aai"), + PackageLoader("onapsdk.cds"), + PackageLoader("onapsdk.clamp"), + PackageLoader("onapsdk.msb"), + PackageLoader("onapsdk.nbi"), + PackageLoader("onapsdk.sdc"), + PackageLoader("onapsdk.sdnc"), + PackageLoader("onapsdk.sdnc"), + PackageLoader("onapsdk.so"), + PackageLoader("onapsdk.ves"), + PackageLoader("onapsdk.vid") + ])) diff --git a/src/onapsdk/utils/mixins.py b/src/onapsdk/utils/mixins.py new file mode 100644 index 0000000..7a64a15 --- /dev/null +++ b/src/onapsdk/utils/mixins.py @@ -0,0 +1,99 @@ +"""Mixins module.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. +from abc import ABC, abstractmethod +from ctypes import c_bool +from multiprocessing import Process, Value +from time import sleep + + +class WaitForFinishMixin(ABC): + """Wait for finish mixin. + + Mixin with wait_for_finish method and two properties: + - completed, + - finished. + + Can be used to wait for result of asynchronous tasks. + + """ + + WAIT_FOR_SLEEP_TIME = 10 + + @property + @abstractmethod + def completed(self) -> bool: + """Store an information if object task is completed or not. + + Returns: + bool: True if object task is completed, False otherwise. + + """ + + @property + @abstractmethod + def finished(self) -> bool: + """Store an information if object task is finished or not. + + Returns: + bool: True if object task is finished, False otherwise. + + """ + + def _wait_for_finish(self, return_value: Value) -> bool: + """Wait until object task is finished. + + Method called in another process. + + Args: + return_value(Value): value shared with main process to pass there + if object task was completed or not + + """ + while not self.finished: + sleep(self.WAIT_FOR_SLEEP_TIME) + self._logger.info(f"{self.__class__.__name__} task finished") + return_value.value = self.completed + + def wait_for_finish(self, timeout: float = None) -> bool: + """Wait until object task is finished. + + It uses time.sleep with WAIT_FOR_SLEEP_TIME value as a parameter to + wait unitl request is finished (object's finished property is + equal to True). + + It runs another process to control time of the function. If process timed out + TimeoutError is going to be raised. + + Args: + timeout(float, optional): positive number, wait at most timeout seconds + + Raises: + TimeoutError: Raised when function timed out + + Returns: + bool: True if object's task is successfully completed, False otherwise + + """ + self._logger.debug(f"Wait until {self.__class__.__name__} task is not finished") + return_value: Value = Value(c_bool) + wait_for_process: Process = Process(target=self._wait_for_finish, args=(return_value,)) + try: + wait_for_process.start() + wait_for_process.join(timeout) + return return_value.value + finally: + if wait_for_process.is_alive(): + wait_for_process.terminate() + raise TimeoutError diff --git a/src/onapsdk/utils/tosca_file_handler.py b/src/onapsdk/utils/tosca_file_handler.py new file mode 100644 index 0000000..921b868 --- /dev/null +++ b/src/onapsdk/utils/tosca_file_handler.py @@ -0,0 +1,106 @@ +"""Utils class.""" +# Copyright 2022 Orange, Deutsche Telekom AG +# +# 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. +import json +import string +import random +from typing import Dict, List + +from onapsdk.exceptions import ValidationError + +def get_parameter_from_yaml(parameter: str, config_file: str): + """Get the value of a given parameter in file.yaml. + + Parameter must be given in string format with dots + Example: general.openstack.image_name + + Args: + parameter (str): + config_file (str): configuration yaml file formtatted as string + + Raises: + ParameterError: parameter not defined + + Returns: + the value of the parameter + + """ + value = json.loads(config_file) + + # Workaround for the .. within the params in the yaml file + ugly_param = parameter.replace("..", "##") + for element in ugly_param.split("."): + value = value.get(element.replace("##", "..")) + if value is None: + msg = f"{element} in the {parameter} is not in YAML config file." + raise ValidationError(msg) + + return value + +def get_vf_list_from_tosca_file(model: str) -> List: + """Get the list of Vfs of a VNF based on the tosca file. + + Args: + model (str): the model retrieved from the tosca file at Vnf + instantiation + + Returns: + list: a list of Vfs + + """ + newlist = [] + node_list = get_parameter_from_yaml( + "topology_template.node_templates", model) + + for node in node_list: + value = get_parameter_from_yaml( + "topology_template.node_templates." + node + ".type", + model) + if "org.openecomp.resource.vf" in value: + print(node, value) + if node not in newlist: + search_value = str(node).split(" ")[0] + newlist.append(search_value) + return newlist + +def get_modules_list_from_tosca_file(model: str) -> Dict: + """Get the list of modules from tosca file. + + Modules are stored on topology_template.groups TOSCA file section. + + Args: + model (str): the model retrieved from the tosca file at Vnf + instantiation. + + Returns: + dict: a list of modules + + """ + return get_parameter_from_yaml( + "topology_template.groups", model + ) + +def random_string_generator(size=6, + chars=string.ascii_uppercase + string.digits) -> str: + """Get a random String for VNF. + + Args: + size (int): the number of alphanumerical chars for CI + chars (str): alphanumerical characters (ASCII uppercase and digits) + + Returns: + str: a sequence of random characters + + """ + return ''.join(random.choice(chars) for _ in range(size)) |