Swarming: conditionally run tests on swarming in chromium_trybot recipe.

scripts/slave/recipe_modules/swarming/api.py

 # Copyright 2014 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 slave import recipe_api
 
 
 # Minimally supported version of swarming.py script (reported by --version).
 MINIMAL_SWARMING_VERSION = (0, 4, 8)
 
@@ skipping 190 matching lines @@
         x.startswith('--test-launcher-summary-output=') for x in extra_args)
     if bad_args:  # pragma: no cover
       raise ValueError('--test-launcher-summary-output should not be used.')
 
     # Append it. output.json name is expected by collect_gtest_task.py.
     extra_args.append(
         '--test-launcher-summary-output=${ISOLATED_OUTDIR}/output.json')
 
     # Make a task, configure it to be collected through shim script.
     task = self.task(title, isolated_hash, make_unique, shards, extra_args)
-    task.collect_step_builder = lambda *args: (
-        self._gtest_collect_step(test_launcher_summary_output, *args))
+    task.collect_step_builder = lambda *args, **kwargs: (
+        self._gtest_collect_step(test_launcher_summary_output, *args, **kwargs))
     return task
 
   def check_client_version(self):
     """Yields steps to verify compatibility with swarming_client version."""
     yield self.m.swarming_client.ensure_script_version(
         'swarming.py', MINIMAL_SWARMING_VERSION)
 
-  def trigger(self, tasks):
+  def trigger(self, tasks, **kwargs):
     """Asynchronously launches a set of tasks on Swarming.
 
     This steps justs posts the tasks and immediately returns. Use 'collect' to
     wait for a task to finish and grab its result.
 
     Args:
       tasks: an enumerable of SwarmingTask instances.
+      kwargs: passed to recipe step constructor as-is, may contain
+        always_run, can_fail_build, etc.
     """
     # TODO(vadimsh): Trigger multiple tasks as a single step.
     assert all(isinstance(t, SwarmingTask) for t in tasks)
     steps = []
     for task in tasks:
       assert task.task_id not in self._pending_tasks, (
           'Triggered same task twice: %s' % task.task_id)
       self._pending_tasks.add(task.task_id)
 
       # Trigger parameters.
@@ skipping 21 matching lines @@
 
       # Additional command line args for isolated command.
       if task.extra_args:
         args.append('--')
         args.extend(task.extra_args)
 
       # Build corresponding step.
       steps.append(self.m.python(
           name=self._get_step_name('trigger', task),
           script=self.m.swarming_client.path.join('swarming.py'),
-          args=args))
+          args=args,
+          **kwargs))
 
     return steps
 
-  def collect(self, tasks, followup_fn=None):
+  def collect(self, tasks, **kwargs):
     """Waits for a set of Swarming tasks to finish.
 
     Always waits for all task results. Failed tasks will be marked as such
     but would not abort the build (corresponds to always_run=True step
     property).
 
     Args:
       tasks: an enumerable of SwarmingTask instances. All of them should have
           been triggered previously with 'trigger' method.
-      followup_fn: if given, will be called with each step result before step is
-          finalized. Can be used to customize step appearance.
+      kwargs: passed to recipe step constructor as-is, may contain
+        always_run, can_fail_build, followup_fn, etc.
     """
     # TODO(vadimsh): Implement "wait for any" to wait for first finished task.
     # TODO(vadimsh): Update |tasks| in-place with results of task execution.
     # TODO(vadimsh): Add timeouts.
     assert all(isinstance(t, SwarmingTask) for t in tasks)
     steps = []
     for task in tasks:
       assert task.task_id in self._pending_tasks, (
           'Trying to collect a task that was not triggered: %s' % task.task_id)
       self._pending_tasks.remove(task.task_id)
-      steps.append(task.collect_step_builder(task, followup_fn))
+      steps.append(task.collect_step_builder(task, **kwargs))
     return steps
 
