Add recipe for split AMP/local CQ.

scripts/slave/recipe_modules/chromium/steps.py

Index: scripts/slave/recipe_modules/chromium/steps.py
diff --git a/scripts/slave/recipe_modules/chromium/steps.py b/scripts/slave/recipe_modules/chromium/steps.py
index e8c5273888ebef4103cafcae4548b798293b5362..a1111f98368e6fa1ef3c98ced5ddddd8ca3b6383 100644
--- a/scripts/slave/recipe_modules/chromium/steps.py
+++ b/scripts/slave/recipe_modules/chromium/steps.py
@@ -618,6 +618,77 @@ class SwarmingGTestTest(SwarmingTest):
     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,