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 586 matching lines @@
     gtest_results = step_result.test_utils.gtest_results
     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):
+  def __init__(self, name, args=None, target_name=None, **runtest_kwargs):
+    super(AMPGTestTest, self).__init__()
+    self._name = name
+    self._args = args
+    self._target_name = target_name
+    self._local_test = LocalGTestTest(name, args, target_name, **runtest_kwargs)
+
+  @property
+  def name(self):
+    return self._name
+
+  @property
+  def target_name(self):
+    return self._target_name or self._name
+
+  @property
+  def isolate_target(self):
+    return self.target_name  # pragma: no cover
+
+  def compile_targets(self, api):
+    return self._local_test.compile_targets(api)
+
+  def trigger(self, api, suffix):
+    isolate_file_path = (
+        api.path['checkout'].join(*isolate_file) if isolate_file else None)
+    deferred_trigger_result = api.amp.trigger_test_suite(
+        self._name, 'gtest',
+        api.amp.gtest_arguments(self._name,
+                                isolate_file_path=isolate_file_path),
+        api.amp.amp_arguments(
+            api_address=AMP_INSTANCE_ADDRESS,
+            api_port=AMP_INSTANCE_PORT,
+            api_protocol=AMP_INSTANCE_PROTOCOL,
+            device_minimum_os=builder.get('device_minimum_os'),
+            device_name=builder.get('device_name'),
+            device_os=builder.get('device_os'),
+            device_timeout=builder.get('device_timeout')))
+    self._trigger_successful = deferred_trigger_result.is_ok
+
+  def collect(self, api, suffix):
+    # If the AMP test was not successfully triggered, run locally.
+    if not self._trigger_successful:
+      return self._local_test.run(api, suffix)
+    else:
+      deffered_step_result = api.amp.collect_test_suite(

[jbudorick, 2015/04/22 16:35:40]

deferred_step_result

I'm somewhat surprised that this made it through simulation, since you spelled
it correctly below.

[navabi, 2015/05/07 19:21:59]

Done.

+        suite, 'gtest',
+        api.amp.gtest_arguments(suite),
+        api.amp.amp_arguments(
+          api_address=AMP_INSTANCE_ADDRESS,
+          api_port=AMP_INSTANCE_PORT,
+          api_protocol=AMP_INSTANCE_PROTOCOL,
+          device_minimum_os=builder.get('device_minimum_os'),
+          device_name=builder.get('device_name'),
+          device_os=builder.get('device_os'),
+          device_timeout=builder.get('device_timeout')))
+      self._collect_successful = deferred_step_result.is_ok
+      # Need to return step_result here

[navabi, 2015/04/22 16:16:48]

need to figure out how to return the step result here. This needs to return the
same info that the LocalGTestTest::run function returns. (help from jbudorick@)

+
+  def run(self, api, suffix):
+    self._trigger(api, suffix)

[navabi, 2015/04/22 16:16:48]

I will change this to return self._local_tests.run(). Thus, the AMPGTestTest.run
method can be used to run the test sequentially as if it was not an AMP test.
This will be used when triggering failed (e.g. because devices are not
available) or when we are re-running a test without the patch.

[Paweł Hajdan Jr., 2015/04/22 16:25:09]

Please take a look at how swarming tests run. There are pre_run and post_run
hooks, which can be used for trigger and collect respectively. We don't fallback
to local runs if trigger fails.

[navabi, 2015/05/07 19:21:59]

Done.

+    result = self._collect(api, suffix)
+    return result
+
+  def has_valid_results(self, api, suffix):
+    # TODO(navabi): implement this.
+    return True
+
+  def failures(self, api, suffix):
+    # TODO(navabi): implement this.
+    return []
+
 
 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)
@@ 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'),
 ]