Release ONAP SDKv10.2

Issue-ID: INT-2150
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: I650047c599a5aae6de7c6b42d38e34aea88578e2

4 files changed, 736 insertions, 0 deletions

diff --git a/src/onapsdk/msb/k8s/__init__.py b/src/onapsdk/msb/k8s/__init__.py
new file mode 100644
index 0000000..655502d
--- /dev/null
+++ b/src/onapsdk/msb/k8s/__init__.py
@@ -0,0 +1,17 @@
+"""K8s 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 .definition import Definition, Profile, ConfigurationTemplate
+from .connectivity_info import ConnectivityInfo
+from .instance import InstantiationParameter, InstantiationRequest, Instance
diff --git a/src/onapsdk/msb/k8s/connectivity_info.py b/src/onapsdk/msb/k8s/connectivity_info.py
new file mode 100644
index 0000000..71a43c1
--- /dev/null
+++ b/src/onapsdk/msb/k8s/connectivity_info.py
@@ -0,0 +1,105 @@
+"""Connectivity-Info 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 onapsdk.utils.jinja import jinja_env
+from ..msb_service import MSB
+
+
+class ConnectivityInfo(MSB):
+    """Connectivity-Info class."""
+
+    api_version = "/api/multicloud-k8s/v1/v1"
+    url = f"{MSB.base_url}{api_version}/connectivity-info"
+
+    def __init__(self, cloud_region_id: str,
+                 cloud_owner: str,
+                 other_connectivity_list: dict,
+                 kubeconfig: str) -> None:
+        """Connectivity-info object initialization.
+
+        Args:
+            cloud_region_id (str): Cloud region ID
+            cloud_owner (str): Cloud owner name
+            other_connectivity_list (dict): Optional other connectivity list
+            kubeconfig (str): kubernetes cluster kubeconfig
+        """
+        super().__init__()
+        self.cloud_region_id: str = cloud_region_id
+        self.cloud_owner: str = cloud_owner
+        self.other_connectivity_list: dict = other_connectivity_list
+        self.kubeconfig: str = kubeconfig
+
+    @classmethod
+    def get_connectivity_info_by_region_id(cls, cloud_region_id: str) -> "ConnectivityInfo":
+        """Get connectivity-info by its name (cloud region id).
+
+        Args:
+            cloud_region_id (str): Cloud region ID
+
+        Returns:
+            ConnectivityInfo: Connectivity-Info object
+
+        """
+        url: str = f"{cls.url}/{cloud_region_id}"
+        connectivity_info: dict = cls.send_message_json(
+            "GET",
+            "Get Connectivity Info",
+            url
+        )
+        return cls(
+            connectivity_info["cloud-region"],
+            connectivity_info["cloud-owner"],
+            connectivity_info.get("other-connectivity-list"),
+            connectivity_info["kubeconfig"]
+        )
+
+    def delete(self) -> None:
+        """Delete connectivity info."""
+        url: str = f"{self.url}/{self.cloud_region_id}"
+        self.send_message(
+            "DELETE",
+            "Delete Connectivity Info",
+            url
+        )
+
+    @classmethod
+    def create(cls,
+               cloud_region_id: str,
+               cloud_owner: str,
+               kubeconfig: bytes = None) -> "ConnectivityInfo":
+        """Create Connectivity Info.
+
+        Args:
+            cloud_region_id (str): Cloud region ID
+            cloud_owner (str): Cloud owner name
+            kubeconfig (bytes): kubernetes cluster kubeconfig file
+
+        Returns:
+            ConnectivityInfo: Created object
+
+        """
+        json_file = jinja_env().get_template("multicloud_k8s_add_connectivity_info.json.j2").render(
+            cloud_region_id=cloud_region_id,
+            cloud_owner=cloud_owner
+        )
+        url: str = f"{cls.url}"
+        cls.send_message(
+            "POST",
+            "Create Connectivity Info",
+            url,
+            files={"file": kubeconfig,
+                   "metadata": (None, json_file)},
+            headers={}
+        )
+        return cls.get_connectivity_info_by_region_id(cloud_region_id)
diff --git a/src/onapsdk/msb/k8s/definition.py b/src/onapsdk/msb/k8s/definition.py
new file mode 100644
index 0000000..6c0def2
--- /dev/null
+++ b/src/onapsdk/msb/k8s/definition.py
@@ -0,0 +1,424 @@
+"""Definition 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 typing import Iterator
+from dataclasses import dataclass
+
+from onapsdk.utils.jinja import jinja_env
+from ..msb_service import MSB
+
+
+# pylint: disable=too-many-arguments, too-few-public-methods
+class DefinitionBase(MSB):
+    """DefinitionBase class."""
+
+    base_url = f"{MSB.base_url}/api/multicloud-k8s/v1/v1/rb/definition"
+
+    def __init__(self, rb_name: str,
+                 rb_version: str) -> None:
+        """Definition-Base object initialization.
+
+        Args:
+            rb_name (str): Definition name
+            rb_version (str): Definition version
+        """
+        super().__init__()
+        self.rb_name: str = rb_name
+        self.rb_version: str = rb_version
+
+    @property
+    def url(self) -> str:
+        """URL address for Definition Based calls.
+
+        Returns:
+            str: URL to RB Definition
+
+        """
+        return f"{self.base_url}/{self.rb_name}/{self.rb_version}"
+
+    def delete(self) -> None:
+        """Delete Definition Based object."""
+        self.send_message(
+            "DELETE",
+            f"Delete {self.__class__.__name__}",
+            self.url
+        )
+
+    def upload_artifact(self, package: bytes = None):
+        """Upload artifact.
+
+        Args:
+            package (bytes): Artifact to be uploaded to multicloud-k8s plugin
+
+        """
+        url: str = f"{self.url}/content"
+        self.send_message(
+            "POST",
+            "Upload Artifact content",
+            url,
+            data=package,
+            headers={}
+        )
+
+
+class Definition(DefinitionBase):
+    """Definition class."""
+
+    def __init__(self, rb_name: str,
+                 rb_version: str,
+                 chart_name: str,
+                 description: str,
+                 labels: dict) -> None:
+        """Definition object initialization.
+
+        Args:
+            rb_name (str): Definition name
+            rb_version (str): Definition version
+            chart_name (str): Chart name, optional field, will be detected if it is not provided
+            description (str): Definition description
+            labels (str): Labels
+        """
+        super().__init__(rb_name, rb_version)
+        self.rb_name: str = rb_name
+        self.rb_version: str = rb_version
+        self.chart_name: str = chart_name
+        self.description: str = description
+        self.labels: dict = labels
+
+    @classmethod
+    def get_all(cls):
+        """Get all definitions.
+
+        Yields:
+            Definition: Definition object
+
+        """
+        for definition in cls.send_message_json("GET",
+                                                "Get definitions",
+                                                cls.base_url):
+            yield cls(
+                definition["rb-name"],
+                definition["rb-version"],
+                definition.get("chart-name"),
+                definition.get("description"),
+                definition.get("labels")
+            )
+
+    @classmethod
+    def get_definition_by_name_version(cls, rb_name: str, rb_version: str) -> "Definition":
+        """Get definition by it's name and version.
+
+        Args:
+            rb_name (str): definition name
+            rb_version (str): definition version
+
+        Returns:
+            Definition: Definition object
+
+        """
+        url: str = f"{cls.base_url}/{rb_name}/{rb_version}"
+        definition: dict = cls.send_message_json(
+            "GET",
+            "Get definition",
+            url
+        )
+        return cls(
+            definition["rb-name"],
+            definition["rb-version"],
+            definition.get("chart-name"),
+            definition.get("description"),
+            definition.get("labels")
+        )
+
+    @classmethod
+    def create(cls, rb_name: str,
+               rb_version: str,
+               chart_name: str = "",
+               description: str = "",
+               labels=None) -> "Definition":
+        """Create Definition.
+
+        Args:
+            rb_name (str): Definition name
+            rb_version (str): Definition version
+            chart_name (str): Chart name, optional field, will be detected if it is not provided
+            description (str): Definition description
+            labels (str): Labels
+
+        Returns:
+            Definition: Created object
+
+        """
+        if labels is None:
+            labels = {}
+        url: str = f"{cls.base_url}"
+        cls.send_message(
+            "POST",
+            "Create definition",
+            url,
+            data=jinja_env().get_template("multicloud_k8s_add_definition.json.j2").render(
+                rb_name=rb_name,
+                rb_version=rb_version,
+                chart_name=chart_name,
+                description=description,
+                labels=labels
+            )
+        )
+        return cls.get_definition_by_name_version(rb_name, rb_version)
+
+    def create_profile(self, profile_name: str,
+                       namespace: str,
+                       kubernetes_version: str,
+                       release_name=None) -> "Profile":
+        """Create Profile for Definition.
+
+        Args:
+            profile_name (str): Name of profile
+            namespace (str): Namespace that service is created in
+            kubernetes_version (str): Required Kubernetes version
+            release_name (str): Release name
+
+        Returns:
+            Profile: Created object
+
+        """
+        url: str = f"{self.url}/profile"
+        if release_name is None:
+            release_name = profile_name
+        self.send_message(
+            "POST",
+            "Create profile for definition",
+            url,
+            data=jinja_env().get_template("multicloud_k8s_create_profile_"
+                                          "for_definition.json.j2").render(
+                                              rb_name=self.rb_name,
+                                              rb_version=self.rb_version,
+                                              profile_name=profile_name,
+                                              release_name=release_name,
+                                              namespace=namespace,
+                                              kubernetes_version=kubernetes_version
+                                          )
+        )
+        return self.get_profile_by_name(profile_name)
+
+    def get_all_profiles(self) -> Iterator["Profile"]:
+        """Get all profiles.
+
+        Yields:
+            Profile: Profile object
+
+        """
+        url: str = f"{self.url}/profile"
+
+        for profile in self.send_message_json("GET",
+                                              "Get profiles",
+                                              url):
+            yield Profile(
+                profile["rb-name"],
+                profile["rb-version"],
+                profile["profile-name"],
+                profile["namespace"],
+                profile.get("kubernetes-version"),
+                profile.get("labels"),
+                profile.get("release-name")
+            )
+
+    def get_profile_by_name(self, profile_name: str) -> "Profile":
+        """Get profile by it's name.
+
+        Args:
+            profile_name (str): profile name
+
+        Returns:
+            Profile: Profile object
+
+        """
+        url: str = f"{self.url}/profile/{profile_name}"
+
+        profile: dict = self.send_message_json(
+            "GET",
+            "Get profile",
+            url
+        )
+        return Profile(
+            profile["rb-name"],
+            profile["rb-version"],
+            profile["profile-name"],
+            profile["namespace"],
+            profile.get("kubernetes-version"),
+            profile.get("labels"),
+            profile.get("release-name")
+        )
+
+    def get_all_configuration_templates(self):
+        """Get all configuration templates.
+
+        Yields:
+            ConfigurationTemplate: ConfigurationTemplate object
+
+        """
+        url: str = f"{self.url}/config-template"
+
+        for template in self.send_message_json("GET",
+                                               "Get configuration templates",
+                                               url):
+            yield ConfigurationTemplate(
+                self.rb_name,
+                self.rb_version,
+                template["template-name"],
+                template.get("description")
+            )
+
+    def create_configuration_template(self, template_name: str,
+                                      description="") -> "ConfigurationTemplate":
+        """Create configuration template.
+
+        Args:
+            template_name (str): Name of the template
+            description (str): Description
+
+        Returns:
+            ConfigurationTemplate: Created object
+
+        """
+        url: str = f"{self.url}/config-template"
+
+        self.send_message(
+            "POST",
+            "Create configuration template",
+            url,
+            data=jinja_env().get_template("multicloud_k8s_create_configuration_"
+                                          "template.json.j2").render(
+                                              template_name=template_name,
+                                              description=description
+                                          )
+        )
+
+        return self.get_configuration_template_by_name(template_name)
+
+    def get_configuration_template_by_name(self, template_name: str) -> "ConfigurationTemplate":
+        """Get configuration template.
+
+        Args:
+            template_name (str): Name of the template
+
+        Returns:
+            ConfigurationTemplate: object
+
+        """
+        url: str = f"{self.url}/config-template/{template_name}"
+
+        template: dict = self.send_message_json(
+            "GET",
+            "Get Configuration template",
+            url
+        )
+        return ConfigurationTemplate(
+            self.rb_name,
+            self.rb_version,
+            template["template-name"],
+            template.get("description")
+        )
+
+
+class ProfileBase(DefinitionBase):
+    """ProfileBase class."""
+
+    def __init__(self, rb_name: str,
+                 rb_version: str,
+                 profile_name: str) -> None:
+        """Profile-Base object initialization.
+
+        Args:
+            rb_name (str): Definition name
+            rb_version (str): Definition version
+            profile_name (str): Name of profile
+        """
+        super().__init__(rb_name, rb_version)
+        self.rb_name: str = rb_name
+        self.rb_version: str = rb_version
+        self.profile_name: str = profile_name
+
+    @property
+    def url(self) -> str:
+        """URL address for Profile calls.
+
+        Returns:
+            str: URL to RB Profile
+
+        """
+        return f"{super().url}/profile/{self.profile_name}"
+
+
+@dataclass
+class Profile(ProfileBase):
+    """Profile class."""
+
+    def __init__(self, rb_name: str,
+                 rb_version: str,
+                 profile_name: str,
+                 namespace: str,
+                 kubernetes_version: str,
+                 labels=None,
+                 release_name=None) -> None:
+        """Profile object initialization.
+
+        Args:
+            rb_name (str): Definition name
+            rb_version (str): Definition version
+            profile_name (str): Name of profile
+            release_name (str): Release name, if release_name is not provided,
+            namespace (str): Namespace that service is created in
+            kubernetes_version (str): Required Kubernetes version
+            labels (dict): Labels
+        """
+        super().__init__(rb_name, rb_version, profile_name)
+        if release_name is None:
+            release_name = profile_name
+        self.release_name: str = release_name
+        self.namespace: str = namespace
+        self.kubernetes_version: str = kubernetes_version
+        self.labels: dict = labels
+        if self.labels is None:
+            self.labels = dict()
+
+
+class ConfigurationTemplate(DefinitionBase):
+    """ConfigurationTemplate class."""
+
+    @property
+    def url(self) -> str:
+        """URL address for ConfigurationTemplate calls.
+
+        Returns:
+            str: URL to Configuration template in Multicloud-k8s API.
+
+        """
+        return f"{super().url}/config-template/{self.template_name}"
+
+    def __init__(self, rb_name: str,
+                 rb_version: str,
+                 template_name: str,
+                 description="") -> None:
+        """Configuration-Template object initialization.
+
+        Args:
+            rb_name (str): Definition name
+            rb_version (str): Definition version
+            template_name (str): Configuration template name
+            description (str): Namespace that service is created in
+        """
+        super().__init__(rb_name, rb_version)
+        self.template_name: str = template_name
+        self.description: str = description
diff --git a/src/onapsdk/msb/k8s/instance.py b/src/onapsdk/msb/k8s/instance.py
new file mode 100644
index 0000000..196b9d2
--- /dev/null
+++ b/src/onapsdk/msb/k8s/instance.py
@@ -0,0 +1,190 @@
+"""Instantiation 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 typing import Iterator
+from dataclasses import dataclass
+
+from onapsdk.msb import MSB
+from onapsdk.utils.jinja import jinja_env
+
+
+# pylint: disable=too-many-arguments
+@dataclass
+class InstantiationRequest:
+    """Instantiation Request class."""
+
+    def __init__(self, request: dict) -> None:
+        """Request object initialization.
+
+        Args:
+            cloud_region_id (str): Cloud region ID
+            profile_name (str): Name of profile
+            rb_name (str): Definition name
+            rb_version (str): Definition version
+            override_values (dict): Optional parameters
+            labels (dict): Optional labels
+        """
+        super().__init__()
+        self.cloud_region_id: str = request["cloud-region"]
+        self.profile_name: str = request["profile-name"]
+        self.rb_name: str = request["rb-name"]
+        self.rb_version: str = request["rb-version"]
+        self.override_values: dict = request["override-values"]
+        self.labels: dict = request["labels"]
+
+
+@dataclass
+class InstantiationParameter:
+    """Class to store instantiation parameters used to pass override_values and labels.
+
+    Contains two values: name of parameter and it's value
+    """
+
+    name: str
+    value: str
+
+
+class Instance(MSB):
+    """Instance class."""
+
+    base_url = f"{MSB.base_url}/api/multicloud-k8s/v1/v1/instance"
+
+    def __init__(self, instance_id: str,
+                 namespace: str,
+                 request: InstantiationRequest,
+                 resources: dict = None,
+                 override_values: dict = None) -> None:
+        """Instance object initialization.
+
+        Args:
+            instance_id (str): instance ID
+            namespace (str): namespace that instance is created in
+            request (InstantiationRequest): datails of the instantiation request
+            resources (dict): Created resources
+            override_values (dict): Optional values
+        """
+        super().__init__()
+        self.instance_id: str = instance_id
+        self.namespace: str = namespace
+        self.request: InstantiationRequest = request
+        self.resources: dict = resources
+        self.override_values: dict = override_values
+
+    @property
+    def url(self) -> str:
+        """URL address.
+
+        Returns:
+            str: URL to Instance
+
+        """
+        return f"{self.base_url}/{self.instance_id}"
+
+    @classmethod
+    def get_all(cls) -> Iterator["Instance"]:
+        """Get all instantiated Kubernetes resources.
+
+        Yields:
+            Instantiation: Instantiation object
+
+        """
+        for resource in cls.send_message_json("GET",
+                                              "Get Kubernetes resources",
+                                              cls.base_url):
+            yield cls(
+                instance_id=resource["id"],
+                namespace=resource["namespace"],
+                request=InstantiationRequest(resource["request"])
+            )
+
+    @classmethod
+    def get_by_id(cls, instance_id: str) -> "Instance":
+        """Get Kubernetes resource by id.
+
+        Args:
+            instance_id (str): instance ID
+
+        Returns:
+            Instantiation: Instantiation object
+
+        """
+        url: str = f"{cls.base_url}/{instance_id}"
+        resource: dict = cls.send_message_json(
+            "GET",
+            "Get Kubernetes resource by id",
+            url
+        )
+        return cls(
+            instance_id=resource["id"],
+            namespace=resource["namespace"],
+            request=InstantiationRequest(resource["request"]),
+            resources=resource["resources"],
+            override_values=resource.get("override-values")
+        )
+
+    @classmethod
+    def create(cls,
+               cloud_region_id: str,
+               profile_name: str,
+               rb_name: str,
+               rb_version: str,
+               override_values: dict = None,
+               labels: dict = None) -> "Instance":
+        """Create Instance.
+
+        Args:
+            cloud_region_id (str): Cloud region ID
+            profile_name (str): Name of profile to be instantiated
+            rb_name: (bytes): Definition name
+            rb_version (str): Definition version
+            override_values (dict): List of optional override values
+            labels (dict): List of optional labels
+
+        Returns:
+            Instance: Created object
+
+        """
+        if labels is None:
+            labels = {}
+        if override_values is None:
+            override_values = {}
+        url: str = f"{cls.base_url}"
+        response: dict = cls.send_message_json(
+            "POST",
+            "Create Instance",
+            url,
+            data=jinja_env().get_template("multicloud_k8s_instantiate.json.j2").render(
+                cloud_region_id=cloud_region_id,
+                profile_name=profile_name,
+                rb_name=rb_name,
+                rb_version=rb_version,
+                override_values=override_values,
+                labels=labels),
+            headers={}
+        )
+        return cls(
+            instance_id=response["id"],
+            namespace=response["namespace"],
+            request=InstantiationRequest(response["request"]),
+            resources=response["resources"],
+            override_values=response.get("override-values")
+        )
+
+    def delete(self) -> None:
+        """Delete Instance object."""
+        self.send_message(
+            "DELETE",
+            f"Delete {self.instance_id} instance",
+            self.url
+        )