Refactor generate_test_report for better library usage.

utils_py/generate_test_report.py

 #!/usr/bin/python
 # Copyright (c) 2010 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 
 """Parses and displays the contents of one or more autoserv result directories.
 
 This script parses the contents of one or more autoserv results folders and
 generates test reports.
@@ skipping 10 matching lines @@
 sys.path.append(constants.CROSUTILS_LIB_DIR)
 from cros_build_lib import Color, Die
 
 _STDOUT_IS_TTY = hasattr(sys.stdout, 'isatty') and sys.stdout.isatty()
 
 # List of crashes which are okay to ignore. This list should almost always be
 # empty. If you add an entry, mark it with a TODO(<your name>) and the issue
 # filed for the crash.
 _CRASH_WHITELIST = {}
 
-class ReportGenerator(object):
-  """Collects and displays data from autoserv results directories.
 
-  This class collects status and performance data from one or more autoserv
-  result directories and generates test reports.
-  """
+class ResultCollector(object):
+  """Collects status and performance data from an autoserv results directory."""
 
-  _KEYVAL_INDENT = 2
+  def __init__(self, collect_perf=True, strip_text=''):
+    """Initialize ResultsCollector class.
 
-  def __init__(self, options, args):
-    self._options = options
-    self._args = args
-    self._color = Color(options.color)
+    Args:
+      collect_perf: Should perf keyvals be collected?
+      strip_text: Prefix to strip from test directory names.
+    """
+    self._collect_perf = collect_perf
+    self._strip_text = strip_text
 
   def _CollectPerf(self, testdir):
     """Parses keyval file under testdir.
 
     If testdir contains a result folder, process the keyval file and return
     a dictionary of perf keyval pairs.
 
     Args:
       testdir: The autoserv test result directory.
 
     Returns:
       If the perf option is disabled or the there's no keyval file under
       testdir, returns an empty dictionary. Otherwise, returns a dictionary of
       parsed keyvals. Duplicate keys are uniquified by their instance number.
     """
 
     perf = {}
-    if not self._options.perf:
+    if not self._collect_perf:
       return perf
 
     keyval_file = os.path.join(testdir, 'results', 'keyval')
     if not os.path.isfile(keyval_file):
       return perf
 
     instances = {}
 
     for line in open(keyval_file):
       match = re.search(r'^(.+){perf}=(.+)$', line)
       if match:
         key = match.group(1)
         val = match.group(2)
 
         # If the same key name was generated multiple times, uniquify all
         # instances other than the first one by adding the instance count
         # to the key name.
         key_inst = key
         instance = instances.get(key, 0)
         if instance:
           key_inst = '%s{%d}' % (key, instance)
         instances[key] = instance + 1
 
         perf[key_inst] = val
 
     return perf
 
-  def _CollectResult(self, testdir):
+  def _CollectResult(self, testdir, results):
     """Adds results stored under testdir to the self._results dictionary.
 
     If testdir contains 'status.log' or 'status' files, assume it's a test
     result directory and add the results data to the self._results dictionary.
     The test directory name is used as a key into the results dictionary.
 
     Args:
       testdir: The autoserv test result directory.
+      results: Results dictionary to store results in.
     """
 
     status_file = os.path.join(testdir, 'status.log')
     if not os.path.isfile(status_file):
       status_file = os.path.join(testdir, 'status')
       if not os.path.isfile(status_file):
         return
 
-    # Remove false positives that are missing a debug dir.
-    if not os.path.exists(os.path.join(testdir, 'debug')):
-      return

[sosa, 2011/03/31 20:03:34]

Why move this into the caller but not the above?  Maybe move both into a
_ShouldFilterResult method that is called from CollectResults

[DaleCurtis, 2011/03/31 20:17:43]

Just checking for the debug directory prunes everything we need to and the above
code introduces variables depended on by the code below.

I could do it by creating two new functions, GetStatusFile and
IsValidResultsDirectory, but there'd still be two calls to GetStatusFile unless
I pass in the status file which seems confusing.

On 2011/03/31 20:03:34, sosa wrote:

-
     status_raw = open(status_file, 'r').read()
     status = 'FAIL'
     if (re.search(r'GOOD.+completed successfully', status_raw) and
         not re.search(r'ABORT|ERROR|FAIL|TEST_NA', status_raw)):
       status = 'PASS'
 
     perf = self._CollectPerf(testdir)
 
-    if testdir.startswith(self._options.strip):
-      testdir = testdir.replace(self._options.strip, '', 1)
+    if testdir.startswith(self._strip_text):
+      testdir = testdir.replace(self._strip_text, '', 1)
 
     crashes = []
     regex = re.compile('Received crash notification for ([-\w]+).+ (sig \d+)')
     for match in regex.finditer(status_raw):
       if (match.group(1) in _CRASH_WHITELIST and
           match.group(2) in _CRASH_WHITELIST[match.group(1)]):
         continue
       crashes.append('%s %s' % match.groups())
 
-    self._results[testdir] = {'crashes': crashes,
-                              'status': status,
-                              'perf': perf}
+    results[testdir] = {'crashes': crashes, 'status': status, 'perf': perf}
 
-  def _CollectResultsRec(self, resdir):
-    """Recursively collect results into the self._results dictionary.
+  def CollectResults(self, resdir):
+    """Recursively collect results into a dictionary.
 
     Args:
       resdir: results/test directory to parse results from and recurse into.
+
+    Returns:
+      Dictionary of results.
     """
