Add recipe for split AMP/local CQ.

scripts/slave/recipe_modules/chromium/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 600 matching lines @@
     if not gtest_results:
       return False, None  # pragma: no cover
 
     global_tags = gtest_results.raw.get('global_tags', [])
     if 'UNRELIABLE_RESULTS' in global_tags:
       return False, None  # pragma: no cover
 
     return True, gtest_results.failures
 
 
+class AMPGTestTest(Test):
+  AMP_INSTANCE_ADDRESS = '172.22.21.180'
+  AMP_INSTANCE_PORT = '80'
+  AMP_INSTANCE_PROTOCOL = 'http'
+  AMP_RESULTS_BUCKET = 'chrome-amp-results'
+  def __init__(self, name, args=None, target_name=None, device_name=['Nexus 5'],
+               device_os=['4.4.2', '4.4.3'], device_timeout=60,
+               android_isolate_path=None, **runtest_kwargs):
+    self._name = name
+    self._args = args
+    self._target_name = target_name
+    self._android_isolate_path = android_isolate_path
+    # LocalGTestTest is used when AMP tests are not triggered successfully.
+    self._local_test = LocalGTestTest(name, args, target_name, **runtest_kwargs)
+    self._device_name = device_name
+    self._device_os = device_os
+    self._device_timeout = device_timeout
+    self._trigger_successful = None
+    self._failures = []
+
+  @property
+  def name(self):
+    return self._name
+
+  def run(self, api, suffix):  # pylint: disable=R0201
+    """Not used. All logic in pre_run, post_run."""
+    return []
+
+  def pre_run(self, api, suffix):
+    """Triggers an AMP test."""
+    amp_arguments = api.amp.amp_arguments(
+        api_address=AMPGTestTest.AMP_INSTANCE_ADDRESS,
+        api_port=AMPGTestTest.AMP_INSTANCE_PORT,
+        api_protocol=AMPGTestTest.AMP_INSTANCE_PROTOCOL,
+        device_name=self._device_name,
+        device_os=self._device_os,
+        device_timeout=self._device_timeout)
+
+    isolate_file_path = (api.path['checkout'].join(self._android_isolate_path)
+                         if self._android_isolate_path else None)
+    try:
+      api.amp.trigger_test_suite(
+          self._name, 'gtest',
+          api.amp.gtest_arguments(self._name,
+                                  isolate_file_path=isolate_file_path),
+          amp_arguments)
+      self._trigger_successful = True
+    except api.step.StepFailure:
+      self._trigger_successful = False
+
+  def post_run(self, api, suffix):
+    # If we were unable to successfully trigger the AMP job, run locally.
+    if not self._trigger_successful:
+      return self._local_test.run(api, suffix)
+    else:
+      amp_arguments = api.amp.amp_arguments(
+          api_address=AMPGTestTest.AMP_INSTANCE_ADDRESS,
+          api_port=AMPGTestTest.AMP_INSTANCE_PORT,
+          api_protocol=AMPGTestTest.AMP_INSTANCE_PROTOCOL,
+          device_name=self._device_name,
+          device_os=self._device_os,
+          device_timeout=self._device_timeout)
+
+      try:
+        api.amp.collect_test_suite(
+            self._name, 'gtest', api.amp.gtest_arguments(self._name),
+            amp_arguments)
+      except api.step.StepFailure as f:
+        self._failures = [str(f)]
+
+
 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):
     super(GTestTest, self).__init__()
     if enable_swarming:
       self._test = SwarmingGTestTest(
           name, args, target_name, swarming_shards, swarming_dimensions,
           swarming_tags, swarming_extra_suffix)
     else:
@@ skipping 611 matching lines @@
   GTestTest('ui_base_unittests'),
   GTestTest('ui_ios_unittests'),
   GTestTest('sync_unit_tests'),
   GTestTest('sql_unittests'),
 ]
 
 GOMA_TESTS = [
   GTestTest('base_unittests'),
   GTestTest('content_unittests'),
 ]