+ ee_id: str = None,
+ path: str = None,
+ charm_id: str = None,
+ charm_type: str = None,
+ timeout: float = None,
+ ) -> str:
+ """This method upgrade charms in VNFs
+
+ Args:
+ ee_id: Execution environment id
+ path: Local path to the charm
+ charm_id: charm-id
+ charm_type: Charm type can be lxc-proxy-charm, native-charm or k8s-proxy-charm
+ timeout: (Float) Timeout for the ns update operation
+
+ Returns:
+ The output of the update operation if status equals to "completed"
+ """
+ raise K8sException(
+ "KDUs deployed with Juju Bundle do not support charm upgrade"
+ )
+
+ async def exec_primitive(
+ self,
+ cluster_uuid: str = None,
+ kdu_instance: str = None,
+ primitive_name: str = None,
+ timeout: float = 300,
+ params: dict = None,
+ db_dict: dict = None,
+ namespace: str = None,
+ **kwargs,
+ ) -> str:
+ """Exec primitive (Juju action)
+
+ :param cluster_uuid str: The UUID of the cluster
+ :param kdu_instance str: The unique name of the KDU instance
+ :param primitive_name: Name of action that will be executed
+ :param timeout: Timeout for action execution
+ :param params: Dictionary of all the parameters needed for the action
+ :param db_dict: Dictionary for any additional data
+ :param namespace str: The namespace (model) where the Bundle was deployed
+ :param kwargs: Additional parameters
+ vca_id (str): VCA ID
+
+ :return: Returns the output of the action
+ """
+ libjuju = await self._get_libjuju(kwargs.get("vca_id"))
+
+ namespace = self._obtain_namespace(
+ kdu_instance=kdu_instance, namespace=namespace
+ )
+
+ if not params or "application-name" not in params:
+ raise K8sException(
+ "Missing application-name argument, \
+ argument needed for K8s actions"
+ )
+ try:
+ self.log.debug(
+ "[exec_primitive] Getting model "
+ "{} for the kdu_instance: {}".format(namespace, kdu_instance)
+ )
+ application_name = params["application-name"]
+ actions = await libjuju.get_actions(
+ application_name=application_name, model_name=namespace
+ )
+ if primitive_name not in actions:
+ raise K8sException("Primitive {} not found".format(primitive_name))
+ output, status = await libjuju.execute_action(
+ application_name=application_name,
+ model_name=namespace,
+ action_name=primitive_name,
+ **params,
+ )
+
+ if status != "completed":
+ raise K8sException(
+ "status is not completed: {} output: {}".format(status, output)
+ )
+ if self.on_update_db:
+ await self.on_update_db(
+ cluster_uuid=cluster_uuid,
+ kdu_instance=kdu_instance,
+ filter=db_dict["filter"],
+ )
+
+ return output
+
+ except Exception as e:
+ error_msg = "Error executing primitive {}: {}".format(primitive_name, e)
+ self.log.error(error_msg)
+ raise K8sException(message=error_msg)
+
+ """Introspection"""
+
+ async def inspect_kdu(self, kdu_model: str) -> dict: