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