Add support for swarming priority and expiration in the test spec

scripts/slave/recipe_modules/chromium_tests/steps.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.
 
 import re
 import string
 
 
 class Test(object):
   """
@@ skipping 401 matching lines @@
             test_spec.get(buildername, {}).get('gtest_tests', [])]
 
   for test in get_tests(api):
     args = test.get('args', [])
     if test['shard_index'] != 0 or test['total_shards'] != 1:
       args.extend(['--test-launcher-shard-index=%d' % test['shard_index'],
                    '--test-launcher-total-shards=%d' % test['total_shards']])
     use_swarming = False
     swarming_shards = 1
     swarming_dimension_sets = None
+    swarming_priority = None
+    swarming_expiration = None
     if enable_swarming:
       swarming_spec = test.get('swarming', {})
       if swarming_spec.get('can_use_on_swarming_builders'):
         use_swarming = True
         swarming_shards = swarming_spec.get('shards', 1)
         swarming_dimension_sets = swarming_spec.get('dimension_sets')
+        swarming_priority = api.swarming.parse_priority_adjustment(
+            swarming_spec.get('priority_adjustment'))
+        swarming_expiration = swarming_spec.get('expiration')
     override_compile_targets = test.get('override_compile_targets', None)
     target_name = str(test['test'])
     name = str(test.get('name', target_name))
     swarming_dimensions = swarming_dimensions or {}
     use_xvfb = test.get('use_xvfb', True)
     if use_swarming and swarming_dimension_sets:
       for dimensions in swarming_dimension_sets:
         # Yield potentially multiple invocations of the same test, on
         # different machine configurations.
         new_dimensions = dict(swarming_dimensions)
         new_dimensions.update(dimensions)
         yield GTestTest(name, args=args, target_name=target_name,
                         flakiness_dash=True,
                         enable_swarming=True,
                         swarming_shards=swarming_shards,
                         swarming_dimensions=new_dimensions,
+                        swarming_priority=swarming_priority,
+                        swarming_expiration=swarming_expiration,
                         override_compile_targets=override_compile_targets,
                         use_xvfb=use_xvfb)
     else:
       yield GTestTest(name, args=args, target_name=target_name,
                       flakiness_dash=True,
                       enable_swarming=use_swarming,
                       swarming_dimensions=swarming_dimensions,
                       swarming_shards=swarming_shards,
+                      swarming_priority=swarming_priority,
+                      swarming_expiration=swarming_expiration,
                       override_compile_targets=override_compile_targets,
                       use_xvfb=use_xvfb)
 
 
 def generate_instrumentation_test(api, chromium_tests_api, mastername,
                                   buildername, test_spec, bot_update_step,
                                   enable_swarming=False,
                                   swarming_dimensions=None,
                                   scripts_compile_targets=None):
   for test in test_spec.get(buildername, {}).get('instrumentation_tests', []):
     test_name = str(test.get('test'))
     use_swarming = False
     if enable_swarming:
       swarming_spec = test.get('swarming', {})
       if swarming_spec.get('can_use_on_swarming_builders'):
         use_swarming = True
         swarming_shards = swarming_spec.get('shards', 1)
         swarming_dimension_sets = swarming_spec.get('dimension_sets')
+        swarming_priority = api.swarming.parse_priority_adjustment(
+            swarming_spec.get('priority_adjustment'))
+        swarming_expiration = swarming_spec.get('expiration')
     if use_swarming and swarming_dimension_sets:
       for dimensions in swarming_dimension_sets:
         # TODO(stip): Swarmify instrumentation tests
         pass
     else:
       yield AndroidInstrumentationTest(
           test_name,
           compile_targets=test.get('override_compile_targets', None),
           isolate_file_path=test.get('isolate_file_path', None),
           apk_under_test=test.get('apk_under_test', None),
@@ skipping 107 matching lines @@
     if failure:
       raise failure
 
   @staticmethod
   def compile_targets(_):
     return []
 
 
 class SwarmingTest(Test):
   def __init__(self, name, dimensions=None, tags=None, target_name=None,
-               extra_suffix=None):
+               extra_suffix=None, priority=None, expiration=None):
     self._name = name
     self._tasks = {}
     self._results = {}
     self._target_name = target_name
     self._dimensions = dimensions
     self._tags = tags
     self._extra_suffix = extra_suffix
+    self._priority = priority
+    self._expiration = expiration
     if dimensions and not extra_suffix:
       self._extra_suffix = self._get_gpu_suffix(dimensions)
 
   @staticmethod
   def _get_gpu_suffix(dimensions):
     if not dimensions.get('gpu'):
       return None
     gpu_vendor_id = dimensions.get('gpu', '').split(':')[0].lower()
     vendor_ids = {
       '8086': 'Intel',
@@ skipping 57 matching lines @@
           import sys
           print '*.isolated file for target %s is missing' % sys.argv[1]
           sys.exit(1)
           """,
           args=[self.isolate_target(api)])
 
     # Create task.
     self._tasks[suffix] = self.create_task(
         api, suffix, isolated_hash, test_filter=test_filter)
 
