本文整理汇总了Python中webkitpy.layout_tests.views.metered_stream.MeteredStream类的典型用法代码示例。如果您正苦于以下问题:Python MeteredStream类的具体用法?Python MeteredStream怎么用?Python MeteredStream使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了MeteredStream类的17个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_basic
def test_basic(self):
buflist = self._basic([0, 1, 1.05, 1.1, 2])
self.assertEqual(
buflist,
[
"foo",
MeteredStream._erasure("foo"),
"bar",
MeteredStream._erasure("bar"),
"baz 2",
MeteredStream._erasure("baz 2"),
"done\n",
],
)
开发者ID:mirror,项目名称:chromium,代码行数:14,代码来源:metered_stream_unittest.py
示例2: __init__
def __init__(self, port, options, regular_output, buildbot_output, configure_logging):
"""
Args
port interface to port-specific routines
options OptionParser object with command line settings
regular_output stream to which output intended only for humans
should be written
buildbot_output stream to which output intended to be read by
the buildbots (and humans) should be written
configure_loggign Whether a logging handler should be registered
"""
self._port = port
self._options = options
self._stream = regular_output
self._buildbot_stream = buildbot_output
self._meter = None
# These are used for --print one-line-progress
self._last_remaining = None
self._last_update_time = None
self.switches = parse_print_options(options.print_options, options.verbose)
self._logging_handler = None
if self._stream.isatty() and not options.verbose:
self._update_interval_seconds = FAST_UPDATES_SECONDS
self._meter = MeteredStream(self._stream)
if configure_logging:
self._logging_handler = _configure_logging(self._meter, options.verbose)
else:
self._update_interval_seconds = SLOW_UPDATES_SECONDS
if configure_logging:
self._logging_handler = _configure_logging(self._stream, options.verbose)
开发者ID:,项目名称:,代码行数:34,代码来源:
示例3: configure
def configure(self, options):
self.options = options
if options.timing:
# --timing implies --verbose
options.verbose = max(options.verbose, 1)
log_level = logging.INFO
if options.quiet:
log_level = logging.WARNING
elif options.verbose == 2:
log_level = logging.DEBUG
self.meter = MeteredStream(self.stream, (options.verbose == 2),
number_of_columns=SystemHost().platform.terminal_width())
handler = logging.StreamHandler(self.stream)
# We constrain the level on the handler rather than on the root
# logger itself. This is probably better because the handler is
# configured and known only to this module, whereas the root logger
# is an object shared (and potentially modified) by many modules.
# Modifying the handler, then, is less intrusive and less likely to
# interfere with modifications made by other modules (e.g. in unit
# tests).
handler.name = __name__
handler.setLevel(log_level)
formatter = logging.Formatter("%(message)s")
handler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.NOTSET)
# Filter out most webkitpy messages.
#
# Messages can be selectively re-enabled for this script by updating
# this method accordingly.
def filter_records(record):
"""Filter out autoinstall and non-third-party webkitpy messages."""
# FIXME: Figure out a way not to use strings here, for example by
# using syntax like webkitpy.test.__name__. We want to be
# sure not to import any non-Python 2.4 code, though, until
# after the version-checking code has executed.
if (record.name.startswith("webkitpy.common.system.autoinstall") or
record.name.startswith("webkitpy.test")):
return True
if record.name.startswith("webkitpy"):
return False
return True
testing_filter = logging.Filter()
testing_filter.filter = filter_records
# Display a message so developers are not mystified as to why
# logging does not work in the unit tests.
_log.info("Suppressing most webkitpy logging while running unit tests.")
handler.addFilter(testing_filter)
if self.options.pass_through:
outputcapture.OutputCapture.stream_wrapper = _CaptureAndPassThroughStream
开发者ID:AndriyKalashnykov,项目名称:webkit,代码行数:60,代码来源:printer.py
示例4: __init__
def __init__(self, port, options, regular_output, logger=None):
self.num_completed = 0
self.num_tests = 0
self._port = port
self._options = options
self._meter = MeteredStream(regular_output, options.debug_rwt_logging, logger=logger,
number_of_columns=self._port.host.platform.terminal_width())
self._running_tests = []
self._completed_tests = []
开发者ID:dreifachstein,项目名称:chromium-src,代码行数:9,代码来源:printing.py
示例5: setUp
def setUp(self):
self.stream = StringIO.StringIO()
self.buflist = self.stream.buflist
self.stream.isatty = lambda: self.isatty
# configure a logger to test that log calls do normally get included.
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.DEBUG)
self.logger.propagate = False
# add a dummy time counter for a default behavior.
self.times = range(10)
self.meter = MeteredStream(self.stream, self.verbose, self.logger, self.time_fn, 8675)
开发者ID:Drakey83,项目名称:steamlink-sdk,代码行数:14,代码来源:metered_stream_unittest.py
示例6: __init__
def __init__(self, port, options, regular_output, buildbot_output, logger=None):
"""
Args
port interface to port-specific routines
options OptionParser object with command line settings
regular_output stream to which output intended only for humans
should be written
buildbot_output stream to which output intended to be read by
the buildbots (and humans) should be written
logger optional logger to integrate into the stream.
"""
self._port = port
self._options = options
self._buildbot_stream = buildbot_output
self._meter = MeteredStream(regular_output, options.verbose, logger=logger)
self.switches = parse_print_options(options.print_options, options.verbose)
开发者ID:Moondee,项目名称:Artemis,代码行数:16,代码来源:printing.py
示例7: test_logging_not_included
def test_logging_not_included(self):
# This tests that if we don't hand a logger to the MeteredStream,
# nothing is logged.
logging_stream = StringIO.StringIO()
handler = logging.StreamHandler(logging_stream)
root_logger = logging.getLogger()
orig_level = root_logger.level
root_logger.addHandler(handler)
root_logger.setLevel(logging.DEBUG)
try:
self.meter = MeteredStream(self.stream, self.verbose, None, self.time_fn, 8675)
self.meter.write_throttled_update('foo')
self.meter.write_update('bar')
self.meter.write('baz')
self.assertEqual(logging_stream.buflist, [])
finally:
root_logger.removeHandler(handler)
root_logger.setLevel(orig_level)
开发者ID:Drakey83,项目名称:steamlink-sdk,代码行数:18,代码来源:metered_stream_unittest.py
示例8: Printer
class Printer(object):
"""Class handling all non-debug-logging printing done by run-webkit-tests.
Printing from run-webkit-tests falls into two buckets: general or
regular output that is read only by humans and can be changed at any
time, and output that is parsed by buildbots (and humans) and hence
must be changed more carefully and in coordination with the buildbot
parsing code (in chromium.org's buildbot/master.chromium/scripts/master/
log_parser/webkit_test_command.py script).
By default the buildbot-parsed code gets logged to stdout, and regular
output gets logged to stderr."""
def __init__(self, port, options, regular_output, buildbot_output, logger=None):
self.num_completed = 0
self.num_tests = 0
self._port = port
self._options = options
self._buildbot_stream = buildbot_output
self._meter = MeteredStream(regular_output, options.debug_rwt_logging, logger=logger)
self._running_tests = []
self._completed_tests = []
def cleanup(self):
self._meter.cleanup()
def __del__(self):
self.cleanup()
def print_config(self):
self._print_default("Using port '%s'" % self._port.name())
self._print_default("Test configuration: %s" % self._port.test_configuration())
self._print_default("Placing test results in %s" % self._options.results_directory)
# FIXME: should these options be in printing_options?
if self._options.new_baseline:
self._print_default("Placing new baselines in %s" % self._port.baseline_path())
fs = self._port.host.filesystem
fallback_path = [fs.split(x)[1] for x in self._port.baseline_search_path()]
self._print_default("Baseline search path: %s -> generic" % " -> ".join(fallback_path))
self._print_default("Using %s build" % self._options.configuration)
if self._options.pixel_tests:
self._print_default("Pixel tests enabled")
else:
self._print_default("Pixel tests disabled")
self._print_default(
"Regular timeout: %s, slow test timeout: %s" % (self._options.time_out_ms, self._options.slow_time_out_ms)
)
self._print_default("Command line: " + " ".join(self._port.driver_cmd_line()))
self._print_default("")
def print_found(self, num_all_test_files, num_to_run, repeat_each, iterations):
num_unique_tests = num_to_run / (repeat_each * iterations)
found_str = "Found %s; running %d" % (grammar.pluralize("test", num_all_test_files), num_unique_tests)
if repeat_each * iterations > 1:
found_str += " (%d times each: --repeat-each=%d --iterations=%d)" % (
repeat_each * iterations,
repeat_each,
iterations,
)
found_str += ", skipping %d" % (num_all_test_files - num_unique_tests)
self._print_default(found_str + ".")
def print_expected(self, result_summary, tests_with_result_type_callback):
self._print_expected_results_of_type(
result_summary, test_expectations.PASS, "passes", tests_with_result_type_callback
)
self._print_expected_results_of_type(
result_summary, test_expectations.FAIL, "failures", tests_with_result_type_callback
)
self._print_expected_results_of_type(
result_summary, test_expectations.FLAKY, "flaky", tests_with_result_type_callback
)
self._print_debug("")
def print_workers_and_shards(self, num_workers, num_shards, num_locked_shards):
driver_name = self._port.driver_name()
if num_workers == 1:
self._print_default("Running 1 %s over %s." % (driver_name, grammar.pluralize("shard", num_shards)))
else:
self._print_default(
"Running %d %ss in parallel over %d shards (%d locked)."
% (num_workers, driver_name, num_shards, num_locked_shards)
)
self._print_default("")
def _print_expected_results_of_type(
self, result_summary, result_type, result_type_str, tests_with_result_type_callback
):
tests = tests_with_result_type_callback(result_type)
now = result_summary.tests_by_timeline[test_expectations.NOW]
wontfix = result_summary.tests_by_timeline[test_expectations.WONTFIX]
# We use a fancy format string in order to print the data out in a
# nicely-aligned table.
fmtstr = "Expect: %%5d %%-8s (%%%dd now, %%%dd wontfix)" % (self._num_digits(now), self._num_digits(wontfix))
#.........这里部分代码省略.........
开发者ID:,项目名称:,代码行数:101,代码来源:
示例9: RegularTest
class RegularTest(unittest.TestCase):
verbose = False
isatty = False
def setUp(self):
self.stream = StringIO.StringIO()
self.buflist = self.stream.buflist
self.stream.isatty = lambda: self.isatty
# configure a logger to test that log calls do normally get included.
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.DEBUG)
self.logger.propagate = False
# add a dummy time counter for a default behavior.
self.times = range(10)
self.meter = MeteredStream(self.stream, self.verbose, self.logger, self.time_fn, 8675)
def tearDown(self):
if self.meter:
self.meter.cleanup()
self.meter = None
def time_fn(self):
return self.times.pop(0)
def test_logging_not_included(self):
# This tests that if we don't hand a logger to the MeteredStream,
# nothing is logged.
logging_stream = StringIO.StringIO()
handler = logging.StreamHandler(logging_stream)
root_logger = logging.getLogger()
orig_level = root_logger.level
root_logger.addHandler(handler)
root_logger.setLevel(logging.DEBUG)
try:
self.meter = MeteredStream(self.stream, self.verbose, None, self.time_fn, 8675)
self.meter.write_throttled_update('foo')
self.meter.write_update('bar')
self.meter.write('baz')
self.assertEqual(logging_stream.buflist, [])
finally:
root_logger.removeHandler(handler)
root_logger.setLevel(orig_level)
def _basic(self, times):
self.times = times
self.meter.write_update('foo')
self.meter.write_update('bar')
self.meter.write_throttled_update('baz')
self.meter.write_throttled_update('baz 2')
self.meter.writeln('done')
self.assertEqual(self.times, [])
return self.buflist
def test_basic(self):
buflist = self._basic([0, 1, 2, 13, 14])
self.assertEqual(buflist, ['foo\n', 'bar\n', 'baz 2\n', 'done\n'])
def _log_after_update(self):
self.meter.write_update('foo')
self.logger.info('bar')
return self.buflist
def test_log_after_update(self):
buflist = self._log_after_update()
self.assertEqual(buflist, ['foo\n', 'bar\n'])
def test_log_args(self):
self.logger.info('foo %s %d', 'bar', 2)
self.assertEqual(self.buflist, ['foo bar 2\n'])
开发者ID:Drakey83,项目名称:steamlink-sdk,代码行数:72,代码来源:metered_stream_unittest.py
示例10: test_regular
def test_regular(self):
a = ArrayStream()
m = MeteredStream(stream=a)
self.assertTrue(a.empty())
# basic test
m.write("foo")
exp = ['foo']
self.assertEquals(a.get(), exp)
# now check that a second write() does not overwrite the first.
m.write("bar")
exp.append('bar')
self.assertEquals(a.get(), exp)
m.update("batter")
exp.append('batter')
self.assertEquals(a.get(), exp)
# The next update() should overwrite the laste update() but not the
# other text. Note that the cursor is effectively positioned at the
# end of 'foo', even though we had to erase three more characters.
m.update("foo")
exp.append('\b\b\b\b\b\b \b\b\b\b\b\b')
exp.append('foo')
self.assertEquals(a.get(), exp)
# now check that a write() does overwrite the update
m.write("foo")
exp.append('\b\b\b \b\b\b')
exp.append('foo')
self.assertEquals(a.get(), exp)
# Now test that we only back up to the most recent newline.
# Note also that we do not back up to erase the most recent write(),
# i.e., write()s do not get erased.
a.reset()
m.update("foo\nbar")
m.update("baz")
self.assertEquals(a.get(), ['foo\nbar', '\b\b\b \b\b\b', 'baz'])
开发者ID:Andolamin,项目名称:LunaSysMgr,代码行数:41,代码来源:metered_stream_unittest.py
示例11: test_basic
def test_basic(self):
buflist = self._basic([0, 1, 1.05, 1.1, 2])
self.assertEqual(buflist, ['foo',
MeteredStream._erasure('foo'), 'bar',
MeteredStream._erasure('bar'), 'baz 2',
MeteredStream._erasure('baz 2'), 'done\n'])
开发者ID:Drakey83,项目名称:steamlink-sdk,代码行数:6,代码来源:metered_stream_unittest.py
示例12: test_log_after_update
def test_log_after_update(self):
buflist = self._log_after_update()
self.assertEqual(buflist, ["foo", MeteredStream._erasure("foo"), "bar\n"])
开发者ID:mirror,项目名称:chromium,代码行数:3,代码来源:metered_stream_unittest.py
示例13: Printer
class Printer(object):
"""Class handling all non-debug-logging printing done by run-webkit-tests.
Printing from run-webkit-tests falls into two buckets: general or
regular output that is read only by humans and can be changed at any
time, and output that is parsed by buildbots (and humans) and hence
must be changed more carefully and in coordination with the buildbot
parsing code (in chromium.org's buildbot/master.chromium/scripts/master/
log_parser/webkit_test_command.py script).
By default the buildbot-parsed code gets logged to stdout, and regular
output gets logged to stderr."""
def __init__(self, port, options, regular_output, buildbot_output, logger=None):
"""
Args
port interface to port-specific routines
options OptionParser object with command line settings
regular_output stream to which output intended only for humans
should be written
buildbot_output stream to which output intended to be read by
the buildbots (and humans) should be written
logger optional logger to integrate into the stream.
"""
self._port = port
self._options = options
self._buildbot_stream = buildbot_output
self._meter = MeteredStream(regular_output, options.verbose, logger=logger)
self.switches = parse_print_options(options.print_options, options.verbose)
def cleanup(self):
self._meter.cleanup()
def __del__(self):
self.cleanup()
# These two routines just hide the implementation of the switches.
def disabled(self, option):
return not option in self.switches
def enabled(self, option):
return option in self.switches
def help_printing(self):
self._write(HELP_PRINTING)
def print_actual(self, msg):
if self.disabled('actual'):
return
self._buildbot_stream.write("%s\n" % msg)
def print_config(self, msg):
self.write(msg, 'config')
def print_expected(self, msg):
self.write(msg, 'expected')
def print_timing(self, msg):
self.write(msg, 'timing')
def print_one_line_summary(self, total, expected, unexpected):
"""Print a one-line summary of the test run to stdout.
Args:
total: total number of tests run
expected: number of expected results
unexpected: number of unexpected results
"""
if self.disabled('one-line-summary'):
return
incomplete = total - expected - unexpected
if incomplete:
self._write("")
incomplete_str = " (%d didn't run)" % incomplete
expected_str = str(expected)
else:
incomplete_str = ""
expected_str = "All %d" % expected
if unexpected == 0:
self._write("%s tests ran as expected%s." %
(expected_str, incomplete_str))
elif expected == 1:
self._write("1 test ran as expected, %d didn't%s:" %
(unexpected, incomplete_str))
else:
self._write("%d tests ran as expected, %d didn't%s:" %
(expected, unexpected, incomplete_str))
self._write("")
def print_test_result(self, result, expected, exp_str, got_str):
"""Print the result of the test as determined by --print.
This routine is used to print the details of each test as it completes.
Args:
result - The actual TestResult object
expected - Whether the result we got was an expected result
exp_str - What we expected to get (used for tracing)
got_str - What we actually got (used for tracing)
#.........这里部分代码省略.........
开发者ID:Moondee,项目名称:Artemis,代码行数:101,代码来源:printing.py
示例14: Printer
class Printer(object):
"""Class handling all non-debug-logging printing done by run-webkit-tests.
Printing from run-webkit-tests falls into two buckets: general or
regular output that is read only by humans and can be changed at any
time, and output that is parsed by buildbots (and humans) and hence
must be changed more carefully and in coordination with the buildbot
parsing code (in chromium.org's buildbot/master.chromium/scripts/master/
log_parser/webkit_test_command.py script).
By default the buildbot-parsed code gets logged to stdout, and regular
output gets logged to stderr."""
def __init__(self, port, options, regular_output, buildbot_output, logger=None):
"""
Args
port interface to port-specific routines
options OptionParser object with command line settings
regular_output stream to which output intended only for humans
should be written
buildbot_output stream to which output intended to be read by
the buildbots (and humans) should be written
logger optional logger to integrate into the stream.
"""
self._port = port
self._options = options
self._buildbot_stream = buildbot_output
self._meter = MeteredStream(regular_output, options.verbose, logger=logger)
self.switches = parse_print_options(options.print_options, options.verbose)
def cleanup(self):
self._meter.cleanup()
def __del__(self):
self.cleanup()
# These two routines just hide the implementation of the switches.
def disabled(self, option):
return not option in self.switches
def enabled(self, option):
return option in self.switches
def help_printing(self):
self._write(HELP_PRINTING)
def print_config(self):
"""Prints the configuration for the test run."""
self._print_config("Using port '%s'" % self._port.name())
self._print_config("Test configuration: %s" % self._port.test_configuration())
self._print_config("Placing test results in %s" % self._options.results_directory)
# FIXME: should these options be in printing_options?
if self._options.new_baseline:
self._print_config("Placing new baselines in %s" % self._port.baseline_path())
fs = self._port.host.filesystem
fallback_path = [fs.split(x)[1] for x in self._port.baseline_search_path()]
self._print_config("Baseline search path: %s -> generic" % " -> ".join(fallback_path))
self._print_config("Using %s build" % self._options.configuration)
if self._options.pixel_tests:
self._print_config("Pixel tests enabled")
else:
self._print_config("Pixel tests disabled")
self._print_config("Regular timeout: %s, slow test timeout: %s" %
(self._options.time_out_ms, self._options.slow_time_out_ms))
self._print_config('Command line: ' + ' '.join(self._port.driver_cmd_line()))
self._print_config('')
def print_found(self, num_all_test_files, num_to_run, repeat_each, iterations):
found_str = 'Found %s; running %d' % (grammar.pluralize('test', num_all_test_files), num_to_run)
if repeat_each * iterations > 1:
found_str += ' (%d times each: --repeat-each=%d --iterations=%d)' % (repeat_each * iterations, repeat_each, iterations)
found_str += ', skipping %d' % (num_all_test_files - num_to_run)
self._print_expected(found_str + '.')
def print_expected(self, result_summary, tests_with_result_type_callback):
self._print_expected_results_of_type(result_summary, test_expectations.PASS, "passes", tests_with_result_type_callback)
self._print_expected_results_of_type(result_summary, test_expectations.FAIL, "failures", tests_with_result_type_callback)
self._print_expected_results_of_type(result_summary, test_expectations.FLAKY, "flaky", tests_with_result_type_callback)
self._print_expected('')
def print_workers_and_shards(self, num_workers, num_shards, num_locked_shards):
driver_name = self._port.driver_name()
if num_workers == 1:
self._print_config("Running 1 %s over %s." %
(driver_name, grammar.pluralize('shard', num_shards)))
else:
self._print_config("Running %d %ss in parallel over %d shards (%d locked)." %
(num_workers, driver_name, num_shards, num_locked_shards))
self._print_config('')
def _print_expected_results_of_type(self, result_summary,
result_type, result_type_str, tests_with_result_type_callback):
"""Print the number of the tests in a given result class.
Args:
result_summary - the object containing all the results to report on
#.........这里部分代码省略.........
开发者ID:kseo,项目名称:webkit,代码行数:101,代码来源:printing.py
示例15: Printer
class Printer(object):
"""Class handling all non-debug-logging printing done by run-webkit-tests."""
def __init__(self, port, options, regular_output, logger=None):
self.num_completed = 0
self.num_tests = 0
self._port = port
self._options = options
self._meter = MeteredStream(regular_output, options.debug_rwt_logging, logger=logger,
number_of_columns=self._port.host.platform.terminal_width())
self._running_tests = []
self._completed_tests = []
def cleanup(self):
self._meter.cleanup()
def __del__(self):
self.cleanup()
def print_config(self, results_directory):
self._print_default("Using port '%s'" % self._port.name())
self._print_default("Test configuration: %s" % self._port.test_configuration())
self._print_default("View the test results at file://%s/results.html" % results_directory)
self._print_default("View the archived results dashboard at file://%s/dashboard.html" % results_directory)
# FIXME: should these options be in printing_options?
if self._options.new_baseline:
self._print_default("Placing new baselines in %s" % self._port.baseline_path())
fs = self._port.host.filesystem
fallback_path = [fs.split(x)[1] for x in self._port.baseline_search_path()]
self._print_default("Baseline search path: %s -> generic" % " -> ".join(fallback_path))
self._print_default("Using %s build" % self._options.configuration)
if self._options.pixel_tests:
self._print_default("Pixel tests enabled")
else:
self._print_default("Pixel tests disabled")
self._print_default("Regular timeout: %s, slow test timeout: %s" %
(self._options.time_out_ms, self._options.slow_time_out_ms))
self._print_default('Command line: ' + ' '.join(self._port.driver_cmd_line()))
self._print_default('')
def print_found(self, num_all_test_files, num_to_run, repeat_each, iterations):
found_str = 'Found %s; running %d' % (grammar.pluralize('test', num_all_test_files), num_to_run)
if repeat_each * iterations > 1:
found_str += ' (%d times each: --repeat-each=%d --iterations=%d)' % (repeat_each * iterations, repeat_each, iterations)
found_str += ', skipping %d' % (num_all_test_files - num_to_run)
self._print_default(found_str + '.')
def print_expected(self, run_results, tests_with_result_type_callback):
self._print_expected_results_of_type(run_results, test_expectations.PASS, "passes", tests_with_result_type_callback)
self._print_expected_results_of_type(run_results, test_expectations.FAIL, "failures", tests_with_result_type_callback)
self._print_expected_results_of_type(run_results, test_expectations.FLAKY, "flaky", tests_with_result_type_callback)
self._print_debug('')
def print_workers_and_shards(self, num_workers, num_shards, num_locked_shards):
driver_name = self._port.driver_name()
if num_workers == 1:
self._print_default("Running 1 %s." % driver_name)
self._print_debug("(%s)." % grammar.pluralize('shard', num_shards))
else:
self._print_default("Running %d %ss in parallel." % (num_workers, driver_name))
self._print_debug("(%d shards; %d locked)." % (num_shards, num_locked_shards))
self._print_default('')
def _print_expected_results_of_type(self, run_results, result_type, result_type_str, tests_with_result_type_callback):
tests = tests_with_result_type_callback(result_type)
now = run_results.tests_by_timeline[test_expectations.NOW]
wontfix = run_results.tests_by_timeline[test_expectations.WONTFIX]
# We use a fancy format string in order to print the data out in a
# nicely-aligned table.
fmtstr = ("Expect: %%5d %%-8s (%%%dd now, %%%dd wontfix)"
% (self._num_digits(now), self._num_digits(wontfix)))
self._print_debug(fmtstr % (len(tests), result_type_str, len(tests & now), len(tests & wontfix)))
def _num_digits(self, num):
ndigits = 1
if len(num):
ndigits = int(math.log10(len(num))) + 1
return ndigits
def print_results(self, run_time, run_results, summarized_results):
self._print_timing_statistics(run_time, run_results)
self._print_one_line_summary(run_time, run_results)
def _print_timing_statistics(self, total_time, run_results):
self._print_debug("Test timing:")
self._print_debug(" %6.2f total testing time" % total_time)
self._print_debug("")
self._print_worker_statistics(run_results, int(self._options.child_processes))
self._print_aggregate_test_statistics(run_results)
self._print_individual_test_times(run_results)
self._print_directory_timings(run_results)
def _print_worker_statistics(self, run_results, num_workers):
#.........这里部分代码省略.........
开发者ID:dreifachstein,项目名称:chromium-src,代码行数:101,代码来源:printing.py
示例16: test_log_after_update
def test_log_after_update(self):
buflist = self._log_after_update()
self.assertEqual(buflist, ['foo',
MeteredStream._erasure('foo'), 'bar\n'])
开发者ID:Drakey83,项目名称:steamlink-sdk,代码行数:4,代码来源:metered_stream_unittest.py
示例17: Printer
class Printer(object):
def __init__(self, stream, options=None):
self.stream = stream
self.meter = None
self.options = options
self.num_tests = 0
self.num_completed = 0
self.num_errors = 0
self.num_failures = 0
self.running_tests = []
self.completed_tests = []
if options:
self.configure(options)
def configure(self, options):
self.options = options
if options.timing:
# --timing implies --verbose
options.verbose = max(options.verbose, 1)
log_level = logging.INFO
if options.quiet:
log_level = logging.WARNING
elif options.verbose == 2:
log_level = logging.DEBUG
self.meter = MeteredStream(self.stream, (options.verbose == 2))
handler = logging.StreamHandler(self.stream)
# We constrain the level on the handler rather than on the root
# logger itself. This is probably better because the handler is
# configured and known only to this module, whereas the root logger
# is an object shared (and potentially modified) by many modules.
# Modifying the handler, then, is less intrusive and less likely to
# interfere with modifications made by other modules (e.g. in unit
# tests).
handler.name = __name__
handler.setLevel(log_level)
formatter = logging.Formatter("%(message)s")
handler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.NOTSET)
# Filter out most webkitpy messages.
#
# Messages can be selectively re-enabled for this script by updating
# this method accordingly.
def filter_records(record):
"""Filter out autoinstall and non-third-party webkitpy messages."""
# FIXME: Figure out a way not to use strings here, for example by
# using syntax like webkitpy.test.__name__. We want to be
# sure not to import any non-Python 2.4 code, though, until
# after the version-checking code has executed.
if (record.name.startswith("webkitpy.common.system.autoinstall") or
record.name.startswith("webkitpy.test")):
return True
if record.name.startswith("webkitpy"):
return False
return True
testing_filter = logging.Filter()
testing_filter.filter = filter_records
# Display a message so developers are not mystified as to why
# logging does not work in the unit tests.
_log.info("Suppressing most webkitpy logging while running unit tests.")
handler.addFilter(testing_filter)
if self.options.pass_through:
outputcapture.OutputCapture.stream_wrapper = _CaptureAndPassThroughStream
def write_update(self, msg):
self.meter.write_update(msg)
def print_started_test(self, source, test_name):
self.running_tests.append(test_name)
if len(self.running_tests) > 1:
suffix = ' (+%d)' % (len(self.running_tests) - 1)
else:
suffix = ''
if self.options.verbose:
write = self.meter.write_update
else:
write = self.meter.write_throttled_update
write(self._test_line(self.running_tests[0], suffix))
def print_finished_test(self, source, test_name, test_time, failures, errors):
write = self.meter.writeln
if failures:
lines = failures[0].splitlines() + ['']
suffix = ' failed:'
self.num_failures += 1
elif errors:
lines = errors[0].splitlines() + ['']
suffix = ' erred:'
#.........这里部分代码省略.........
开发者ID:EQ4,项目名称:h5vcc,代码行数:101,代码来源:printer.py
注:本文中的webkitpy.layout_tests.views.metered_stream.MeteredStream类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论