This change squashes output while pregenerationg updates.

bin/cros_au_test_harness.py

 #!/usr/bin/python
 
 # Copyright (c) 2011 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.
 
 """This module runs a suite of Auto Update tests.
 
   The tests can be run on either a virtual machine or actual device depending
   on parameters given.  Specific tests can be run by invoking --test_prefix.
@@ skipping 683 matching lines @@
     self._starting_semaphore = starting_semaphore
     self._ending_semaphore = ending_semaphore
     self._output = None
     self._completed = False
 
   def run(self):
     """Thread override.  Runs the method specified and sets output."""
     try:
       self._output = self._target(*self._args)
     finally:
-      # From threading.py to avoid a refcycle.
-      del self._target, self._args
       # Our own clean up.
       self._Cleanup()
       self._completed = True
+      # From threading.py to avoid a refcycle.
+      del self._target, self._args
 
   def GetOutput(self):
     """Returns the output of the method run."""
     assert self._completed, 'GetOutput called before thread was run.'
     return self._output
 
   def _Cleanup(self):
     """Releases semaphores for a waiting caller."""
+    Info('Completed job %s' % self)
     self._starting_semaphore.release()
     self._ending_semaphore.release()
 
   def __str__(self):
     return '%s(%s)' % (self._target, self._args)
 
 
 class DevServerWrapper(threading.Thread):
   """A Simple wrapper around a devserver instance."""
 
@@ skipping 29 matching lines @@
 
 
 def _GenerateUpdateId(target, src):
   """Returns a simple representation id of target and src paths."""
   if src:
     return '%s->%s' % (target, src)
   else:
     return target
 
 
-def _RunParallelJobs(number_of_sumultaneous_jobs, jobs, jobs_args):
+def _RunParallelJobs(number_of_sumultaneous_jobs, jobs, jobs_args, print_status):
+
   """Runs set number of specified jobs in parallel.
 
   Args:
     number_of_simultaneous_jobs:  Max number of threads to be run in parallel.
     jobs:  Array of methods to run.
     jobs_args:  Array of args associated with method calls.
+    print_status: True if you'd like this to print out .'s as it runs jobs.
   Returns:
     Returns an array of results corresponding to each thread.
   """
   def _TwoTupleize(x, y):
     return (x, y)
 
   threads = []
   job_start_semaphore = threading.Semaphore(number_of_sumultaneous_jobs)
   join_semaphore = threading.Semaphore(0)
   assert len(jobs) == len(jobs_args), 'Length of args array is wrong.'
 
   # Create the parallel jobs.
   for job, args in map(_TwoTupleize, jobs, jobs_args):
     thread = ParallelJob(job_start_semaphore, join_semaphore, target=job,
                          args=args)
     threads.append(thread)
 
+  # Cache sudo access.
+  RunCommand(['sudo', 'echo', 'Starting test harness'],
+             print_cmd=False, redirect_stdout=True, redirect_stderr=True)
+
   # We use a semaphore to ensure we don't run more jobs that required.
   # After each thread finishes, it releases (increments semaphore).
   # Acquire blocks of num jobs reached and continues when a thread finishes.
   for next_thread in threads:
     job_start_semaphore.acquire(blocking=True)
     Info('Starting job %s' % next_thread)
     next_thread.start()
 
   # Wait on the rest of the threads to finish.
+  Info('Waiting for threads to complete.')
   for thread in threads:
-    join_semaphore.acquire(blocking=True)
+    while not join_semaphore.acquire(blocking=False):
+      time.sleep(5)
+      if print_status:
+        print >> sys.stderr, '.',
 
   return [thread.GetOutput() for thread in threads]
 
 
 def _PrepareTestSuite(parser, options, test_class):
   """Returns a prepared test suite given by the options and test class."""
   test_class.ProcessOptions(parser, options)
   test_loader = unittest.TestLoader()
   test_loader.testMethodPrefix = options.test_prefix
   return test_loader.loadTestsFromTestCase(test_class)
 
 
 def _PregenerateUpdates(parser, options):
   """Determines all deltas that will be generated and generates them.
 
   This method effectively pre-generates the dev server cache for all tests.
 
   Args:
     parser: parser from main.
     options: options from parsed parser.
   Returns:
     Dictionary of Update Identifiers->Relative cache locations.