+    if self._priority:
+      self._tasks[suffix].priority = self._priority
+
+    if self._expiration:
+      self._tasks[suffix].expiration = self._expiration
+
     # Add custom dimensions.
     if self._dimensions:  # pragma: no cover
       #TODO(stip): concoct a test case that will trigger this codepath
       for k, v in self._dimensions.iteritems():
          if v is None:
            # Remove key if it exists. This allows one to use None to remove
            # default dimensions.
            self._tasks[suffix].dimensions.pop(k, None)
          else:
            self._tasks[suffix].dimensions[k] = v
@@ skipping 69 matching lines @@
     assert self.has_valid_results(api, suffix)
     return self._results[suffix]['failures']
 
   @property
   def uses_swarming(self):
     return True
 
 
 class SwarmingGTestTest(SwarmingTest):
   def __init__(self, name, args=None, target_name=None, shards=1,
-               dimensions=None, tags=None, extra_suffix=None,
-               upload_test_results=True, override_compile_targets=None):
+               dimensions=None, tags=None, extra_suffix=None, priority=None,
+               expiration=None, upload_test_results=True,
+               override_compile_targets=None):
     super(SwarmingGTestTest, self).__init__(name, dimensions, tags, target_name,
-                                            extra_suffix)
+                                            extra_suffix, priority, expiration)
     self._args = args or []
     self._shards = shards
     self._upload_test_results = upload_test_results
     self._override_compile_targets = override_compile_targets
 
   def compile_targets(self, api):
     # <X>_run target depends on <X>, and then isolates it invoking isolate.py.
     # It is a convention, not a hard coded rule.
     # Also include name without the _run suffix to help recipes correctly
     # interpret results returned by "analyze".
@@ skipping 369 matching lines @@
       return self._test_runs[suffix].json.output['valid']
     except Exception:  # pragma: no cover
       return False
 
   def failures(self, api, suffix):
     return self._test_runs[suffix].json.output['failures']
 
 
 class SwarmingIsolatedScriptTest(SwarmingTest):
   def __init__(self, name, args=None, target_name=None, shards=1,
-               dimensions=None, tags=None, extra_suffix=None,
-               upload_test_results=True, override_compile_targets=None):
+               dimensions=None, tags=None, extra_suffix=None, priority=None,
+               expiration=None, upload_test_results=True,
+               override_compile_targets=None):
     super(SwarmingIsolatedScriptTest, self).__init__(
-        name, dimensions, tags, target_name, extra_suffix)
+        name, dimensions, tags, target_name, extra_suffix, priority, expiration)
     self._args = args or []
     self._shards = shards
     self._upload_test_results = upload_test_results
     self._override_compile_targets = override_compile_targets
 
   @property
   def target_name(self):
     return self._target_name or self._name
 
   def compile_targets(self, _):
