Pin black version in tox.ini to 23.12.1

[osm/N2VC.git] / n2vc / k8s_helm3_conn.py

diff --git a/n2vc/k8s_helm3_conn.py b/n2vc/k8s_helm3_conn.py
index d828da2..14f7fe0 100644 (file)
--- a/n2vc/k8s_helm3_conn.py
+++ b/n2vc/k8s_helm3_conn.py
@@ -20,6 +20,7 @@
 # contact with: nfvlabs@tid.es
 ##
 from typing import Union
+from shlex import quote
 import os
 import yaml
 
@@ -118,6 +119,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         if namespace and namespace != "kube-system":
             if not await self._namespace_exists(cluster_uuid, namespace):
                 try:
+                    # TODO: refactor to use kubernetes API client
                     await self._create_namespace(cluster_uuid, namespace)
                 except Exception as e:
                     if not await self._namespace_exists(cluster_uuid, namespace):
@@ -149,7 +151,6 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         return True
 
     async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
-
         self.log.debug(
             "inspect kdu_model {} from (optional) repo: {}".format(kdu_model, repo_url)
         )
@@ -250,7 +251,6 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         return namespace in namespaces if namespaces else False
 
     async def _get_namespaces(self, cluster_id: str):
-
         self.log.debug("get namespaces cluster_id {}".format(cluster_id))
 
         # init config, env