+  Raises:
+    UpdateException if we fail to generate an update.
   """
   def _GenerateVMUpdate(target, src):
     """Generates an update using the devserver."""
     target = ReinterpretPathForChroot(target)
     if src:
       src = ReinterpretPathForChroot(src)
 
-    return RunCommand(['sudo',
-                       './start_devserver',
-                       '--pregenerate_update',
-                       '--exit',
-                       '--image=%s' % target,
-                       '--src_image=%s' % src,
-                       '--for_vm',
-                       ], redirect_stdout=True, enter_chroot=True,
-                       print_cmd=False)
+    return RunCommandCaptureOutput(['sudo',
+                                    './start_devserver',
+                                    '--pregenerate_update',
+                                    '--exit',
+                                    '--image=%s' % target,
+                                    '--src_image=%s' % src,
+                                    '--for_vm',
+                                   ], combine_stdout_stderr=True,
+                                   enter_chroot=True,
+                                   print_cmd=False)
 
   # Get the list of deltas by mocking out update method in test class.
   test_suite = _PrepareTestSuite(parser, options, GenerateVirtualAUDeltasTest)
   test_result = unittest.TextTestRunner(verbosity=0).run(test_suite)
 
   Info('The following delta updates are required.')
   update_ids = []
   jobs = []
   args = []
   for target, srcs in GenerateVirtualAUDeltasTest.delta_list.items():
     for src in srcs:
       update_id = _GenerateUpdateId(target=target, src=src)
       print >> sys.stderr, 'AU: %s' % update_id
       update_ids.append(update_id)
       jobs.append(_GenerateVMUpdate)
       args.append((target, src))
 
-  raw_results = _RunParallelJobs(options.jobs, jobs, args)
+  raw_results = _RunParallelJobs(options.jobs, jobs, args, print_status=True)
   results = []
 
-  # Parse the output.
+  # Looking for this line in the output.
   key_line_re = re.compile('^PREGENERATED_UPDATE=([\w/.]+)')
   for result in raw_results:
-    for line in result.splitlines():
-      match = key_line_re.search(line)
-      if match:
-        # Convert blah/blah/update.gz -> update/blah/blah.
-        path_to_update_gz = match.group(1).rstrip()
-        (path_to_update_dir, _, _) = path_to_update_gz.rpartition('/update.gz')
-        results.append('/'.join(['update', path_to_update_dir]))
-        break
+    (return_code, output, _) = result
+    if return_code != 0:
+      Warning(output)
+      raise UpdateException(return_code, 'Failed to generate all updates.')
+    else:
+      for line in output.splitlines():
+        match = key_line_re.search(line)
+        if match:
+          # Convert blah/blah/update.gz -> update/blah/blah.
+          path_to_update_gz = match.group(1).rstrip()
+          (path_to_update_dir, _, _) = path_to_update_gz.rpartition(
+              '/update.gz')
+          results.append('/'.join(['update', path_to_update_dir]))
+          break
 
-  assert len(raw_results) == len(results), \
-         'Insufficient number cache directories returned.'
+  # Make sure all generation of updates returned cached locations.
+  if len(raw_results) != len(results):
+    raise UpdateException(1, 'Insufficient number cache directories returned.')
 
   # Build the dictionary from our id's and returned cache paths.
   cache_dictionary = {}
   for index, id in enumerate(update_ids):
     cache_dictionary[id] = results[index]
 
   return cache_dictionary
 
 
 def _RunTestsInParallel(parser, options, test_class):
   """Runs the tests given by the options and test_class in parallel."""
   threads = []
   args = []
   test_suite = _PrepareTestSuite(parser, options, test_class)
   for test in test_suite:
     test_name = test.id()
     test_case = unittest.TestLoader().loadTestsFromName(test_name)
     threads.append(unittest.TextTestRunner().run)
     args.append(test_case)
 
-  results = _RunParallelJobs(options.jobs, threads, args)
+  results = _RunParallelJobs(options.jobs, threads, args, print_status=False)
   if not (test_result.wasSuccessful() for test_result in results):
     Die('Test harness was not successful')
 
 
 def main():
   parser = optparse.OptionParser()
   parser.add_option('-b', '--base_image',
                     help='path to the base image.')
   parser.add_option('-r', '--board',
                     help='board for the images.')
@@ skipping 42 matching lines @@
   else:
     dev_server_cache = None
     test_suite = _PrepareTestSuite(parser, options, test_class)
     test_result = unittest.TextTestRunner(verbosity=2).run(test_suite)
     if not test_result.wasSuccessful():
       Die('Test harness was not successful.')
 
 
 if __name__ == '__main__':
   main()