+    results = {}
+    self._CollectResult(resdir, results)
+    for testdir in glob.glob(os.path.join(resdir, '*')):
+      # Remove false positives that are missing a debug dir.
+      if not os.path.exists(os.path.join(testdir, 'debug')):
+        continue
 
-    self._CollectResult(resdir)
-    for testdir in glob.glob(os.path.join(resdir, '*')):
-      self._CollectResultsRec(testdir)
+      results.update(self.CollectResults(testdir))
+    return results
+
+
+class ReportGenerator(object):
+  """Collects and displays data from autoserv results directories.
+
+  This class collects status and performance data from one or more autoserv
+  result directories and generates test reports.
+  """
+
+  _KEYVAL_INDENT = 2
+
+  def __init__(self, options, args):
+    self._options = options
+    self._args = args
+    self._color = Color(options.color)
 
   def _CollectResults(self):
     """Parses results into the self._results dictionary.
 
     Initializes a dictionary (self._results) with test folders as keys and
     result data (status, perf keyvals) as values.
     """
     self._results = {}
+    collector = ResultCollector(self._options.perf, self._options.strip)
     for resdir in self._args:
       if not os.path.isdir(resdir):
         Die('\'%s\' does not exist' % resdir)
-      self._CollectResultsRec(resdir)
+      self._results.update(collector.CollectResults(resdir))
 
     if not self._results:
       Die('no test directories found')
 
-  def GetTestColumnWidth(self):
+  def _GetTestColumnWidth(self):
     """Returns the test column width based on the test data.
 
     Aligns the test results by formatting the test directory entry based on
     the longest test directory or perf key string stored in the self._results
     dictionary.
 
     Returns:
       The width for the test columnt.
     """
     width = len(max(self._results, key=len))
@@ skipping 10 matching lines @@
     Prints a result table to stdout. Each row of the table contains the test
     result directory and the test result (PASS, FAIL). If the perf option is
     enabled, each test entry is followed by perf keyval entries from the test
     results.
     """
     tests = self._results.keys()
     tests.sort()
 
     tests_with_errors = []
 
-    width = self.GetTestColumnWidth()
+    width = self._GetTestColumnWidth()
     line = ''.ljust(width + 5, '-')
 
     crashes = {}
     tests_pass = 0
     print line
     for test in tests:
       # Emit the test/status entry first
       test_entry = test.ljust(width)
       result = self._results[test]
       status_entry = result['status']
@@ skipping 49 matching lines @@
         print 'Total unique crashes: ' + self._color.Color(Color.BOLD,
                                                            str(len(crashes)))
       else:
         print self._color.Color(Color.GREEN,
                                 'No crashes detected during testing.')
 
     # Print out error log for failed tests.
     if self._options.print_debug:
       for test in tests_with_errors:
         debug_file_regex = os.path.join(self._options.strip, test, 'debug',
-                                       '%s*.ERROR' % os.path.basename(test))
+                                        '%s*.ERROR' % os.path.basename(test))
         for path in glob.glob(debug_file_regex):
           try:
             fh = open(path)
             print >> sys.stderr, (
                 '\n========== ERROR FILE %s FOR TEST %s ==============\n' % (
                 path, test))
             out = fh.read()
             while out:
               print >> sys.stderr, out
               out = fh.read()
@@ skipping 48 matching lines @@
   if not args:
     parser.print_help()
     Die('no result directories provided')
 
   generator = ReportGenerator(options, args)
   generator.Run()
 
 
 if __name__ == '__main__':
   main()