recipe_modules/testsharder/api.py - infra/recipes - Git at Google

 # Copyright 2018 The Fuchsia Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 """
 Recipe module that wraps the testsharder tool, which searches a Fuchsia
 build for test specifications and groups them into shards.

 The tool assigns a unique name to each produced shard.

 Testsharder tool:
 https://fuchsia.googlesource.com/fuchsia/+/refs/heads/master/tools/integration/testsharder/cmd/
 """

 import attr

 from recipe_engine import recipe_api


 def assert_list_of_strings(_, __, lst):
   assert all(isinstance(item, basestring) for item in lst)


 @attr.s(frozen=True)
 class Test(object):
   """Represents a test defined within the Fuchsia project.

     Args:
       name (str): The name of the test.
       label (str): The GN label for the test target.
       os (str): The OS on which the test is intended to run.
       package_url (str): The package URL for the test. This field only makes
         sense for Fuchsia tests.
       path (str): The path to the test. If unset, `command` must be set.
       command (list(str)): The command used to invoke the test.
       deps (list(str)): The runtime dependencies of the test given as
         relative paths within the build directory. This field only makes sense
         for Linux and Mac tests.
   """

   name = attr.ib(validator=attr.validators.instance_of(basestring))
   os = attr.ib(validator=attr.validators.in_(('fuchsia', 'linux', 'mac')))
   # TODO(olivernewman): make label required
   label = attr.ib('', validator=attr.validators.instance_of(basestring))
   package_url = attr.ib('', validator=attr.validators.instance_of(basestring))
   path = attr.ib('', validator=attr.validators.instance_of(basestring))
   command = attr.ib(factory=list, validator=assert_list_of_strings)
   deps = attr.ib(factory=list, validator=assert_list_of_strings)

   @classmethod
   def from_jsonish(cls, jsond):
     """Creates a new Test from a JSON-compatible dict."""
     # TODO(olivernewman): We have to accept install_path and deps_file for
     # backwards compatibility. Revert change
     # Ibe606b6e62aad751cd2b43ab4df758d984dc34b7 after backwards compatibility
     # is no longer needed.
     deprecated_fields = {
         'install_path': 'path',
         'deps_file': 'deps',
     }
     for old_field, new_field in deprecated_fields.items():
       if old_field in jsond:
         jsond[new_field] = jsond.pop(old_field)
     valid_kwargs = attr.fields_dict(cls)
     kwargs = {k: v for k, v in jsond.items() if k in valid_kwargs}
     return cls(**kwargs)

   def render_to_jsonish(self):
     """Returns a JSON-compatible dict representing the Test.

     The format follows the format of testsharder.Test found here:
     https://fuchsia.googlesource.com/fuchsia/+/master/tools/integration/testsharder/test_spec.go
     """
     ret = {
         'name': self.name,
         'label': self.label,
         'os': self.os,
         'path': self.path,
         'install_path': self.path,  # TODO(olivernewman): Delete
     }
     if self.command:
       ret['command'] = self.command
     if self.package_url:
       ret['package_url'] = self.package_url
     return ret


 class Shard(object):
   """Represents a shard of several tests with one common environment."""

   @classmethod
   def from_jsonish(cls, jsond):
     """Creates a new Shard from a JSON-compatible Python dict."""
     return cls(
         name=jsond['name'],
         tests=[Test.from_jsonish(test) for test in jsond['tests']],
         dimensions=jsond['environment']['dimensions'],
         service_account=jsond['environment'].get('service_account', ''),
         deps=jsond.get('deps', ()),
         netboot=jsond['environment'].get('netboot', False),
     )

   def __init__(self,
                name,
                tests,
                dimensions,
                service_account='',
                deps=(),
                netboot=False):
     """Initializes a Shard.

     Args:
       name (str): The name of the shard.
       tests (seq[Test]): A sequence of tests.
       dimensions (dict[str]str): A dictionary of Swarming dimensions, mapping
         name to value.
       service_account (str): A service account to be attached to a swarming
         task when this shard is run.
       deps (seq(str)): The runtime dependencies of the tests in the shard
         given as relative paths within the build directory. This field
         only makes sense for Linux and Mac tests.
       netboot (bool): Whether to netboot instead of paving before running the
         tests.
     """
     assert isinstance(name, str)
     self._name = name
     self._tests = tests
     self._dimensions = {}
     for k, v in dimensions.iteritems():
       assert isinstance(k, str)
       self._dimensions[k] = str(v)
     self._service_account = service_account
     self._netboot = netboot

     # TODO(fxbug.dev/10411): over the course of this bug, the testsharder will
     # be refactored to have deps as a top-level field in the shard json.
     # This synthesizes that for a soft transition; after it is completed, just
     # set self._deps as deps.
     aggregated_deps = set(deps)
     if not deps:
       for test in tests:
         aggregated_deps.update(test.deps)
     self._deps = tuple(aggregated_deps)

   @property
   def name(self):
     """The name (str) of the shard."""
     return self._name

   @property
   def tests(self):
     """The list of tests (list(Test)) in the shard."""
     return self._tests

   @property
   def dimensions(self):
     """The Swarming dimensions (dict[str]str) specified by the shard."""
     return self._dimensions.copy()

   def __getattr__(self, dimension):
     """Returns the value of a given dimension is set in the shard; else, None."""
     return self._dimensions.get(dimension)

   @property
   def targets_fuchsia(self):
     """Returns whether the shard of tests is meant to run on or against fuchsia."""
     return self.os not in ['Linux', 'Mac']

   @property
   def service_account(self):
     """Returns any service account (string) to be attached to a swarming
       task when this shard is run. Empty if unset."""
     return self._service_account

   @property
   def deps(self):
     """The runtime dependencies (tuple(str)) of the test.

     The dependencies are given as relative paths within the build directory and
     are only specified when test's OS is Linux or Mac."""
     return self._deps

   @property
   def netboot(self):
     """Return whether to netboot instead of paving before running the tests."""
     return self._netboot

   def render_to_jsonish(self):
     """Returns a JSON-compatible dict representing the Shard.

     The format follows the format os testsharder.Shard found here:
     https://fuchsia.googlesource.com/fuchsia/+/master/tools/integration/testsharder/shard.go
     """
     ret = {
         'name': self.name,
         'tests': [test.render_to_jsonish() for test in self.tests],
         'environment': {
             'dimensions': self.dimensions
         },
         'deps': self.deps,
     }
     if self.service_account:
       ret['environment']['service_account'] = self.service_account
     if self.netboot:
       ret['environment']['netboot'] = self.netboot
     return ret


 class TestsharderApi(recipe_api.RecipeApi):
   """Module for interacting with the Testsharder tool.

   The testsharder tool accepts a set of test specifications and produces
   a file containing shards of execution.
   """
   Test = Test
   Shard = Shard

   def __init__(self, *args, **kwargs):
     super(TestsharderApi, self).__init__(*args, **kwargs)
     self._testsharder_path = None

   def execute(self,
               step_name,
               testsharder_path,
               build_dir,
               max_shard_size=None,
               target_duration_secs=0,
               mode=None,
               multipliers=None,
               output_file=None,
               tags=()):
     """Executes the testsharder tool.

     Args:
       step_name (str): name of the step.
       testsharder_path (Path): path to the testsharder tool.
       build_dir (Path): path to a Fuchsia build directory root, in
         which GN has been run (ninja need not have been executed).
       max_shard_size (long or None):  Additional shards will be created if needed
         to keep the number of tests per shard <= this number.
       target_duration_secs (int): If >0, testsharder will try to produce
         shards of approximately this duration.
       mode (str or None): mode in which the testsharder may run (e.g.,
         normal or restricted); if None, the default normal mode will
         be used.
         See https://fuchsia.googlesource.com/fuchsia/+/master/tools/integration/testsharder/mode.go
         for more details
       multipliers (Path or None): path of a json manifest specifying tests to run
         multiple times; if supplied, new shards will be created to run them.
       output_file (Path): optional file path to leak output to.
       tags (list(str)): tags on which to filter test specs.

     Returns:
       A list of Shards, each representing one test shard.
     """
     cmd = [
         testsharder_path,
         '-build-dir',
         build_dir,
         '-output-file',
         self.m.json.output(leak_to=output_file),
     ]
     if max_shard_size:
       cmd += ['-max-shard-size', max_shard_size]
     if target_duration_secs:
       cmd += ['-target-duration-secs', target_duration_secs]
     if multipliers:
       cmd += ['-multipliers', multipliers]
     for tag in tags:
       cmd += ['-tag', tag]
     if mode:
       cmd += ['-mode', mode]
     result = self.m.step(step_name, cmd).json.output
     return [Shard.from_jsonish(shard) for shard in result]
	# Copyright 2018 The Fuchsia Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.
	"""
	Recipe module that wraps the testsharder tool, which searches a Fuchsia
	build for test specifications and groups them into shards.

	The tool assigns a unique name to each produced shard.

	Testsharder tool:
	https://fuchsia.googlesource.com/fuchsia/+/refs/heads/master/tools/integration/testsharder/cmd/
	"""

	import attr

	from recipe_engine import recipe_api


	def assert_list_of_strings(_, __, lst):
	assert all(isinstance(item, basestring) for item in lst)


	@attr.s(frozen=True)
	class Test(object):
	"""Represents a test defined within the Fuchsia project.

	Args:
	name (str): The name of the test.
	label (str): The GN label for the test target.
	os (str): The OS on which the test is intended to run.
	package_url (str): The package URL for the test. This field only makes
	sense for Fuchsia tests.
	path (str): The path to the test. If unset, `command` must be set.
	command (list(str)): The command used to invoke the test.
	deps (list(str)): The runtime dependencies of the test given as
	relative paths within the build directory. This field only makes sense
	for Linux and Mac tests.
	"""

	name = attr.ib(validator=attr.validators.instance_of(basestring))
	os = attr.ib(validator=attr.validators.in_(('fuchsia', 'linux', 'mac')))
	# TODO(olivernewman): make label required
	label = attr.ib('', validator=attr.validators.instance_of(basestring))
	package_url = attr.ib('', validator=attr.validators.instance_of(basestring))
	path = attr.ib('', validator=attr.validators.instance_of(basestring))
	command = attr.ib(factory=list, validator=assert_list_of_strings)
	deps = attr.ib(factory=list, validator=assert_list_of_strings)

	@classmethod
	def from_jsonish(cls, jsond):
	"""Creates a new Test from a JSON-compatible dict."""
	# TODO(olivernewman): We have to accept install_path and deps_file for
	# backwards compatibility. Revert change
	# Ibe606b6e62aad751cd2b43ab4df758d984dc34b7 after backwards compatibility
	# is no longer needed.
	deprecated_fields = {
	'install_path': 'path',
	'deps_file': 'deps',
	}
	for old_field, new_field in deprecated_fields.items():
	if old_field in jsond:
	jsond[new_field] = jsond.pop(old_field)
	valid_kwargs = attr.fields_dict(cls)
	kwargs = {k: v for k, v in jsond.items() if k in valid_kwargs}
	return cls(**kwargs)

	def render_to_jsonish(self):
	"""Returns a JSON-compatible dict representing the Test.

	The format follows the format of testsharder.Test found here:
	https://fuchsia.googlesource.com/fuchsia/+/master/tools/integration/testsharder/test_spec.go
	"""
	ret = {
	'name': self.name,
	'label': self.label,
	'os': self.os,
	'path': self.path,
	'install_path': self.path, # TODO(olivernewman): Delete
	}
	if self.command:
	ret['command'] = self.command
	if self.package_url:
	ret['package_url'] = self.package_url
	return ret


	class Shard(object):
	"""Represents a shard of several tests with one common environment."""

	@classmethod
	def from_jsonish(cls, jsond):
	"""Creates a new Shard from a JSON-compatible Python dict."""
	return cls(
	name=jsond['name'],
	tests=[Test.from_jsonish(test) for test in jsond['tests']],
	dimensions=jsond['environment']['dimensions'],
	service_account=jsond['environment'].get('service_account', ''),
	deps=jsond.get('deps', ()),
	netboot=jsond['environment'].get('netboot', False),
	)

	def __init__(self,
	name,
	tests,
	dimensions,
	service_account='',
	deps=(),
	netboot=False):
	"""Initializes a Shard.

	Args:
	name (str): The name of the shard.
	tests (seq[Test]): A sequence of tests.
	dimensions (dict[str]str): A dictionary of Swarming dimensions, mapping
	name to value.
	service_account (str): A service account to be attached to a swarming
	task when this shard is run.
	deps (seq(str)): The runtime dependencies of the tests in the shard
	given as relative paths within the build directory. This field
	only makes sense for Linux and Mac tests.
	netboot (bool): Whether to netboot instead of paving before running the
	tests.
	"""
	assert isinstance(name, str)
	self._name = name
	self._tests = tests
	self._dimensions = {}
	for k, v in dimensions.iteritems():
	assert isinstance(k, str)
	self._dimensions[k] = str(v)
	self._service_account = service_account
	self._netboot = netboot

	# TODO(fxbug.dev/10411): over the course of this bug, the testsharder will
	# be refactored to have deps as a top-level field in the shard json.
	# This synthesizes that for a soft transition; after it is completed, just
	# set self._deps as deps.
	aggregated_deps = set(deps)
	if not deps:
	for test in tests:
	aggregated_deps.update(test.deps)
	self._deps = tuple(aggregated_deps)

	@property
	def name(self):
	"""The name (str) of the shard."""
	return self._name

	@property
	def tests(self):
	"""The list of tests (list(Test)) in the shard."""
	return self._tests

	@property
	def dimensions(self):
	"""The Swarming dimensions (dict[str]str) specified by the shard."""
	return self._dimensions.copy()

	def __getattr__(self, dimension):
	"""Returns the value of a given dimension is set in the shard; else, None."""
	return self._dimensions.get(dimension)

	@property
	def targets_fuchsia(self):
	"""Returns whether the shard of tests is meant to run on or against fuchsia."""
	return self.os not in ['Linux', 'Mac']

	@property
	def service_account(self):
	"""Returns any service account (string) to be attached to a swarming
	task when this shard is run. Empty if unset."""
	return self._service_account

	@property
	def deps(self):
	"""The runtime dependencies (tuple(str)) of the test.

	The dependencies are given as relative paths within the build directory and
	are only specified when test's OS is Linux or Mac."""
	return self._deps

	@property
	def netboot(self):
	"""Return whether to netboot instead of paving before running the tests."""
	return self._netboot

	def render_to_jsonish(self):
	"""Returns a JSON-compatible dict representing the Shard.

	The format follows the format os testsharder.Shard found here:
	https://fuchsia.googlesource.com/fuchsia/+/master/tools/integration/testsharder/shard.go
	"""
	ret = {
	'name': self.name,
	'tests': [test.render_to_jsonish() for test in self.tests],
	'environment': {
	'dimensions': self.dimensions
	},
	'deps': self.deps,
	}
	if self.service_account:
	ret['environment']['service_account'] = self.service_account
	if self.netboot:
	ret['environment']['netboot'] = self.netboot
	return ret


	class TestsharderApi(recipe_api.RecipeApi):
	"""Module for interacting with the Testsharder tool.

	The testsharder tool accepts a set of test specifications and produces
	a file containing shards of execution.
	"""
	Test = Test
	Shard = Shard

	def __init__(self, args, *kwargs):
	super(TestsharderApi, self).__init__(args, *kwargs)
	self._testsharder_path = None

	def execute(self,
	step_name,
	testsharder_path,
	build_dir,
	max_shard_size=None,
	target_duration_secs=0,
	mode=None,
	multipliers=None,
	output_file=None,
	tags=()):
	"""Executes the testsharder tool.

	Args:
	step_name (str): name of the step.
	testsharder_path (Path): path to the testsharder tool.
	build_dir (Path): path to a Fuchsia build directory root, in
	which GN has been run (ninja need not have been executed).
	max_shard_size (long or None): Additional shards will be created if needed
	to keep the number of tests per shard <= this number.
	target_duration_secs (int): If >0, testsharder will try to produce
	shards of approximately this duration.
	mode (str or None): mode in which the testsharder may run (e.g.,
	normal or restricted); if None, the default normal mode will
	be used.
	See https://fuchsia.googlesource.com/fuchsia/+/master/tools/integration/testsharder/mode.go
	for more details
	multipliers (Path or None): path of a json manifest specifying tests to run
	multiple times; if supplied, new shards will be created to run them.
	output_file (Path): optional file path to leak output to.
	tags (list(str)): tags on which to filter test specs.

	Returns:
	A list of Shards, each representing one test shard.
	"""
	cmd = [
	testsharder_path,
	'-build-dir',
	build_dir,
	'-output-file',
	self.m.json.output(leak_to=output_file),
	]
	if max_shard_size:
	cmd += ['-max-shard-size', max_shard_size]
	if target_duration_secs:
	cmd += ['-target-duration-secs', target_duration_secs]
	if multipliers:
	cmd += ['-multipliers', multipliers]
	for tag in tags:
	cmd += ['-tag', tag]
	if mode:
	cmd += ['-mode', mode]
	result = self.m.step(step_name, cmd).json.output
	return [Shard.from_jsonish(shard) for shard in result]