aboutsummaryrefslogtreecommitdiffstats
path: root/src/onapsdk/utils
diff options
context:
space:
mode:
Diffstat (limited to 'src/onapsdk/utils')
-rw-r--r--src/onapsdk/utils/__init__.py40
-rw-r--r--src/onapsdk/utils/configuration.py25
-rw-r--r--src/onapsdk/utils/gui.py35
-rw-r--r--src/onapsdk/utils/headers_creator.py245
-rw-r--r--src/onapsdk/utils/jinja.py50
-rw-r--r--src/onapsdk/utils/mixins.py99
-rw-r--r--src/onapsdk/utils/tosca_file_handler.py106
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))