recipe_modules/testsharder/examples/full.py

# Copyright 2018 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 recipe_engine.recipe_api import Property

DEPS = [
    'fuchsia/status_check',
    'fuchsia/testsharder',
    'recipe_engine/path',
    'recipe_engine/properties',
]

PROPERTIES = {
    'multipliers':
        Property(
            kind=str,
            help='path to the json file that specifies how many times to run a test',
            default=None),
}


def RunSteps(api, multipliers):
  # Run the test sharder.
  shards = api.testsharder.execute(
      'shard test specs',
      testsharder_path='path/to/testsharder',
      build_dir=api.path['start_dir'].join('out'),
      # It's actually invalid to pass both `max_shard_size` and
      # `target_duration_secs` to the testshadder executable; we just do it
      # here for code coverage.
      max_shard_size=200,
      target_duration_secs=10 * 60,
      mode='restricted',
      multipliers=multipliers,
      output_file=api.path['start_dir'].join('leak_output_here'),
      tags=('one-tag', 'two-tag', 'red-tag', 'blue-tag'),
  )

  # One may access a number of different aspects of the shard easily as it is a
  # Python object.
  # pylint: disable=pointless-statement
  for shard in shards:
    shard.name
    shard.deps
    # One may access all of the underlying swarming dimensions...
    shard.dimensions
    # ...or any of them individually.
    shard.device_type
    shard.os
    shard.any_swarming_dimension
    shard.targets_fuchsia
    for test in shard.tests:
      test.name
      test.path
      test.os
      test.command
  # pylint: enable=pointless-statement


def GenTests(api):

  def step_data(name, multiply=False):
    shards = [
        api.testsharder.shard(
            name='QEMU',
            tests=[api.testsharder.test('test1')],
            dimensions=dict(device_type='QEMU'),
            service_account='myacct@example.iam.gserviceaccount.com'),
        api.testsharder.shard(
            name='Linux',
            tests=[api.testsharder.test('test2', os='linux')],
            dimensions=dict(os='Linux'),
        ),
        api.testsharder.shard(
            name='NUC-netboot',
            tests=[api.testsharder.test('test2', os='linux')],
            dimensions=dict(device_type='NUC'),
            netboot=True,
        ),
    ]
    if multiply:
      shards += [
          api.testsharder.shard(
              name='QEMU - test1',
              tests=[api.testsharder.test('test1') for _ in range(2)],
              dimensions=dict(device_type='QEMU'),
              service_account='myacct@example.iam.gserviceaccount.com'),
          api.testsharder.shard(
              name='Linux - test2',
              tests=[
                  api.testsharder.test('test2', os='linux') for _ in range(2)
              ],
              dimensions=dict(os='Linux'),
          ),
      ]
    return api.testsharder.execute(step_name=name, shards=shards)

  yield (api.status_check.test('basic') + step_data('shard test specs'))

  yield (api.status_check.test('shard_with_multiplied_tests') +
         api.properties(multipliers='test_multipliers.json') +
         step_data('shard test specs', multiply=True))