testing/trigger_scripts/chromeos_device_trigger.py

#!/usr/bin/env python
# Copyright 2019 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.

"""Custom swarming trigger script for ChromeOS device tests.

CrOS device tests are unique in that the device OS they prefer to run on is
continuously changing. The LKGM file, checked into src at
//chromeos/CHROMEOS_LKGM, represents the ChromeOS version Chrome's ToT aims
to be compatible with. Therefore, a CrOS test for Chrome ideally targets a
device running the LKGM.

Since the LKGM file gets updated frequently (~daily), we can't reasonably
hardcode the LKGM in the test specs. So this special trigger script will read
the current LKGM (at the time of trigger) and append that to the task's
dimensions. If such a device isn't available in time, the task will fallback
to one running any OS.
"""

import argparse
import os
import re
import sys

import base_test_triggerer


SRC_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(
  __file__))))
LKGM_FILE_PATH = os.path.join(SRC_DIR, 'chromeos', 'CHROMEOS_LKGM')
# Should match something that looks like "12345.0.0".
LKGM_RE = re.compile(r'\d+\.\d+\.\d+')


def read_current_lkgm():
  if not os.path.exists(LKGM_FILE_PATH):
    sys.stderr.write('LKGM file not present at %s\n' % LKGM_FILE_PATH)
    return None

  with open(LKGM_FILE_PATH) as f:
    lkgm = f.read().strip()

  if not LKGM_RE.match(lkgm):
    sys.stderr.write('Unknown format of LKGM: %s\n' % lkgm)
    return None

  # Just the major version should be sufficient.
  return lkgm.split('.')[0]


def parse_args(triggerer):
  # This script will do nothing but inspect and tweak the dimension args to
  # `swarming.py trigger`. So let's pull just those out.
  parser = argparse.ArgumentParser(description=__doc__)
  parser.add_argument(
      '-d', '--dimension', default=[], action='append', nargs=2,
      dest='dimensions',
      help='Dimensions to filter on. Duplicated from the `swarming.py trigger` '
           'command. Parsed here to ensure `device_os` is not added.'
  )
  parser.add_argument(
      '--optional-dimension', default=[], action='append', nargs=3,
      dest='optional_dimensions',
      help='Optional dimensions which will result in additional task slices. '
           'Duplicated from the `swarming.py trigger` command.')
  parser.add_argument(
      '--primary-expiration', type=int, default=600,
      help='How long to wait (in seconds) for an available bot in the primary '
           'task slice.')
  # BaseTestTriggerer's setup_parser_contract() takes care of adding needed
  # swarming.py args if they're not already present. But only do this if
  # '--shard-index' is passed in. (The exact usage of trigger scripts are
  # currently changing. See crbug.com/926987 for more info.)
  if '--shard-index' in sys.argv:
    base_test_triggerer.BaseTestTriggerer.setup_parser_contract(parser)
    args, additional_args = parser.parse_known_args()
    additional_args = triggerer.modify_args(
        additional_args, 0, args.shard_index, args.shards, args.dump_json)
  else:
    args, additional_args = parser.parse_known_args()

  if additional_args[0] != 'trigger':
    parser.error(
        'This script is only supported for `swarming.py trigger` invocations.')

  for k, _ in args.dimensions:
    if k == 'device_os':
      parser.error(
          'Must not specify the device_os dimension when using this script. '
          '(It will be added automatically.)')

  # It might be a valid use-case to include optional-dimensions in the initial
  # invocation. But it'd be difficult to integrate them into what we're doing
  # here. So let's just ensure there aren't any.
  if args.optional_dimensions:
    parser.error(
        'Must not specify optional dimensions when using this script.')

  return args, additional_args


def main():
  triggerer = base_test_triggerer.BaseTestTriggerer()
  args, additional_args = parse_args(triggerer)

  current_lkgm = read_current_lkgm()
  if not current_lkgm:
    return 1

  new_args = additional_args[:1]
  # Insert our modified dimension args in between the 1st and 2nd args of the
  # initial `swarming.py` invocation. This avoids the presence of the special
  # `--` arg from causing swarming.py to ignore them.
  needs_device_status = True
  for k, v in args.dimensions:
    new_args.extend(['--dimension', k, v])
    if k == 'device_status':
      needs_device_status = False

  # Only CrOS device bots with a device_status dimension of "available" should
  # run tests. So target those explicitly if we aren't already.
  if needs_device_status:
    new_args.extend(['--dimension', 'device_status', 'available'])

  new_args.extend([
      '--optional-dimension',
      'device_os',
      current_lkgm,
      str(args.primary_expiration),
  ])
  new_args += additional_args[1:]

  return triggerer.run_swarming(new_args, True)


if __name__ == '__main__':
  sys.exit(main())