recipe_modules/pw_presubmit/api.py

# Copyright 2021 The Pigweed Authors
#
# 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
#
#     https://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.
"""Wrapper for 'pw presubmit' in the project source tree."""

import collections

import attr
from recipe_engine import config_types, recipe_api
from RECIPE_MODULES.fuchsia.utils import memoize


@attr.s
class Step(object):
    _api = attr.ib()
    name = attr.ib()
    dir = attr.ib()
    _export_dir_name = attr.ib(default=None)

    @property
    def export_dir(self):
        if not self._export_dir_name:
            return None  # pragma: no cover
        return self.dir.join(self._export_dir_name)


class PwPresubmitApi(recipe_api.RecipeApi):
    """Calls to checkout code."""

    def __init__(self, props, *args, **kwargs):
        super(PwPresubmitApi, self).__init__(*args, **kwargs)
        self._command_name = props.command_name or 'python -m pw_cli'
        self._input_steps = list(props.step)
        self._input_programs = list(props.program)
        self._only_on_changed_files = props.only_on_changed_files
        self._export_dir_name = props.export_dir_name
        self._root = None
        self._checkout_root = None
        self._step_objects = None

    @property
    def command_name(self):
        return self._command_name

    @property
    def root(self):
        return self._root

    @property
    def export_dir_name(self):
        return self._export_dir_name

    def _step(self, name):
        return Step(
            self.m,
            name,
            self.root.join(name),
            export_dir_name=self.export_dir_name,
        )

    def init(self, checkout_root):
        if not self._input_steps and not self._input_programs:
            raise self.m.step.StepFailure('no step or program properties')

        self._root = self.m.path['start_dir'].join('presubmit')
        self._checkout_root = checkout_root

        self._step_objects = collections.OrderedDict()

        for step_name in self._input_steps:
            self._step_objects[step_name] = self._step(step_name)

        if self._input_programs:
            with self.m.step.nest('get steps from programs'):
                for program in self._input_programs:
                    # To get step_test_data line to pass pylint.
                    raw_io_stream_output = (
                        self.m.raw_io.test_api.stream_output_text
                    )

                    program_steps = (
                        self._run(
                            ['--program', program, '--only-list-steps'],
                            name=program,
                            stdout=self.m.raw_io.output_text(),
                            step_test_data=lambda: raw_io_stream_output(
                                '{0}_0\n{0}_1\n'.format(program),
                            ),
                        )
                        .stdout.strip()
                        .splitlines()
                    )

                    for step_name in program_steps:
                        self._step_objects[step_name] = self._step(step_name)

        if not self._step_objects:
            raise self.m.step.StepFailure('no steps to execute')

    def steps(self):
        # We shouldn't get to here, but in case the caller doesn't call init()
        # first we'll check anyway.
        if not self._step_objects:  # pragma: no cover
            raise self.m.step.StepFailure('no steps to execute')

        return self._step_objects.values()

    def _step_timeout(self):
        # Amount of time elapsed in the run.
        elapsed_time = (
            self.m.time.time() - self.m.buildbucket.build.start_time.seconds
        )

        # Amount of time before build times out.
        time_remaining = (
            self.m.buildbucket.build.execution_timeout.seconds - elapsed_time
        )

        # Give a buffer before build times out and kill this step then. This
        # should give enough time to read any logfiles and maybe upload to
        # logdog/GCS before the build times out.
        step_timeout = time_remaining - 60

        # If the timeout would be negative or very small set it to 30 seconds.
        # We likely won't have enough information to debug these steps, but in
        # case they're fast there's no reason to kill them much before the
        # build is terminated.
        if step_timeout < 30:
            step_timeout = 30

        return step_timeout

    def _run(self, args, name='run', **kwargs):
        cmd = self._command_name.split()
        cmd += [
            '--directory',
            self._checkout_root,
            '--loglevel',
            'debug',
            'presubmit',
            '--package-root',
            self.m.path['cache'],
            '--output-directory',
            self._root,
        ]

        cmd.extend(args)

        return self.m.step(name, cmd, timeout=self._step_timeout(), **kwargs)

    def run(self, step):
        with self.m.step.nest(step.name) as pres:
            args = []

            if self._only_on_changed_files:
                args.extend(('--base', 'HEAD~1'))

            args.extend(('--step', step.name))

            self._run(args, name=step.name)

            if step.export_dir:
                self.m.file.ensure_directory(
                    'mkdir {}'.format(self.export_dir_name), step.export_dir,
                )

            self.m.build.save_logs(step.dir, step.export_dir)