michael@0: # Copyright (c) 2012 The Chromium Authors. All rights reserved. michael@0: # Use of this source code is governed by a BSD-style license that can be michael@0: # found in the LICENSE file. michael@0: michael@0: """Collect debug info for a test.""" michael@0: michael@0: import datetime michael@0: import logging michael@0: import os michael@0: import re michael@0: import shutil michael@0: import string michael@0: import subprocess michael@0: import tempfile michael@0: michael@0: import cmd_helper michael@0: michael@0: michael@0: TOMBSTONE_DIR = '/data/tombstones/' michael@0: michael@0: michael@0: class GTestDebugInfo(object): michael@0: """A helper class to collect related debug information for a gtest. michael@0: michael@0: Debug info is collected in two steps: michael@0: - first, object(s) of this class (one per device), accumulate logs michael@0: and screenshots in tempdir. michael@0: - once the test has finished, call ZipAndCleanResults to create michael@0: a zip containing the logs from all devices, and clean them up. michael@0: michael@0: Args: michael@0: adb: ADB interface the tests are using. michael@0: device: Serial# of the Android device in which the specified gtest runs. michael@0: testsuite_name: Name of the specified gtest. michael@0: gtest_filter: Test filter used by the specified gtest. michael@0: """ michael@0: michael@0: def __init__(self, adb, device, testsuite_name, gtest_filter): michael@0: """Initializes the DebugInfo class for a specified gtest.""" michael@0: self.adb = adb michael@0: self.device = device michael@0: self.testsuite_name = testsuite_name michael@0: self.gtest_filter = gtest_filter michael@0: self.logcat_process = None michael@0: self.has_storage = False michael@0: self.log_dir = os.path.join(tempfile.gettempdir(), michael@0: 'gtest_debug_info', michael@0: self.testsuite_name, michael@0: self.device) michael@0: if not os.path.exists(self.log_dir): michael@0: os.makedirs(self.log_dir) michael@0: self.log_file_name = os.path.join(self.log_dir, michael@0: self._GeneratePrefixName() + '_log.txt') michael@0: self.old_crash_files = self._ListCrashFiles() michael@0: michael@0: def _GetSignatureFromGTestFilter(self): michael@0: """Gets a signature from gtest_filter. michael@0: michael@0: Signature is used to identify the tests from which we collect debug michael@0: information. michael@0: michael@0: Returns: michael@0: A signature string. Returns 'all' if there is no gtest filter. michael@0: """ michael@0: if not self.gtest_filter: michael@0: return 'all' michael@0: filename_chars = "-_()%s%s" % (string.ascii_letters, string.digits) michael@0: signature = ''.join(c for c in self.gtest_filter if c in filename_chars) michael@0: if len(signature) > 64: michael@0: # The signature can't be too long, as it'll be part of a file name. michael@0: signature = signature[:64] michael@0: return signature michael@0: michael@0: def _GeneratePrefixName(self): michael@0: """Generates a prefix name for debug information of the test. michael@0: michael@0: The prefix name consists of the following: michael@0: (1) root name of test_suite_base. michael@0: (2) device serial number. michael@0: (3) prefix of filter signature generate from gtest_filter. michael@0: (4) date & time when calling this method. michael@0: michael@0: Returns: michael@0: Name of the log file. michael@0: """ michael@0: return (os.path.splitext(self.testsuite_name)[0] + '_' + self.device + '_' + michael@0: self._GetSignatureFromGTestFilter() + '_' + michael@0: datetime.datetime.utcnow().strftime('%Y-%m-%d-%H-%M-%S-%f')) michael@0: michael@0: def StartRecordingLog(self, clear=True, filters=['*:v']): michael@0: """Starts recording logcat output to a file. michael@0: michael@0: This call should come before running test, with calling StopRecordingLog michael@0: following the tests. michael@0: michael@0: Args: michael@0: clear: True if existing log output should be cleared. michael@0: filters: A list of logcat filters to be used. michael@0: """ michael@0: self.StopRecordingLog() michael@0: if clear: michael@0: cmd_helper.RunCmd(['adb', '-s', self.device, 'logcat', '-c']) michael@0: logging.info('Start dumping log to %s ...', self.log_file_name) michael@0: command = 'adb -s %s logcat -v threadtime %s > %s' % (self.device, michael@0: ' '.join(filters), michael@0: self.log_file_name) michael@0: self.logcat_process = subprocess.Popen(command, shell=True) michael@0: michael@0: def StopRecordingLog(self): michael@0: """Stops an existing logcat recording subprocess.""" michael@0: if not self.logcat_process: michael@0: return michael@0: # Cannot evaluate directly as 0 is a possible value. michael@0: if self.logcat_process.poll() is None: michael@0: self.logcat_process.kill() michael@0: self.logcat_process = None michael@0: logging.info('Finish log dump.') michael@0: michael@0: def TakeScreenshot(self, identifier_mark): michael@0: """Takes a screen shot from current specified device. michael@0: michael@0: Args: michael@0: identifier_mark: A string to identify the screen shot DebugInfo will take. michael@0: It will be part of filename of the screen shot. Empty michael@0: string is acceptable. michael@0: Returns: michael@0: Returns the file name on the host of the screenshot if successful, michael@0: None otherwise. michael@0: """ michael@0: assert isinstance(identifier_mark, str) michael@0: screenshot_path = os.path.join(os.getenv('ANDROID_HOST_OUT', ''), michael@0: 'bin', michael@0: 'screenshot2') michael@0: if not os.path.exists(screenshot_path): michael@0: logging.error('Failed to take screen shot from device %s', self.device) michael@0: return None michael@0: shot_path = os.path.join(self.log_dir, ''.join([self._GeneratePrefixName(), michael@0: identifier_mark, michael@0: '_screenshot.png'])) michael@0: re_success = re.compile(re.escape('Success.'), re.MULTILINE) michael@0: if re_success.findall(cmd_helper.GetCmdOutput([screenshot_path, '-s', michael@0: self.device, shot_path])): michael@0: logging.info('Successfully took a screen shot to %s', shot_path) michael@0: return shot_path michael@0: logging.error('Failed to take screen shot from device %s', self.device) michael@0: return None michael@0: michael@0: def _ListCrashFiles(self): michael@0: """Collects crash files from current specified device. michael@0: michael@0: Returns: michael@0: A dict of crash files in format {"name": (size, lastmod), ...}. michael@0: """ michael@0: return self.adb.ListPathContents(TOMBSTONE_DIR) michael@0: michael@0: def ArchiveNewCrashFiles(self): michael@0: """Archives the crash files newly generated until calling this method.""" michael@0: current_crash_files = self._ListCrashFiles() michael@0: files = [] michael@0: for f in current_crash_files: michael@0: if f not in self.old_crash_files: michael@0: files += [f] michael@0: elif current_crash_files[f] != self.old_crash_files[f]: michael@0: # Tombstones dir can only have maximum 10 files, so we need to compare michael@0: # size and timestamp information of file if the file exists. michael@0: files += [f] michael@0: if files: michael@0: logging.info('New crash file(s):%s' % ' '.join(files)) michael@0: for f in files: michael@0: self.adb.Adb().Pull(TOMBSTONE_DIR + f, michael@0: os.path.join(self.log_dir, f)) michael@0: michael@0: @staticmethod michael@0: def ZipAndCleanResults(dest_dir, dump_file_name): michael@0: """A helper method to zip all debug information results into a dump file. michael@0: michael@0: Args: michael@0: dest_dir: Dir path in where we put the dump file. michael@0: dump_file_name: Desired name of the dump file. This method makes sure michael@0: '.zip' will be added as ext name. michael@0: """ michael@0: if not dest_dir or not dump_file_name: michael@0: return michael@0: cmd_helper.RunCmd(['mkdir', '-p', dest_dir]) michael@0: log_basename = os.path.basename(dump_file_name) michael@0: log_zip_file = os.path.join(dest_dir, michael@0: os.path.splitext(log_basename)[0] + '.zip') michael@0: logging.info('Zipping debug dumps into %s ...', log_zip_file) michael@0: # Add new dumps into the zip file. The zip may exist already if previous michael@0: # gtest also dumps the debug information. It's OK since we clean up the old michael@0: # dumps in each build step. michael@0: log_src_dir = os.path.join(tempfile.gettempdir(), 'gtest_debug_info') michael@0: cmd_helper.RunCmd(['zip', '-q', '-r', log_zip_file, log_src_dir]) michael@0: assert os.path.exists(log_zip_file) michael@0: assert os.path.exists(log_src_dir) michael@0: shutil.rmtree(log_src_dir)