-  def _default_collect_step(self, task, followup_fn):
+  def _default_collect_step(self, task, **kwargs):
     """Produces a step that collects a result of an arbitrary task."""
-    # Always wait for all tasks to finish even if some of them failed.
+    # By default wait for all tasks to finish even if some of them failed.
+    kwargs.setdefault('always_run', True)
     return self.m.python(
         name=self._get_step_name('swarming', task),
         script=self.m.swarming_client.path.join('swarming.py'),
         args=self._get_collect_cmd_args(task),
-        always_run=True,
-        followup_fn=followup_fn)
+        **kwargs)
 
-  def _gtest_collect_step(self, merged_test_output, task, followup_fn):
+  def _gtest_collect_step(self, merged_test_output, task, **kwargs):
     """Produces a step that collects and processes a result of gtest task."""
+    # By default wait for all tasks to finish even if some of them failed.
+    kwargs.setdefault('always_run', True)
+
     # Shim script's own arguments.
     args = [
       '--swarming-client-dir', self.m.swarming_client.path,
       '--temp-root-dir', self.m.path['tmp_base'],
     ]
 
     # Where to put combined summary to, consumed by recipes. Also emit
     # test expectation only if |merged_test_output| is really used.
     step_test_data = None
     if merged_test_output:
       args.extend(['--merged-test-output', merged_test_output])
       step_test_data = lambda: self.m.json.test_api.canned_gtest_output(True)
 
     # Arguments for actual 'collect' command.
     args.append('--')
     args.extend(self._get_collect_cmd_args(task))
 
     # Always wait for all tasks to finish even if some of them failed. Allow
     # collect_gtest_task.py to emit all necessary annotations itself.
     return self.m.python(
         name=self._get_step_name('swarming', task),
         script=self.resource('collect_gtest_task.py'),
         args=args,
-        always_run=True,
         allow_subannotations=True,
-        followup_fn=followup_fn,
-        step_test_data=step_test_data)
+        step_test_data=step_test_data,
+        **kwargs)
 
   def _get_step_name(self, prefix, task):
     """SwarmingTask -> name of a step of a waterfall.
 
     Will take a task title (+ step name prefix) and optionally append
     OS dimension to it in case the task is triggered on OS that is different
     from OS this recipe is running on. It shortens step names for the most
     common case of triggering a task on the same OS as one that recipe
     is running on.
 
@@ skipping 50 matching lines @@
           isolated tests based on gtest. Swarming uses GTEST_SHARD_INDEX
           and GTEST_TOTAL_SHARDS environment variables to tell the executable
           what shard to run.
       builder: buildbot builder this task was triggered from.
       build_number: build number of a build this task was triggered from.
       profile: True to enable swarming profiling.
       suffix: string suffix to append to task ID.
       extra_args: list of command line arguments to pass to isolated tasks.
       collect_step_builder: callback that will be called to generate recipe step
           that collects and processes results of task execution, signature is
-          collect_step_builder(task, followup_fn).
+          collect_step_builder(task, **step_kwargs).
     """
     assert 'os' in dimensions
     self.title = title
     self.isolated_hash = isolated_hash
     self.dimensions = dimensions.copy()
     self.env = env.copy()
     self.priority = priority
     self.shards = shards
     self.builder = builder
     self.build_number = build_number
     self.profile = profile
     self.suffix = suffix
     self.extra_args = tuple(extra_args or [])
     self.collect_step_builder = collect_step_builder
 
   @property
   def task_id(self):
     """ID of this task, derived from its other properties.
 
     Task ID identifies what task is doing and what machine configuration it
     expects. It is used as a key in table of a cached results. If Swarming
     service figures out that a task with given ID has successfully finished
     before, it will reuse the result right away, without even running
     the task again.
     """
     return '%s/%s/%s/%s/%d%s' % (
         self.title, self.dimensions['os'], self.isolated_hash,
         self.builder, self.build_number, self.suffix)