1# Copyright (c) 2012 The Chromium Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5"""Collect debug info for a test.""" 6 7import datetime 8import logging 9import os 10import re 11import shutil 12import string 13import subprocess 14import tempfile 15 16import cmd_helper 17 18 19TOMBSTONE_DIR = '/data/tombstones/' 20 21 22class GTestDebugInfo(object): 23 """A helper class to collect related debug information for a gtest. 24 25 Debug info is collected in two steps: 26 - first, object(s) of this class (one per device), accumulate logs 27 and screenshots in tempdir. 28 - once the test has finished, call ZipAndCleanResults to create 29 a zip containing the logs from all devices, and clean them up. 30 31 Args: 32 adb: ADB interface the tests are using. 33 device: Serial# of the Android device in which the specified gtest runs. 34 testsuite_name: Name of the specified gtest. 35 gtest_filter: Test filter used by the specified gtest. 36 """ 37 38 def __init__(self, adb, device, testsuite_name, gtest_filter): 39 """Initializes the DebugInfo class for a specified gtest.""" 40 self.adb = adb 41 self.device = device 42 self.testsuite_name = testsuite_name 43 self.gtest_filter = gtest_filter 44 self.logcat_process = None 45 self.has_storage = False 46 self.log_dir = os.path.join(tempfile.gettempdir(), 47 'gtest_debug_info', 48 self.testsuite_name, 49 self.device) 50 if not os.path.exists(self.log_dir): 51 os.makedirs(self.log_dir) 52 self.log_file_name = os.path.join(self.log_dir, 53 self._GeneratePrefixName() + '_log.txt') 54 self.old_crash_files = self._ListCrashFiles() 55 56 def _GetSignatureFromGTestFilter(self): 57 """Gets a signature from gtest_filter. 58 59 Signature is used to identify the tests from which we collect debug 60 information. 61 62 Returns: 63 A signature string. Returns 'all' if there is no gtest filter. 64 """ 65 if not self.gtest_filter: 66 return 'all' 67 filename_chars = "-_()%s%s" % (string.ascii_letters, string.digits) 68 signature = ''.join(c for c in self.gtest_filter if c in filename_chars) 69 if len(signature) > 64: 70 # The signature can't be too long, as it'll be part of a file name. 71 signature = signature[:64] 72 return signature 73 74 def _GeneratePrefixName(self): 75 """Generates a prefix name for debug information of the test. 76 77 The prefix name consists of the following: 78 (1) root name of test_suite_base. 79 (2) device serial number. 80 (3) prefix of filter signature generate from gtest_filter. 81 (4) date & time when calling this method. 82 83 Returns: 84 Name of the log file. 85 """ 86 return (os.path.splitext(self.testsuite_name)[0] + '_' + self.device + '_' + 87 self._GetSignatureFromGTestFilter() + '_' + 88 datetime.datetime.utcnow().strftime('%Y-%m-%d-%H-%M-%S-%f')) 89 90 def StartRecordingLog(self, clear=True, filters=['*:v']): 91 """Starts recording logcat output to a file. 92 93 This call should come before running test, with calling StopRecordingLog 94 following the tests. 95 96 Args: 97 clear: True if existing log output should be cleared. 98 filters: A list of logcat filters to be used. 99 """ 100 self.StopRecordingLog() 101 if clear: 102 cmd_helper.RunCmd(['adb', '-s', self.device, 'logcat', '-c']) 103 logging.info('Start dumping log to %s ...', self.log_file_name) 104 command = 'adb -s %s logcat -v threadtime %s > %s' % (self.device, 105 ' '.join(filters), 106 self.log_file_name) 107 self.logcat_process = subprocess.Popen(command, shell=True) 108 109 def StopRecordingLog(self): 110 """Stops an existing logcat recording subprocess.""" 111 if not self.logcat_process: 112 return 113 # Cannot evaluate directly as 0 is a possible value. 114 if self.logcat_process.poll() is None: 115 self.logcat_process.kill() 116 self.logcat_process = None 117 logging.info('Finish log dump.') 118 119 def TakeScreenshot(self, identifier_mark): 120 """Takes a screen shot from current specified device. 121 122 Args: 123 identifier_mark: A string to identify the screen shot DebugInfo will take. 124 It will be part of filename of the screen shot. Empty 125 string is acceptable. 126 Returns: 127 Returns the file name on the host of the screenshot if successful, 128 None otherwise. 129 """ 130 assert isinstance(identifier_mark, str) 131 screenshot_path = os.path.join(os.getenv('ANDROID_HOST_OUT', ''), 132 'bin', 133 'screenshot2') 134 if not os.path.exists(screenshot_path): 135 logging.error('Failed to take screen shot from device %s', self.device) 136 return None 137 shot_path = os.path.join(self.log_dir, ''.join([self._GeneratePrefixName(), 138 identifier_mark, 139 '_screenshot.png'])) 140 re_success = re.compile(re.escape('Success.'), re.MULTILINE) 141 if re_success.findall(cmd_helper.GetCmdOutput([screenshot_path, '-s', 142 self.device, shot_path])): 143 logging.info('Successfully took a screen shot to %s', shot_path) 144 return shot_path 145 logging.error('Failed to take screen shot from device %s', self.device) 146 return None 147 148 def _ListCrashFiles(self): 149 """Collects crash files from current specified device. 150 151 Returns: 152 A dict of crash files in format {"name": (size, lastmod), ...}. 153 """ 154 return self.adb.ListPathContents(TOMBSTONE_DIR) 155 156 def ArchiveNewCrashFiles(self): 157 """Archives the crash files newly generated until calling this method.""" 158 current_crash_files = self._ListCrashFiles() 159 files = [] 160 for f in current_crash_files: 161 if f not in self.old_crash_files: 162 files += [f] 163 elif current_crash_files[f] != self.old_crash_files[f]: 164 # Tombstones dir can only have maximum 10 files, so we need to compare 165 # size and timestamp information of file if the file exists. 166 files += [f] 167 if files: 168 logging.info('New crash file(s):%s' % ' '.join(files)) 169 for f in files: 170 self.adb.Adb().Pull(TOMBSTONE_DIR + f, 171 os.path.join(self.log_dir, f)) 172 173 @staticmethod 174 def ZipAndCleanResults(dest_dir, dump_file_name): 175 """A helper method to zip all debug information results into a dump file. 176 177 Args: 178 dest_dir: Dir path in where we put the dump file. 179 dump_file_name: Desired name of the dump file. This method makes sure 180 '.zip' will be added as ext name. 181 """ 182 if not dest_dir or not dump_file_name: 183 return 184 cmd_helper.RunCmd(['mkdir', '-p', dest_dir]) 185 log_basename = os.path.basename(dump_file_name) 186 log_zip_file = os.path.join(dest_dir, 187 os.path.splitext(log_basename)[0] + '.zip') 188 logging.info('Zipping debug dumps into %s ...', log_zip_file) 189 # Add new dumps into the zip file. The zip may exist already if previous 190 # gtest also dumps the debug information. It's OK since we clean up the old 191 # dumps in each build step. 192 log_src_dir = os.path.join(tempfile.gettempdir(), 'gtest_debug_info') 193 cmd_helper.RunCmd(['zip', '-q', '-r', log_zip_file, log_src_dir]) 194 assert os.path.exists(log_zip_file) 195 assert os.path.exists(log_src_dir) 196 shutil.rmtree(log_src_dir) 197