@@ -259,7 +259,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         )
 
         command = "{} --kubeconfig={} get namespaces -o=yaml".format(
-            self.kubectl_command, paths["kube_config"]
+            self.kubectl_command, quote(paths["kube_config"])
         )
         output, _rc = await self._local_async_exec(
             command=command, raise_exception_on_error=True, env=env
@@ -272,7 +272,6 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         return namespaces
 
     async def _create_namespace(self, cluster_id: str, namespace: str):
-
         self.log.debug(f"create namespace: {cluster_id} for cluster_id: {namespace}")
 
         # init config, env
@@ -281,7 +280,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         )
 
         command = "{} --kubeconfig={} create namespace {}".format(
-            self.kubectl_command, paths["kube_config"], namespace
+            self.kubectl_command, quote(paths["kube_config"]), quote(namespace)
         )
         _, _rc = await self._local_async_exec(
             command=command, raise_exception_on_error=True, env=env
@@ -293,16 +292,17 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
     async def _get_services(
         self, cluster_id: str, kdu_instance: str, namespace: str, kubeconfig: str
     ):
-
         # init config, env
         paths, env = self._init_paths_env(
             cluster_name=cluster_id, create_if_not_exist=True
         )
 
         command1 = "env KUBECONFIG={} {} get manifest {} --namespace={}".format(
-            kubeconfig, self._helm_command, kdu_instance, namespace
+            kubeconfig, self._helm_command, quote(kdu_instance), quote(namespace)
+        )
+        command2 = "{} get --namespace={} -f -".format(
+            self.kubectl_command, quote(namespace)
         )
-        command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
         output, _rc = await self._local_async_exec_pipe(
             command1, command2, env=env, raise_exception_on_error=True
         )
@@ -318,6 +318,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         if namespace != "kube-system":
             namespaces = await self._get_namespaces(cluster_id)
             if namespace not in namespaces:
+                # TODO: refactor to use kubernetes API client
                 await self._create_namespace(cluster_id, namespace)
 
         repo_list = await self.repo_list(cluster_id)
@@ -333,7 +334,6 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         pass
 
     async def _instances_list(self, cluster_id: str):
-
         # init paths, env
         paths, env = self._init_paths_env(
             cluster_name=cluster_id, create_if_not_exist=True
@@ -358,8 +358,8 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
 
         Args:
             show_command: the second part of the command (`helm show <show_command>`)
-            kdu_model: The name or path of an Helm Chart
-            repo_url: Helm Chart repository url
+            kdu_model: The name or path of a Helm Chart
+            repo_str: Helm Chart repository url
             version: constraint with specific version of the Chart to use
 
         Returns:
@@ -367,7 +367,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         """
 
         inspect_command = "{} show {} {}{} {}".format(
-            self._helm_command, show_command, kdu_model, repo_str, version
+            self._helm_command, show_command, quote(kdu_model), repo_str, version
         )
         return inspect_command
 
@@ -376,7 +376,11 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
     ):
         get_command = (
             "env KUBECONFIG={} {} get {} {} --namespace={} --output yaml".format(
-                kubeconfig, self._helm_command, get_command, kdu_instance, namespace
+                kubeconfig,
+                self._helm_command,
+                get_command,
+                quote(kdu_instance),
+                quote(namespace),
             )
         )
         return get_command
@@ -389,7 +393,6 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         yaml_format: bool = False,
         show_error_log: bool = False,
     ) -> Union[str, dict]:
-
         self.log.debug(
             "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
         )
@@ -402,7 +405,10 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
             cluster_name=cluster_id, create_if_not_exist=True
         )
         command = "env KUBECONFIG={} {} status {} --namespace={} --output yaml".format(
-            paths["kube_config"], self._helm_command, kdu_instance, namespace
+            paths["kube_config"],
+            self._helm_command,
+            quote(kdu_instance),
+            quote(namespace),
         )
 
         output, rc = await self._local_async_exec(
@@ -448,7 +454,6 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         timeout: float,
         kubeconfig: str,
     ) -> str:
-
         timeout_str = ""
         if timeout:
             timeout_str = "--timeout {}s".format(timeout)
@@ -460,7 +465,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         # namespace
         namespace_str = ""
         if namespace:
-            namespace_str = "--namespace {}".format(namespace)
+            namespace_str = "--namespace {}".format(quote(namespace))
 
         # version
         version_str = ""
@@ -472,12 +477,12 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
             "{params} {timeout} {ns} {model} {ver}".format(
                 kubeconfig=kubeconfig,
                 helm=self._helm_command,
-                name=kdu_instance,
+                name=quote(kdu_instance),
                 atomic=atomic_str,
                 params=params_str,
                 timeout=timeout_str,
                 ns=namespace_str,
-                model=kdu_model,
+                model=quote(kdu_model),
                 ver=version_str,
             )
         )
@@ -544,6 +549,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         atomic: bool,
         timeout: float,
         kubeconfig: str,
+        force: bool = False,
     ) -> str:
         """Generates the command to upgrade a Helm Chart release
 
@@ -557,7 +563,7 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
                 The --wait flag will be set automatically if --atomic is used
             timeout (float): The time, in seconds, to wait
             kubeconfig (str): Kubeconfig file path
-
+            force (bool): If set, helm forces resource updates through a replacement strategy. This may recreate pods.
         Returns:
             str: command to upgrade a Helm Chart release
         """
@@ -571,28 +577,34 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         if atomic:
             atomic_str = "--atomic"
 
+        # force
+        force_str = ""
+        if force:
+            force_str = "--force "
+
         # version
         version_str = ""
         if version:
-            version_str = "--version {}".format(version)
+            version_str = "--version {}".format(quote(version))
 
         # namespace
         namespace_str = ""
         if namespace:
-            namespace_str = "--namespace {}".format(namespace)
+            namespace_str = "--namespace {}".format(quote(namespace))
 
         command = (
-            "env KUBECONFIG={kubeconfig} {helm} upgrade {name} {model} {namespace} {atomic} "
+            "env KUBECONFIG={kubeconfig} {helm} upgrade {name} {model} {namespace} {atomic} {force}"
             "--output yaml {params} {timeout} --reuse-values {ver}"
         ).format(
             kubeconfig=kubeconfig,
             helm=self._helm_command,
-            name=kdu_instance,
+            name=quote(kdu_instance),
             namespace=namespace_str,
             atomic=atomic_str,
+            force=force_str,
             params=params_str,
             timeout=timeout_str,
-            model=kdu_model,
+            model=quote(kdu_model),
             ver=version_str,
         )
         return command
@@ -601,15 +613,18 @@ class K8sHelm3Connector(K8sHelmBaseConnector):
         self, kdu_instance: str, namespace: str, revision: float, kubeconfig: str
     ) -> str:
         return "env KUBECONFIG={} {} rollback {} {} --namespace={} --wait".format(
-            kubeconfig, self._helm_command, kdu_instance, revision, namespace
+            kubeconfig,
+            self._helm_command,
+            quote(kdu_instance),
+            revision,
+            quote(namespace),
         )
 
     def _get_uninstall_command(
         self, kdu_instance: str, namespace: str, kubeconfig: str
     ) -> str:
-
         return "env KUBECONFIG={} {} uninstall {} --namespace={}".format(
-            kubeconfig, self._helm_command, kdu_instance, namespace
+            kubeconfig, self._helm_command, quote(kdu_instance), quote(namespace)
         )
 
     def _get_helm_chart_repos_ids(self, cluster_uuid) -> list: