tests/helpers/helm.py

# Copyright (C) 2019 The Android Open Source Project
#
# 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 subprocess


class Helm:
    def __init__(self, kubeconfig, kubecontext):
        """Wrapper for Helm CLI.

  Arguments:
    kubeconfig {str} -- Path to kubeconfig-file describing the cluster to
                        connect to.
    kubecontext {str} -- Name of the context to use.
  """

        self.kubeconfig = kubeconfig
        self.kubecontext = kubecontext

    def _exec_command(self, cmd, fail_on_err=True):
        base_cmd = [
            "helm",
            "--kubeconfig",
            self.kubeconfig,
            "--kube-context",
            self.kubecontext,
        ]
        return subprocess.run(
            base_cmd + cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=fail_on_err,
            text=True,
        )

    def init(self, serviceaccount):
        """Installs tiller on the cluster.

    Arguments:
      serviceaccount {str} -- Name of the service account, which tiller is meant
                              to use.

    Returns:
      CompletedProcess -- CompletedProcess-object returned by subprocess
                          containing details about the result and output of the
                          executed command.
    """

        helm_cmd = ["init", "--wait", "--service-account", serviceaccount]
        return self._exec_command(helm_cmd)

    def install(
        self,
        chart,
        name,
        values_file=None,
        set_values=None,
        namespace=None,
        fail_on_err=True,
    ):
        """Installs a chart on the cluster

    Arguments:
      chart {str} -- Release name or path of a helm chart
      name {str} -- Name with which the chart will be installed on the cluster

    Keyword Arguments:
      values_file {str} -- Path to a custom values.yaml file (default: {None})
      set_values {dict} -- Dictionary containing key-value-pairs that are used
                           to overwrite values in the values.yaml-file.
                           (default: {None})
      namespace {str} -- Namespace to install the release into (default: {default})
      fail_on_err {bool} -- Whether to fail with an exception if the installation
                            fails (default: {True})

    Returns:
      CompletedProcess -- CompletedProcess-object returned by subprocess
                          containing details about the result and output of the
                          executed command.
    """

        helm_cmd = ["install", chart, "--dep-up", "-n", name, "--wait"]
        if values_file:
            helm_cmd.extend(("-f", values_file))
        if set_values:
            opt_list = ["%s=%s" % (k, v) for k, v in set_values.items()]
            helm_cmd.extend(("--set", ",".join(opt_list)))
        if namespace:
            helm_cmd.extend(("--namespace", namespace))
        return self._exec_command(helm_cmd, fail_on_err)

    def list(self):
        """Lists helm charts installed on the cluster.

    Returns:
      list -- List of helm chart realeases installed on the cluster.
    """

        helm_cmd = ["list", "--all", "--output", "json"]
        output = self._exec_command(helm_cmd).stdout
        output = json.loads(output)
        return output["Releases"]

    def upgrade(
        self,
        chart,
        name,
        values_file=None,
        set_values=None,
        reuse_values=True,
        recreate_pods=False,
        fail_on_err=True,
    ):
        """Updates a chart on the cluster

      Arguments:
        chart {str} -- Release name or path of a helm chart
        name {str} -- Name with which the chart will be installed on the cluster

      Keyword Arguments:
        values_file {str} -- Path to a custom values.yaml file (default: {None})
        set_values {dict} -- Dictionary containing key-value-pairs that are used
                            to overwrite values in the values.yaml-file.
                            (default: {None})
        reuse_values {bool} -- Whether to reuse existing not overwritten values
                              (default: {True})
        recreate_pods {bool} -- Whether to restart changed pods (default: {False})
        fail_on_err {bool} -- Whether to fail with an exception if the installation
                              fails (default: {True})

      Returns:
        CompletedProcess -- CompletedProcess-object returned by subprocess
                            containing details about the result and output of the
                            executed command.
      """
        helm_cmd = ["upgrade", name, chart, "--wait"]
        if values_file:
            helm_cmd.extend(("-f", values_file))
        if reuse_values:
            helm_cmd.append("--reuse-values")
        if recreate_pods:
            helm_cmd.append("--recreate-pods")
        if set_values:
            opt_list = ["%s=%s" % (k, v) for k, v in set_values.items()]
            helm_cmd.extend(("--set", ",".join(opt_list)))
        return self._exec_command(helm_cmd, fail_on_err)

    def delete(self, name, purge=True):
        """Deletes a chart from the cluster

      Arguments:
        name {str} -- Name of the chart to delete

      Keyword Arguments:
        purge {bool} -- Whether to also remove the release metadata as well
                        (default: {True})

      Returns:
        CompletedProcess -- CompletedProcess-object returned by subprocess
                            containing details about the result and output of the
                            executed command.
      """

        helm_cmd = ["delete", name]
        if purge:
            helm_cmd.append("--purge")
        return self._exec_command(helm_cmd)

    def delete_all(self, exceptions=None):
        """Deletes all charts on the cluster

    Keyword Arguments:
      exceptions {list} -- List of chart names not to delete (default: {None})
    """

        charts = self.list()
        for chart in charts:
            if chart["Name"] in exceptions:
                continue
            self.delete(chart["Name"])

    def reset(self):
        """Uninstall Tiller from cluster

    Returns:
      CompletedProcess -- CompletedProcess-object returned by subprocess
                          containing details about the result and output of the
                          executed command.
    """

        helm_cmd = ["reset", "--force"]
        return self._exec_command(helm_cmd, fail_on_err=True)