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
5import logging
6import time
7import unittest
8
9from telemetry.internal.util import exception_formatter
10from telemetry.testing import progress_reporter
11
12
13def _FormatTestName(test):
14  chunks = test.id().split('.')[2:]
15  return '.'.join(chunks)
16
17
18class GTestProgressReporter(progress_reporter.ProgressReporter):
19  def __init__(self, output_stream):
20    super(GTestProgressReporter, self).__init__(output_stream)
21    self._suite_start_time = None
22    self._test_start_time = None
23
24  def _Print(self, *args):
25    print >> self._output_stream, ' '.join(map(str, args))
26    self._output_stream.flush()
27
28  def _TestTimeMs(self):
29    return (time.time() - self._test_start_time) * 1000
30
31  def StartTest(self, test):
32    self._Print('[ RUN      ]', _FormatTestName(test))
33    self._test_start_time = time.time()
34
35  def StartTestSuite(self, suite):
36    contains_test_suites = any(isinstance(test, unittest.TestSuite)
37                               for test in suite)
38    if not contains_test_suites:
39      test_count = len([test for test in suite])
40      unit = 'test' if test_count == 1 else 'tests'
41      self._Print('[----------]', test_count, unit)
42      self._suite_start_time = time.time()
43
44  def StopTestSuite(self, suite):
45    contains_test_suites = any(isinstance(test, unittest.TestSuite)
46                               for test in suite)
47    if not contains_test_suites:
48      test_count = len([test for test in suite])
49      unit = 'test' if test_count == 1 else 'tests'
50      elapsed_ms = (time.time() - self._suite_start_time) * 1000
51      self._Print('[----------]', test_count, unit,
52                  '(%d ms total)' % elapsed_ms)
53      self._Print()
54
55  def StopTestRun(self, result):
56    unit = 'test' if len(result.successes) == 1 else 'tests'
57    self._Print('[  PASSED  ]', len(result.successes), '%s.' % unit)
58    if result.errors or result.failures:
59      all_errors = result.errors[:]
60      all_errors.extend(result.failures)
61      unit = 'test' if len(all_errors) == 1 else 'tests'
62      self._Print('[  FAILED  ]', len(all_errors), '%s, listed below:' % unit)
63      for test, _ in all_errors:
64        self._Print('[  FAILED  ] ', _FormatTestName(test))
65    if not result.wasSuccessful():
66      self._Print()
67      count = len(result.errors) + len(result.failures)
68      unit = 'TEST' if count == 1 else 'TESTS'
69      self._Print(count, 'FAILED', unit)
70    self._Print()
71
72  def Error(self, test, err):
73    self.Failure(test, err)
74
75  def Failure(self, test, err):
76    exception_formatter.PrintFormattedException(*err)
77    test_name = _FormatTestName(test)
78    self._Print('[  FAILED  ]', test_name, '(%0.f ms)' % self._TestTimeMs())
79
80  def Success(self, test):
81    test_name = _FormatTestName(test)
82    self._Print('[       OK ]', test_name, '(%0.f ms)' % self._TestTimeMs())
83
84  def Skip(self, test, reason):
85    test_name = _FormatTestName(test)
86    logging.warning('===== SKIPPING TEST %s: %s =====', test_name, reason)
87    self.Success(test)
88