| """Build step class for stripping OTA test binaries.""" |
| |
| # pylint triggers false positives in _setup/_teardown_mocks |
| # pylint: disable=method-hidden |
| |
| from __future__ import absolute_import |
| import logging |
| import os |
| import sys |
| |
| from slave import base_step |
| |
| _UNIT_TEST_ARCHIVE_MODULE = 'unit_test.unit_test_archive_lib' |
| _UNIT_TEST_HELPER_MODULE = 'unit_test.unit_test_helper' |
| |
| # If a file is larger than this, the strip tool will fail to recognize the file. |
| _MAX_32BIT_FILESIZE = 2**32 |
| _MAX_32BIT_FILESIZE_STR = '4GB' |
| _STRIPPED_EXT = '.stripped' |
| |
| |
| def _InjectPythonPath(path): |
| if path not in sys.path: |
| sys.path.append(path) |
| |
| |
| class OtaStripBinariesStep(base_step.BaseStep): |
| """Build step class for stripping OTA test binaries.""" |
| |
| def __init__(self, **kwargs): |
| """Creates a OtaStripBinariesStep instance. |
| |
| Args: |
| **kwargs: Any additional args to pass to BaseStep. |
| """ |
| base_step.BaseStep.__init__(self, name='strip binaries', **kwargs) |
| # Use the unit_test_helper modules to be consistent with Catabuilder. |
| self._unit_test_helper = None |
| self._unit_test_archive_lib = None |
| self._const = None |
| # Save list of files that need to be cleaned up at the end. |
| # Can't cleanup stripped binaries until after done archiving them. |
| self._stripped_binaries = [] |
| |
| @property |
| def const(self): |
| if not self._const: |
| self._const = self.unit_test_helper.GetUnitTestConst(None) |
| return self._const |
| |
| @property |
| def gcs_archive_dir(self): |
| output_dir = os.path.join(self.get_gcs_dir(), 'artifacts', 'tests') |
| if not os.path.exists(output_dir): |
| os.makedirs(output_dir) |
| return output_dir |
| |
| @property |
| def unit_test_helper(self): |
| """Use the unit_test_helper module to be consistent with Catabuilder.""" |
| if not self._unit_test_helper: |
| _InjectPythonPath(self.get_project_path('test')) |
| self._unit_test_helper = __import__( |
| _UNIT_TEST_HELPER_MODULE, fromlist=['']) |
| return self._unit_test_helper |
| |
| @property |
| def unit_test_archive_lib(self): |
| """Use unit_test_archive_lib module to be consistent with Catabuilder.""" |
| if not self._unit_test_archive_lib: |
| _InjectPythonPath(self.get_project_path('test')) |
| self._unit_test_archive_lib = __import__( |
| _UNIT_TEST_ARCHIVE_MODULE, fromlist=['']) |
| return self._unit_test_archive_lib |
| |
| def _get_out_dir(self): |
| chromium_src_dir = self.get_project_path('chromium/src') |
| return self.unit_test_helper.GetOutDir(chromium_src_dir) |
| |
| def _get_binaries_to_strip(self, out_dir): |
| return [ |
| binary for binary in |
| self.unit_test_helper.GetListOfBinariesToBeStripped(out_dir) |
| if binary.strip() |
| ] |
| |
| def _clear_test_binaries(self, test_binary_paths): |
| for binary in test_binary_paths: |
| stripped_binary = binary + _STRIPPED_EXT |
| if os.path.exists(stripped_binary): |
| try: |
| os.remove(stripped_binary) |
| except OSError: |
| continue |
| |
| def _get_test_files_to_archive(self, out_dir, stripped_binary_paths): |
| """Gets the set of test binaries & dependency files to archive.""" |
| stripped_binary_names = set( |
| [os.path.basename(path) for path in stripped_binary_paths]) |
| # Get a flattened deduped list of all binaries & dependencies to tar. |
| tar_files = set() |
| deps_dict = self.unit_test_archive_lib.GetAllTestDependenciesDict( |
| self.const, out_dir) |
| for test_deps in deps_dict.values(): |
| for dependency in test_deps: |
| # Use the stripped version of test binaries where available. |
| if os.path.basename( |
| dependency) in stripped_binary_names and os.path.exists( |
| os.path.join(self.const.chrome_srcroot, |
| dependency + _STRIPPED_EXT)): |
| tar_files.add(dependency + _STRIPPED_EXT) |
| else: |
| tar_files.add(dependency) |
| return tar_files |
| |
| def _strip_binary(self, file_path, output_binary): |
| return self.unit_test_helper.StripBinary( |
| file_path, output_file=output_binary) |
| |
| def _strip_error_to_comment(self, binary, stderr): |
| """Convert errors from the strip tool into comments for gerrit.""" |
| if not stderr: |
| return '' |
| |
| comment = stderr.replace(os.path.abspath(os.getcwd()), '') |
| binary_size = os.path.getsize(binary) |
| if binary_size > _MAX_32BIT_FILESIZE: |
| logging.warning('%s=%s bytes > %s', binary, binary_size, |
| _MAX_32BIT_FILESIZE_STR) |
| comment += ('{binary} ({size:,}bytes) is larger than {max_size}.' |
| ' Too large to strip. (b/33284348)'.format( |
| binary=os.path.basename(binary), |
| size=binary_size, |
| max_size=_MAX_32BIT_FILESIZE_STR)) |
| return comment |
| |
| def _strip_binaries(self, test_binaries): |
| """Strip the test binaries built for the most recent ota.""" |
| failure_messages = [] |
| for binary in test_binaries: |
| # Try to strip the binary. |
| output_binary = binary + _STRIPPED_EXT |
| try: |
| self._strip_binary(binary, output_binary) |
| except IOError as e: |
| # If stripping failed, report the error. |
| failure_messages.append(self._strip_error_to_comment(binary, str(e))) |
| return failure_messages |
| |
| def _archive_binaries(self, files_to_archive): |
| self.unit_test_archive_lib.ZipWithRelativeFileList( |
| self.const.chrome_srcroot, |
| os.path.join(self.gcs_archive_dir, 'test_deps.tar.gz'), |
| files_to_archive, |
| # Remove .stripped from all filenames within the tar. |
| transform=r's|{}$||'.format(_STRIPPED_EXT)) |
| |
| def run(self): |
| """Run the strip tool over the unittest binaries and archive them to GCS.""" |
| try: |
| out_dir = self._get_out_dir() |
| except ValueError as error: |
| # |out_dir| may not exist if a previous build step failed. However, we |
| # still want this step to be marked as failing in case the out_dir does |
| # actually exist somewhere and this step isn't finding it correctly. |
| self.add_review({'message': error.message}) |
| return False |
| |
| test_binary_paths = self._get_binaries_to_strip(out_dir) |
| # Clear in case recipe aborted in a previous run before able to cleanup |
| self._clear_test_binaries(test_binary_paths) |
| |
| failure_messages = self._strip_binaries(test_binary_paths) |
| if failure_messages: |
| self.add_review({'message': '\n'.join(failure_messages)}) |
| return False |
| |
| # Archive stripped test binaries into a tar file in gcs_archive_dir |
| # so that they'll be uploaded to GCS. Catatester will use the tar file |
| # for running these tests on-device in the lab. |
| files_to_archive = self._get_test_files_to_archive(out_dir, |
| test_binary_paths) |
| self._archive_binaries(files_to_archive) |
| |
| self._clear_test_binaries(test_binary_paths) |
| return True |
