cbuildbot/stages/tast_test_stages.py - mirrors/cros/chromiumos/chromite - Git at Google

 # -*- coding: utf-8 -*-
 # Copyright 2017 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 """Module containing stages for running Tast integration tests.

 This module contains cbuildbot test stages that run Tast integration tests. See
 https://chromium.googlesource.com/chromiumos/platform/tast/ for more details.
 """

 from __future__ import print_function

 import json
 import os
 import shutil

 from chromite.cbuildbot import commands
 from chromite.cbuildbot.stages import generic_stages
 from chromite.lib import cgroups
 from chromite.lib import constants
 from chromite.lib import cros_build_lib
 from chromite.lib import cros_logging as logging
 from chromite.lib import failures_lib
 from chromite.lib import osutils
 from chromite.lib import timeout_util


 # Error format string used when tast exits with a non-zero exit code.
 FAILURE_EXIT_CODE = '** Tests failed with code %d **'

 # Error format string used when one or more tests fail.
 FAILURE_TESTS_FAILED = '** %d test(s) failed **'

 # Error format string used when results dir is missing or empty.
 FAILURE_NO_RESULTS = '** No results in %s **'

 # Error format string used when results file is unreadable.
 FAILURE_BAD_RESULTS = '** Failed to read results from %s: %s **'

 # Prefix for results download link.
 RESULTS_LINK_PREFIX = 'results: '

 # Prefix for results link for informational tests.
 INFORMATIONAL_PREFIX = 'informational: '

 # Name of JSON file containing individual tests' results written by the tast
 # command to the results dir.
 RESULTS_FILENAME = 'results.json'

 # Names of properties in test objects from results JSON files.
 RESULTS_NAME_KEY = 'name'
 RESULTS_ERRORS_KEY = 'errors'
 RESULTS_ATTR_KEY = 'attr'

 # Attribute used to label informational tests.
 RESULTS_INFORMATIONAL_ATTR = 'informational'

 # Directory written within main results dir by tast command containing per-test
 # results.
 RESULTS_TESTS_DIR = 'tests'


 def _CopyResultsDir(src, dest):
   """Copies a results dir to a new directory for archiving.

   Args:
     src: String source path.
     dest: String destination path (presumably under
           generic_stages.ArchivingStageMixin.archive_path). Must not exist
           already.

   Raises:
     OSError if dest already exists or the copy fails.
   """
   # Skip symlinks since gsutil chokes on broken ones (and just duplicates
   # symlinked files, in any case).
   def GetSymlinks(dirname, files):
     return [x for x in files if os.path.islink(os.path.join(dirname, x))]
   shutil.copytree(src, dest, ignore=GetSymlinks)


 class TastVMTestStage(generic_stages.BoardSpecificBuilderStage,
                       generic_stages.ArchivingStageMixin):
   """Runs Tast integration tests in a virtual machine."""

   category = constants.TEST_INFRA_STAGE

   # Time allotted to cros_run_vm_test to clean up (i.e. shut down the VM) after
   # receiving SIGTERM. After this, SIGKILL is sent.
   CLEANUP_TIMEOUT_SEC = 10 * 60

   # These magic attributes can be used to turn off the stage via the build
   # config. See generic_stages.BuilderStage.
   option_name = 'tests'
   config_name = 'tast_vm_tests'

   def PerformStage(self):
     """Performs the stage. Overridden from generic_stages.BuilderStage."""

     # CreateTestRoot creates a results directory and returns its path relative
     # to the chroot.
     chroot_results_dir = commands.CreateTestRoot(self._build_root)

     try:
       got_exception = False
       try:
         self._RunAllSuites(self._run.config.tast_vm_tests, chroot_results_dir)
       except Exception:
         # sys.exc_info() returns (None, None, None) in the finally block, so we
         # need to record the fact that we already have an error here.
         got_exception = True
         raise
       finally:
         self._ProcessAndArchiveResults(
             self._MakeChrootPathAbsolute(chroot_results_dir),
             [t.suite_name for t in self._run.config.tast_vm_tests],
             got_exception)
     except Exception:
       logging.exception('Tast VM tests failed')
       raise

   def _MakeChrootPathAbsolute(self, path):
     """Appends the supplied path to the chroot's path.

     Args:
       path: String containing path (either relative or absolute) to be rooted
             in chroot.

     Returns:
       String containing chroot suffixed by path.
     """
     # When os.path.join encounters an absolute path, it throws away everything
     # it's already seen.
     return os.path.join(self._build_root, constants.DEFAULT_CHROOT_DIR,
                         path.lstrip('/'))

   def _RunAllSuites(self, suites, base_chroot_results_dir):
     """Runs multiple test suites sequentially.

     Args:
       suites: List of TastVMTestConfig objects describing suites to run.
       base_chroot_results_dir: Base results directory relative to chroot.

     Raises:
       failures_lib.TestFailure if an internal error is encountered.
     """
     with cgroups.SimpleContainChildren('TastVMTest'):
       for suite in suites:
         logging.info('Running Tast VM test suite %s (%s)',
                      suite.suite_name, (' '.join(suite.test_exprs)))
         # We apparently always prefix reasons with spaces because timeout_util
         # appends them directly to error messages.
         reason = ' Reached TastVMTestStage test run timeout.'
         with timeout_util.Timeout(suite.timeout, reason_message=reason):
           self._RunSuite(suite.test_exprs,
                          os.path.join(base_chroot_results_dir,
                                       suite.suite_name))

   def _RunSuite(self, test_exprs, suite_chroot_results_dir):
     """Runs a collection of tests.

     Args:
       test_exprs: List of string expressions describing which tests to run; this
                   is passed directly to the 'tast run' command. See
                   https://goo.gl/UPNEgT for info about test expressions.
       suite_chroot_results_dir: String containing path of directory where the
                                 tast command should store test results,
                                 relative to chroot.

     Raises:
       failures_lib.TestFailure if an internal error is encountered.
     """
     vm_path = os.path.join(self.GetImageDirSymlink(), constants.VM_IMAGE_BIN)
     results_dir = self._MakeChrootPathAbsolute(suite_chroot_results_dir)
     cmd = ['./cros_run_vm_test', '--no-display', '--copy-on-write', '--debug',
            '--board=%s' % self._current_board, '--image-path=%s' % vm_path,
            '--results-dir=%s' % results_dir, '--tast'] + test_exprs

     result = cros_build_lib.RunCommand(
         cmd, error_code_ok=True, cwd=constants.CHROMITE_BIN_DIR,
         kill_timeout=TastVMTestStage.CLEANUP_TIMEOUT_SEC
     )
     if result.returncode:
       raise failures_lib.TestFailure(FAILURE_EXIT_CODE % result.returncode)

   @failures_lib.SetFailureType(failures_lib.InfrastructureFailure,
                                exclude_exceptions=[failures_lib.TestFailure])
   def _ProcessAndArchiveResults(self, abs_results_dir, suite_names,
                                 already_have_error):
     """Processes and archives test results.

     Args:
       abs_results_dir: Absolute path to directory containing test results.
       suite_names: List of string test suite names.
       already_have_error: Boolean for whether testing has already failed.

     Raises:
       failures_lib.TestFailure if one or more tests failed or results were
         unavailable. Suppressed if already_have_error is True.
     """
     if not os.path.isdir(abs_results_dir) or not os.listdir(abs_results_dir):
       raise failures_lib.TestFailure(FAILURE_NO_RESULTS % abs_results_dir)

     archive_base = constants.TAST_VM_TEST_RESULTS % {'attempt': self._attempt}
     _CopyResultsDir(abs_results_dir,
                     os.path.join(self.archive_path, archive_base))

     # TODO(crbug.com/770562): Collect stack traces once the tast executable is
     # symbolizing and collecting them (see VMTestStage._ArchiveTestResults).

     # Now archive the results to Cloud Storage.
     logging.info('Uploading artifacts to Cloud Storage...')
     self.UploadArtifact(archive_base, archive=False, strict=False)
     self.PrintDownloadLink(archive_base, RESULTS_LINK_PREFIX)

     try:
       self._ProcessResultsFile(abs_results_dir, archive_base, suite_names)
     except Exception as e:
       # Don't raise a new exception if testing already failed.
       if already_have_error:
         logging.exception('Got error while archiving or processing results')
       else:
         raise e
     finally:
       osutils.RmDir(abs_results_dir, ignore_missing=True, sudo=True)

   def _ProcessResultsFile(self, abs_results_dir, url_base, suite_names):
     """Parses the results file and prints links to failed tests.

     Args:
       abs_results_dir: Absolute path to directory containing test results.
       url_base: Relative path within the archive dir where results are stored.
       suite_names: List of string test suite names.

     Raises:
       failures_lib.TestFailure if one or more tests failed or results were
         missing or unreadable.
     """
     num_failed = 0

     for suite_name in sorted(suite_names):
       results_path = os.path.join(abs_results_dir, suite_name, RESULTS_FILENAME)

       # The results file contains an array with objects representing tests.
       # Each object should contain the test name in a 'name' attribute and a
       # list of errors in an 'error' attribute.
       try:
         with open(results_path, 'r') as f:
           for test in json.load(f):
             if test[RESULTS_ERRORS_KEY]:
               name = test[RESULTS_NAME_KEY]
               informational = (RESULTS_INFORMATIONAL_ATTR in
                                test.get(RESULTS_ATTR_KEY, []))
               test_url = os.path.join(
                   url_base, suite_name, RESULTS_TESTS_DIR, name)
               desc = INFORMATIONAL_PREFIX + name if informational else name
               self.PrintDownloadLink(test_url, text_to_display=desc)

               # Ignore the failure if the test was marked informational.
               if not informational:
                 num_failed += 1
       except Exception as e:
         raise failures_lib.TestFailure(FAILURE_BAD_RESULTS %
                                        (results_path, str(e)))

     if num_failed > 0:
       logging.error('%d test(s) failed', num_failed)
       raise failures_lib.TestFailure(FAILURE_TESTS_FAILED % num_failed)
	# -- coding: utf-8 --
	# Copyright 2017 The Chromium OS Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	"""Module containing stages for running Tast integration tests.

	This module contains cbuildbot test stages that run Tast integration tests. See
	https://chromium.googlesource.com/chromiumos/platform/tast/ for more details.
	"""

	from __future__ import print_function

	import json
	import os
	import shutil

	from chromite.cbuildbot import commands
	from chromite.cbuildbot.stages import generic_stages
	from chromite.lib import cgroups
	from chromite.lib import constants
	from chromite.lib import cros_build_lib
	from chromite.lib import cros_logging as logging
	from chromite.lib import failures_lib
	from chromite.lib import osutils
	from chromite.lib import timeout_util


	# Error format string used when tast exits with a non-zero exit code.
	FAILURE_EXIT_CODE = ' Tests failed with code %d '

	# Error format string used when one or more tests fail.
	FAILURE_TESTS_FAILED = ' %d test(s) failed '

	# Error format string used when results dir is missing or empty.
	FAILURE_NO_RESULTS = ' No results in %s '

	# Error format string used when results file is unreadable.
	FAILURE_BAD_RESULTS = ' Failed to read results from %s: %s '

	# Prefix for results download link.
	RESULTS_LINK_PREFIX = 'results: '

	# Prefix for results link for informational tests.
	INFORMATIONAL_PREFIX = 'informational: '

	# Name of JSON file containing individual tests' results written by the tast
	# command to the results dir.
	RESULTS_FILENAME = 'results.json'

	# Names of properties in test objects from results JSON files.
	RESULTS_NAME_KEY = 'name'
	RESULTS_ERRORS_KEY = 'errors'
	RESULTS_ATTR_KEY = 'attr'

	# Attribute used to label informational tests.
	RESULTS_INFORMATIONAL_ATTR = 'informational'

	# Directory written within main results dir by tast command containing per-test
	# results.
	RESULTS_TESTS_DIR = 'tests'


	def _CopyResultsDir(src, dest):
	"""Copies a results dir to a new directory for archiving.

	Args:
	src: String source path.
	dest: String destination path (presumably under
	generic_stages.ArchivingStageMixin.archive_path). Must not exist
	already.

	Raises:
	OSError if dest already exists or the copy fails.
	"""
	# Skip symlinks since gsutil chokes on broken ones (and just duplicates
	# symlinked files, in any case).
	def GetSymlinks(dirname, files):
	return [x for x in files if os.path.islink(os.path.join(dirname, x))]
	shutil.copytree(src, dest, ignore=GetSymlinks)


	class TastVMTestStage(generic_stages.BoardSpecificBuilderStage,
	generic_stages.ArchivingStageMixin):
	"""Runs Tast integration tests in a virtual machine."""

	category = constants.TEST_INFRA_STAGE

	# Time allotted to cros_run_vm_test to clean up (i.e. shut down the VM) after
	# receiving SIGTERM. After this, SIGKILL is sent.
	CLEANUP_TIMEOUT_SEC = 10 * 60

	# These magic attributes can be used to turn off the stage via the build
	# config. See generic_stages.BuilderStage.
	option_name = 'tests'
	config_name = 'tast_vm_tests'

	def PerformStage(self):
	"""Performs the stage. Overridden from generic_stages.BuilderStage."""

	# CreateTestRoot creates a results directory and returns its path relative
	# to the chroot.
	chroot_results_dir = commands.CreateTestRoot(self._build_root)

	try:
	got_exception = False
	try:
	self._RunAllSuites(self._run.config.tast_vm_tests, chroot_results_dir)
	except Exception:
	# sys.exc_info() returns (None, None, None) in the finally block, so we
	# need to record the fact that we already have an error here.
	got_exception = True
	raise
	finally:
	self._ProcessAndArchiveResults(
	self._MakeChrootPathAbsolute(chroot_results_dir),
	[t.suite_name for t in self._run.config.tast_vm_tests],
	got_exception)
	except Exception:
	logging.exception('Tast VM tests failed')
	raise

	def _MakeChrootPathAbsolute(self, path):
	"""Appends the supplied path to the chroot's path.

	Args:
	path: String containing path (either relative or absolute) to be rooted
	in chroot.

	Returns:
	String containing chroot suffixed by path.
	"""
	# When os.path.join encounters an absolute path, it throws away everything
	# it's already seen.
	return os.path.join(self._build_root, constants.DEFAULT_CHROOT_DIR,
	path.lstrip('/'))

	def _RunAllSuites(self, suites, base_chroot_results_dir):
	"""Runs multiple test suites sequentially.

	Args:
	suites: List of TastVMTestConfig objects describing suites to run.
	base_chroot_results_dir: Base results directory relative to chroot.

	Raises:
	failures_lib.TestFailure if an internal error is encountered.
	"""
	with cgroups.SimpleContainChildren('TastVMTest'):
	for suite in suites:
	logging.info('Running Tast VM test suite %s (%s)',
	suite.suite_name, (' '.join(suite.test_exprs)))
	# We apparently always prefix reasons with spaces because timeout_util
	# appends them directly to error messages.
	reason = ' Reached TastVMTestStage test run timeout.'
	with timeout_util.Timeout(suite.timeout, reason_message=reason):
	self._RunSuite(suite.test_exprs,
	os.path.join(base_chroot_results_dir,
	suite.suite_name))

	def _RunSuite(self, test_exprs, suite_chroot_results_dir):
	"""Runs a collection of tests.

	Args:
	test_exprs: List of string expressions describing which tests to run; this
	is passed directly to the 'tast run' command. See
	https://goo.gl/UPNEgT for info about test expressions.
	suite_chroot_results_dir: String containing path of directory where the
	tast command should store test results,
	relative to chroot.

	Raises:
	failures_lib.TestFailure if an internal error is encountered.
	"""
	vm_path = os.path.join(self.GetImageDirSymlink(), constants.VM_IMAGE_BIN)
	results_dir = self._MakeChrootPathAbsolute(suite_chroot_results_dir)
	cmd = ['./cros_run_vm_test', '--no-display', '--copy-on-write', '--debug',
	'--board=%s' % self._current_board, '--image-path=%s' % vm_path,
	'--results-dir=%s' % results_dir, '--tast'] + test_exprs

	result = cros_build_lib.RunCommand(
	cmd, error_code_ok=True, cwd=constants.CHROMITE_BIN_DIR,
	kill_timeout=TastVMTestStage.CLEANUP_TIMEOUT_SEC
	)
	if result.returncode:
	raise failures_lib.TestFailure(FAILURE_EXIT_CODE % result.returncode)

	@failures_lib.SetFailureType(failures_lib.InfrastructureFailure,
	exclude_exceptions=[failures_lib.TestFailure])
	def _ProcessAndArchiveResults(self, abs_results_dir, suite_names,
	already_have_error):
	"""Processes and archives test results.

	Args:
	abs_results_dir: Absolute path to directory containing test results.
	suite_names: List of string test suite names.
	already_have_error: Boolean for whether testing has already failed.

	Raises:
	failures_lib.TestFailure if one or more tests failed or results were
	unavailable. Suppressed if already_have_error is True.
	"""
	if not os.path.isdir(abs_results_dir) or not os.listdir(abs_results_dir):
	raise failures_lib.TestFailure(FAILURE_NO_RESULTS % abs_results_dir)

	archive_base = constants.TAST_VM_TEST_RESULTS % {'attempt': self._attempt}
	_CopyResultsDir(abs_results_dir,
	os.path.join(self.archive_path, archive_base))

	# TODO(crbug.com/770562): Collect stack traces once the tast executable is
	# symbolizing and collecting them (see VMTestStage._ArchiveTestResults).

	# Now archive the results to Cloud Storage.
	logging.info('Uploading artifacts to Cloud Storage...')
	self.UploadArtifact(archive_base, archive=False, strict=False)
	self.PrintDownloadLink(archive_base, RESULTS_LINK_PREFIX)

	try:
	self._ProcessResultsFile(abs_results_dir, archive_base, suite_names)
	except Exception as e:
	# Don't raise a new exception if testing already failed.
	if already_have_error:
	logging.exception('Got error while archiving or processing results')
	else:
	raise e
	finally:
	osutils.RmDir(abs_results_dir, ignore_missing=True, sudo=True)

	def _ProcessResultsFile(self, abs_results_dir, url_base, suite_names):
	"""Parses the results file and prints links to failed tests.

	Args:
	abs_results_dir: Absolute path to directory containing test results.
	url_base: Relative path within the archive dir where results are stored.
	suite_names: List of string test suite names.

	Raises:
	failures_lib.TestFailure if one or more tests failed or results were
	missing or unreadable.
	"""
	num_failed = 0

	for suite_name in sorted(suite_names):
	results_path = os.path.join(abs_results_dir, suite_name, RESULTS_FILENAME)

	# The results file contains an array with objects representing tests.
	# Each object should contain the test name in a 'name' attribute and a
	# list of errors in an 'error' attribute.
	try:
	with open(results_path, 'r') as f:
	for test in json.load(f):
	if test[RESULTS_ERRORS_KEY]:
	name = test[RESULTS_NAME_KEY]
	informational = (RESULTS_INFORMATIONAL_ATTR in
	test.get(RESULTS_ATTR_KEY, []))
	test_url = os.path.join(
	url_base, suite_name, RESULTS_TESTS_DIR, name)
	desc = INFORMATIONAL_PREFIX + name if informational else name
	self.PrintDownloadLink(test_url, text_to_display=desc)

	# Ignore the failure if the test was marked informational.
	if not informational:
	num_failed += 1
	except Exception as e:
	raise failures_lib.TestFailure(FAILURE_BAD_RESULTS %
	(results_path, str(e)))

	if num_failed > 0:
	logging.error('%d test(s) failed', num_failed)
	raise failures_lib.TestFailure(FAILURE_TESTS_FAILED % num_failed)