Swarming: conditionally run tests on swarming in chromium_trybot recipe.

scripts/slave/recipe_modules/test_utils/api.py

 # Copyright 2013 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
 
 class TestUtilsApi(recipe_api.RecipeApi):
   @staticmethod
   def format_step_text(data):
     """
@@ skipping 22 matching lines @@
       else:  # pragma: no cover
         raise ValueError(
             'Expected a one or two-element list, got %r instead.' % section)
     return ''.join(step_text)
 
   class Test(object):
     """
     Base class for tests that can be retried after deapplying a previously
     applied patch.
     """
+    # If True, the test supports asynchronous execution. In that case 'trigger'
+    # and 'collect' will be used instead of 'run'. 'trigger' produces a step
+    # that asynchronously launches the test (just starts the test and
+    # immediately returns control back to the recipe), and 'collect' produces
+    # a step that blocks until the test is finished. The recipe can execute any
+    # other steps in between. This mode is used by tests running on Swarming.
+    async = False
 
     @property
     def name(self):  # pragma: no cover
       """Name of the test."""
       raise NotImplementedError()
 
     def run(self, suffix):  # pragma: no cover
       """Run the test. suffix is 'with patch' or 'without patch'."""
       raise NotImplementedError()
 
+    def trigger(self, suffix):  # pragma: no cover
+      """Launch the test asynchronously, used if self.async == True."""
+      raise NotImplementedError()
+
+    def collect(self, suffix):  # pragma: no cover
+      """Wait for triggered test to finish, used if self.async == True."""
+      raise NotImplementedError()
+
     def has_valid_results(self, suffix):  # pragma: no cover
       """
       Returns True if results (failures) are valid.
 
       This makes it possible to distinguish between the case of no failures
       and the test failing to even report its results in machine-readable
       format.
       """
       raise NotImplementedError()
 
     def failures(self, suffix):  # pragma: no cover
       """Return list of failures (list of strings)."""
       raise NotImplementedError()
 
     def _step_name(self, suffix):
       """Helper to uniformly combine tests's name with a suffix."""
       return '%s (%s)' % (self.name, suffix)
 
   def determine_new_failures(self, tests, deapply_patch_fn):
     """
     Utility function for running steps with a patch applied, and retrying
     failing steps without the patch. Failures from the run without the patch are
     ignored.
 
     Args:
       tests - iterable of objects implementing the Test interface above
       deapply_patch_fn - function that takes a list of failing tests
                          and undoes any effect of the previously applied patch
     """
+    # Convert iterable to list, since it is enumerated multiple times.
+    tests = list(tests)
+
     if self.m.step_history.failed:
       yield self.m.python.inline(
         'Aborting due to failed build state.',
         "import sys; sys.exit(1)",
         always_run=True, abort_on_failure=True)
       return  # won't actually hit this, but be explicit
 
-    yield (t.run('with patch') for t in tests)
+    def run(prefix, tests):
+      """Runs synchronous and asynchronous tests (at the same time).
+
+      Asynchronous tests are launched first (just launched, not being blocked
+      on). While they are running, the recipe blocks on synchronous tests
+      (sequentially, one by one). And then finally waits for all asynchronous
+      tests to finish. Effectively asynchronous tests are running in parallel
+      with each other and with synchronous tests.
+      """
+      # Trigger all async tests first, so that they are running in parallel
+      # with synchronous tests.
+      yield (t.trigger(prefix) for t in tests if t.async)
+      # Now block on all synchronous tests.
+      yield (t.run(prefix) for t in tests if not t.async)
+      # And finally wait for all pending asynchronous tests to complete.
+      yield (t.collect(prefix) for t in tests if t.async)
+
+    yield run('with patch', tests)
 
     failing_tests = []
     for t in tests:
       if not t.has_valid_results('with patch'):
         yield self.m.python.inline(
           t.name,
           r"""
           import sys
           print 'TEST RESULTS WERE INVALID'
           sys.exit(1)
           """,
           always_run=True)
       elif t.failures('with patch'):
         failing_tests.append(t)
     if not failing_tests:
       return
 
     yield deapply_patch_fn(failing_tests)
 
-    yield (t.run('without patch') for t in failing_tests)
+    yield run('without patch', failing_tests)
     yield (self._summarize_retried_test(t) for t in failing_tests)
 
   def _summarize_retried_test(self, test):
     if not test.has_valid_results('without patch'):
       return self.m.python.inline(
         test.name,
         r"""
         import sys
         print 'TEST RESULTS WERE INVALID'
         sys.exit(1)
@@ skipping 39 matching lines @@
       """,
       args=[
         self.m.json.input({
           'new': list(new_failures),
           'ignored': list(ignored_failures),
         })
       ],
       followup_fn=followup_fn,
       always_run=True,
     )