123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135 |
- #!/usr/bin/env python3
- # 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+')
- PRIMARY_SLICE_EXPIRATION_S = 300
- 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.')
- 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)
- 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=%s:%d' % (current_lkgm, PRIMARY_SLICE_EXPIRATION_S),
- ])
- new_args += additional_args[1:]
- return triggerer.run_swarming_go(new_args, args.dump_json,
- args.shard_index or 0, args.shards)
- if __name__ == '__main__':
- sys.exit(main())
|