src/tools/telemetry/telemetry/unittest/run_chromeos_tests.py

   1 # Copyright 2014 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 import logging
   5 import os
   6 import sys
   7
   8 from telemetry.unittest import gtest_progress_reporter
   9 from telemetry.unittest import run_tests
  10 from telemetry.core import util
  11
  12
  13 def RunTestsForChromeOS(browser_type, unit_tests, perf_tests):
  14   stream = _LoggingOutputStream()
  15   error_string = ''
  16
  17   logging.info('Running telemetry unit tests with browser_type "%s".' %
  18                browser_type)
  19   ret = _RunOneSetOfTests(browser_type, 'telemetry',
  20                           os.path.join('telemetry', 'telemetry'),
  21                           unit_tests, stream)
  22   if ret:
  23     error_string += 'The unit tests failed.\n'
  24
  25   logging.info('Running telemetry perf tests with browser_type "%s".' %
  26                browser_type)
  27   ret = _RunOneSetOfTests(browser_type, 'perf', 'perf', perf_tests, stream)
  28   if ret:
  29     error_string = 'The perf tests failed.\n'
  30
  31   return error_string
  32
  33
  34 def _RunOneSetOfTests(browser_type, root_dir, sub_dir, tests, stream):
  35   if not tests:
  36     return
  37   top_level_dir = os.path.join(util.GetChromiumSrcDir(), 'tools', root_dir)
  38   sub_dir = os.path.join(util.GetChromiumSrcDir(), 'tools', sub_dir)
  39
  40   sys.path.append(top_level_dir)
  41
  42   output_formatters = [gtest_progress_reporter.GTestProgressReporter(stream)]
  43   run_tests.config = run_tests.Config(top_level_dir, [sub_dir],
  44                                       output_formatters)
  45   return run_tests.RunTestsCommand.main(['--browser', browser_type] + tests)
  46
  47
  48 class _LoggingOutputStream(object):
  49
  50   def __init__(self):
  51     self._buffer = []
  52
  53   def write(self, s):
  54     """Buffer a string write. Log it when we encounter a newline."""
  55     if '\n' in s:
  56       segments = s.split('\n')
  57       segments[0] = ''.join(self._buffer + [segments[0]])
  58       log_level = logging.getLogger().getEffectiveLevel()
  59       try:  # TODO(dtu): We need this because of crbug.com/394571
  60         logging.getLogger().setLevel(logging.INFO)
  61         for line in segments[:-1]:
  62           logging.info(line)
  63       finally:
  64         logging.getLogger().setLevel(log_level)
  65       self._buffer = [segments[-1]]
  66     else:
  67       self._buffer.append(s)
  68
  69   def flush(self):  # pylint: disable=W0612
  70     pass