@@ skipping 39 matching lines @@
 
 
 def generate_isolated_script(api, chromium_tests_api, mastername, buildername,
                              test_spec, bot_update_step, enable_swarming=False,
                              swarming_dimensions=None,
                              scripts_compile_targets=None):
   for spec in test_spec.get(buildername, {}).get('isolated_scripts', []):
     use_swarming = False
     swarming_shards = 1
     swarming_dimension_sets = None
+    swarming_priority = None
+    swarming_expiration = None
     if enable_swarming:
       swarming_spec = spec.get('swarming', {})
       if swarming_spec.get('can_use_on_swarming_builders', False):
         use_swarming = True
         swarming_shards = swarming_spec.get('shards', 1)
         swarming_dimension_sets = swarming_spec.get('dimension_sets')
+        swarming_priority = api.swarming.parse_priority_adjustment(
+            swarming_spec.get('priority_adjustment'))
+        swarming_expiration = swarming_spec.get('expiration')
     name = str(spec['name'])
     # The variable substitution and precommit/non-precommit arguments
     # could be supported for the other test types too, but that wasn't
     # desired at the time of this writing.
     args = get_args_for_test(api, chromium_tests_api, spec, bot_update_step)
     target_name = spec['isolate_name']
     # This features is only needed for the cases in which the *_run compile
     # target is needed to generate isolate files that contains dynamically libs.
     # TODO(nednguyen, kbr): Remove this once all the GYP builds are converted
     # to GN.
     override_compile_targets = spec.get('override_compile_targets', None)
     swarming_dimensions = swarming_dimensions or {}
     if use_swarming:
       if swarming_dimension_sets:
         for dimensions in swarming_dimension_sets:
           # Yield potentially multiple invocations of the same test,
           # on different machine configurations.
           new_dimensions = dict(swarming_dimensions)
           new_dimensions.update(dimensions)
           yield SwarmingIsolatedScriptTest(
               name=name, args=args, target_name=target_name,
               shards=swarming_shards, dimensions=new_dimensions,
-              override_compile_targets=override_compile_targets)
+              override_compile_targets=override_compile_targets,
+              priority=swarming_priority, expiration=swarming_expiration)
       else:
         yield SwarmingIsolatedScriptTest(
             name=name, args=args, target_name=target_name,
             shards=swarming_shards, dimensions=swarming_dimensions,
-            override_compile_targets=override_compile_targets)
+            override_compile_targets=override_compile_targets,
+            priority=swarming_priority, expiration=swarming_expiration)
     else:
       yield LocalIsolatedScriptTest(
           name=name, args=args, target_name=target_name,
           override_compile_targets=override_compile_targets)
 
 
 class GTestTest(Test):
   def __init__(self, name, args=None, target_name=None, enable_swarming=False,
                swarming_shards=1, swarming_dimensions=None, swarming_tags=None,
-               swarming_extra_suffix=None, **runtest_kwargs):
+               swarming_extra_suffix=None, swarming_priority=None,
+               swarming_expiration=None, **runtest_kwargs):
     super(GTestTest, self).__init__()
     if enable_swarming:
       self._test = SwarmingGTestTest(
           name, args, target_name, swarming_shards, swarming_dimensions,
-          swarming_tags, swarming_extra_suffix,
-          override_compile_targets=runtest_kwargs.get(
+          swarming_tags, swarming_extra_suffix, swarming_priority,
+          swarming_expiration, override_compile_targets=runtest_kwargs.get(
             'override_compile_targets'))
     else:
       self._test = LocalGTestTest(name, args, target_name, **runtest_kwargs)
 
     self.enable_swarming = enable_swarming
 
   @property
   def name(self):
     return self._test.name
 
@@ skipping 581 matching lines @@
   def run(self, api, suffix, test_filter=None):
     api.chromium_android.coverage_report(upload=False)
     api.chromium_android.get_changed_lines_for_revision()
     api.chromium_android.incremental_coverage_report()
 
 
 GOMA_TESTS = [
   GTestTest('base_unittests'),
   GTestTest('content_unittests'),
 ]