blob: 0994e4509f12f532146a4ca093e2b5881214fc19 [file] [log] [blame]
# 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
# 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
class Step(object):
_api = attr.ib()
name = attr.ib()
dir = attr.ib()
_export_dir_name = attr.ib(default=None)
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._use_full_argument = not props.do_not_use_full_argument
self._export_dir_name = props.export_dir_name
self._root = None
self._checkout_root = None
self._step_objects = None
self._initialized = False
def command_name(self):
return self._command_name
def root(self):
return self._root
def export_dir_name(self):
return self._export_dir_name
def _step(self, name):
return Step(
def has_props(self):
return self._input_steps or self._input_programs
def init(self, checkout_root):
self._initialized = True
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 = (
program_steps = (
['--program', program, '--only-list-steps'],
step_test_data=lambda: raw_io_stream_output(
for step_name in program_steps:
self._step_objects[step_name] = self._step(step_name)
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._initialized: # pragma: no cover
raise self.m.step.StepFailure('api.pw_presubmit.init() not called')
return self._step_objects.values()
def _step_timeout(self):
# Amount of time elapsed in the run.
elapsed_time = (
self.m.time.time() -
# Amount of time before build times out.
time_remaining = ( - 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 += [
return self.m.step(name, cmd, timeout=self._step_timeout(), **kwargs)
def run(self, step, log_dir=None):
with self.m.step.nest( as pres:
args = []
if self._only_on_changed_files:
args.extend(('--base', 'HEAD~1'))
elif self._use_full_argument:
with self.m.step.defer_results():
if log_dir:
step_log_dir = log_dir.join(
log_dir = step.export_dir
if step.export_dir:
'mkdir {}'.format(self.export_dir_name), step.export_dir
if log_dir and log_dir != step.export_dir:
self.m.file.ensure_directory('create log dir', log_dir), log_dir)