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