recipe_modules/kubectl/api.py

# Copyright 2019 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from recipe_engine import recipe_api


class KubectlApi(recipe_api.RecipeApi):
    """Provides steps to connect and run Kubectl images."""

    def __init__(self, luci_context, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._luci_context = luci_context

    def __call__(self, *args, **kwargs):
        """Executes specified kubectl command.

        Args:
        args: arguments passed to the 'kubectl' command including subcommand name,
            e.g. api.kubectl('config', 'view', step_name='view kubectl configuration').
        kwargs: arguments passed down to api.step module.
        """
        cmd = [self._kubectl_executable]
        step_name = kwargs.pop("step_name", "kubectl %s" % args[0])
        return self.m.step(step_name, cmd + list(args), **kwargs)

    @property
    def _kubectl_executable(self):
        self.m.gcloud.patch_gcloud_invoker()
        return self.m.ensure_tool("kubectl", self.resource("tool_manifest.json"))

    @property
    def version(self):
        """Returns kubectl client version installed or None if failed to detect."""
        step = self(
            "version",
            stdout=self.m.raw_io.output_text(),
            step_name="retrieve kubectl version",
            step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(
                'Client Version: version.Info{Major:"1", Minor:"16", GitVersion:"v1.16.1", GitCommit:"d647ddbd755faf07169599a625faf302ffc34458", GitTreeState:"clean", BuildDate:"2019-10-02T17:01:15Z", GoVersion:"go1.12.10", Compiler:"gc", Platform:"linux/amd64"}'
            ),
        )
        if 'GitCommit:"' in step.stdout:
            index = step.stdout.find('GitCommit:"')
            index_end = step.stdout.find('"', index + len('GitCommit:"'))
            version = step.stdout[index + len('GitCommit:"') : index_end]
        else:
            version = "unknown"
        step.presentation.step_text = version
        return version