Reformat webkitpy.layout_tests w/ format-webkitpy.

Tools/Scripts/webkitpy/layout_tests/controllers/layout_test_runner.py

 # Copyright (C) 2011 Google Inc. All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions are
 # met:
 #
 #     * Redistributions of source code must retain the above copyright
 # notice, this list of conditions and the following disclaimer.
 #     * Redistributions in binary form must reproduce the above
 # copyright notice, this list of conditions and the following disclaimer
@@ skipping 32 matching lines @@
 _log = logging.getLogger(__name__)
 
 
 TestExpectations = test_expectations.TestExpectations
 
 # Export this so callers don't need to know about message pools.
 WorkerException = message_pool.WorkerException
 
 
 class TestRunInterruptedException(Exception):
+
     """Raised when a test run should be stopped immediately."""
+
     def __init__(self, reason):
         Exception.__init__(self)
         self.reason = reason
         self.msg = reason
 
     def __reduce__(self):
         return self.__class__, (self.reason,)
 
 
 class LayoutTestRunner(object):
+
     def __init__(self, options, port, printer, results_directory, test_is_slow_fn):
         self._options = options
         self._port = port
         self._printer = printer
         self._results_directory = results_directory
         self._test_is_slow = test_is_slow_fn
         self._sharder = Sharder(self._port.split_test, self._options.max_locked_shards)
         self._filesystem = self._port.host.filesystem
 
         self._expectations = None
@@ skipping 17 matching lines @@
         if not retrying:
             self._printer.print_expected(run_results, self._expectations.get_tests_with_result_type)
 
         for test_name in set(tests_to_skip):
             result = test_results.TestResult(test_name)
             result.type = test_expectations.SKIP
             run_results.add(result, expected=True, test_is_slow=self._test_is_slow(test_name))
 
         self._printer.write_update('Sharding tests ...')
         locked_shards, unlocked_shards = self._sharder.shard_tests(test_inputs,
-            int(self._options.child_processes), self._options.fully_parallel,
-            self._options.run_singly or (self._options.batch_size == 1))
+                                                                   int(self._options.child_processes), self._options.fully_parallel,
+                                                                   self._options.run_singly or (self._options.batch_size == 1))
 
         # We don't have a good way to coordinate the workers so that they don't
         # try to run the shards that need a lock. The easiest solution is to
         # run all of the locked shards first.
         all_shards = locked_shards + unlocked_shards
         num_workers = min(num_workers, len(all_shards))
         self._printer.print_workers_and_shards(num_workers, len(all_shards), len(locked_shards))
 
         if self._options.dry_run:
             return run_results
 
         self._printer.write_update('Starting %s ...' % grammar.pluralize('worker', num_workers))
 
         start_time = time.time()
         try:
             with message_pool.get(self, self._worker_factory, num_workers, self._port.host) as pool:
                 pool.run(('test_list', shard.name, shard.test_inputs) for shard in all_shards)
 
             if self._shards_to_redo:
                 num_workers -= len(self._shards_to_redo)
                 if num_workers > 0:
                     with message_pool.get(self, self._worker_factory, num_workers, self._port.host) as pool:
                         pool.run(('test_list', shard.name, shard.test_inputs) for shard in self._shards_to_redo)
-        except TestRunInterruptedException, e:
+        except TestRunInterruptedException as e:
             _log.warning(e.reason)
             run_results.interrupted = True
         except KeyboardInterrupt:
             self._printer.flush()
             self._printer.writeln('Interrupted, exiting ...')
             run_results.keyboard_interrupted = True
-        except Exception, e:
+        except Exception as e:
             _log.debug('%s("%s") raised, exiting' % (e.__class__.__name__, str(e)))
             raise
         finally:
             run_results.run_time = time.time() - start_time
 
         return run_results
 
     def _worker_factory(self, worker_connection):
         results_directory = self._results_directory
         if self._retrying:
             self._filesystem.maybe_make_directory(self._filesystem.join(self._results_directory, 'retries'))
             results_directory = self._filesystem.join(self._results_directory, 'retries')
         return Worker(worker_connection, results_directory, self._options)
 
     def _mark_interrupted_tests_as_skipped(self, run_results):
         for test_input in self._test_inputs:
             if test_input.test_name not in run_results.results_by_name:
                 result = test_results.TestResult(test_input.test_name, [test_failures.FailureEarlyExit()])
                 # FIXME: We probably need to loop here if there are multiple iterations.
                 # FIXME: Also, these results are really neither expected nor unexpected. We probably
                 # need a third type of result.
                 run_results.add(result, expected=False, test_is_slow=self._test_is_slow(test_input.test_name))
 
     def _interrupt_if_at_failure_limits(self, run_results):
         # Note: The messages in this method are constructed to match old-run-webkit-tests
         # so that existing buildbot grep rules work.
         def interrupt_if_at_failure_limit(limit, failure_count, run_results, message):
             if limit and failure_count >= limit:
-                message += " %d tests run." % (run_results.expected + run_results.unexpected)
+                message += ' %d tests run.' % (run_results.expected + run_results.unexpected)
                 self._mark_interrupted_tests_as_skipped(run_results)
                 raise TestRunInterruptedException(message)
 
         interrupt_if_at_failure_limit(
             self._options.exit_after_n_failures,
             run_results.unexpected_failures,
             run_results,
-            "Exiting early after %d failures." % run_results.unexpected_failures)
+            'Exiting early after %d failures.' % run_results.unexpected_failures)
         interrupt_if_at_failure_limit(
             self._options.exit_after_n_crashes_or_timeouts,
             run_results.unexpected_crashes + run_results.unexpected_timeouts,
             run_results,
             # This differs from ORWT because it does not include WebProcess crashes.
-            "Exiting early after %d crashes and %d timeouts." % (run_results.unexpected_crashes, run_results.unexpected_timeouts))
+            'Exiting early after %d crashes and %d timeouts.' % (run_results.unexpected_crashes, run_results.unexpected_timeouts))
 
     def _update_summary_with_result(self, run_results, result):
-        expected = self._expectations.matches_an_expected_result(result.test_name, result.type, self._options.pixel_tests or result.reftest_type, self._options.enable_sanitizer)
+        expected = self._expectations.matches_an_expected_result(
+            result.test_name,
+            result.type,
+            self._options.pixel_tests or result.reftest_type,
+            self._options.enable_sanitizer)
         exp_str = self._expectations.get_expectations_string(result.test_name)
         got_str = self._expectations.expectation_to_string(result.type)
 
         if result.device_failed:
-            self._printer.print_finished_test(result, False, exp_str, "Aborted")
+            self._printer.print_finished_test(result, False, exp_str, 'Aborted')
             return
 
         run_results.add(result, expected, self._test_is_slow(result.test_name))
         self._printer.print_finished_test(result, expected, exp_str, got_str)
         self._interrupt_if_at_failure_limits(run_results)
 
     def handle(self, name, source, *args):
         method = getattr(self, '_handle_' + name)
         if method:
             return method(source, *args)
         raise AssertionError('unknown message %s received from %s, args=%s' % (name, source, repr(args)))
 
     def _handle_started_test(self, worker_name, test_input, test_timeout_sec):
         self._printer.print_started_test(test_input.test_name)
 
     def _handle_finished_test_list(self, worker_name, list_name):
         pass
 
     def _handle_finished_test(self, worker_name, result, log_messages=[]):
         self._update_summary_with_result(self._current_run_results, result)
 
     def _handle_device_failed(self, worker_name, list_name, remaining_tests):
-        _log.warning("%s has failed" % worker_name)
+        _log.warning('%s has failed' % worker_name)
         if remaining_tests:
             self._shards_to_redo.append(TestShard(list_name, remaining_tests))
 
+
 class Worker(object):
+
     def __init__(self, caller, results_directory, options):
         self._caller = caller
         self._worker_number = caller.worker_number
         self._name = caller.name
         self._results_directory = results_directory
         self._options = options
 
         # The remaining fields are initialized in start()
         self._host = None
         self._port = None
@@ skipping 56 matching lines @@
             self._driver = self._port.create_driver(self._worker_number)
 
         if not self._driver:
             # FIXME: Is this the best way to handle a device crashing in the middle of the test, or should we create
             # a new failure type?
             device_failed = True
             return device_failed
 
         self._caller.post('started_test', test_input, test_timeout_sec)
         result = single_test_runner.run_single_test(self._port, self._options, self._results_directory,
-            self._name, self._driver, test_input, stop_when_done)
+                                                    self._name, self._driver, test_input, stop_when_done)
 
         result.shard_name = shard_name
         result.worker_name = self._name
         result.total_run_time = time.time() - start
         result.test_number = self._num_tests
         self._num_tests += 1
         self._caller.post('finished_test', result)
         self._clean_up_after_test(test_input, result)
         return result.device_failed
 
     def stop(self):
-        _log.debug("%s cleaning up" % self._name)
+        _log.debug('%s cleaning up' % self._name)
         self._kill_driver()
 
     def _timeout(self, test_input):
         """Compute the appropriate timeout value for a test."""
         # The driver watchdog uses 2.5x the timeout; we want to be
         # larger than that. We also add a little more padding if we're
         # running tests in a separate thread.
         #
         # Note that we need to convert the test timeout from a
         # string value in milliseconds to a float for Python.
 
         # FIXME: Can we just return the test_input.timeout now?
         driver_timeout_sec = 3.0 * float(test_input.timeout) / 1000.0
 
     def _kill_driver(self):
         # Be careful about how and when we kill the driver; if driver.stop()
         # raises an exception, this routine may get re-entered via __del__.
         driver = self._driver
         self._driver = None
         if driver:
-            _log.debug("%s killing driver" % self._name)
+            _log.debug('%s killing driver' % self._name)
             driver.stop()
 
-
     def _clean_up_after_test(self, test_input, result):
         test_name = test_input.test_name
 
         if result.failures:
             # Check and kill the driver if we need to.
             if any([f.driver_needs_restart() for f in result.failures]):
                 self._kill_driver()
                 # Reset the batch count since the shell just bounced.
                 self._batch_count = 0
 
             # Print the error message(s).
-            _log.debug("%s %s failed:" % (self._name, test_name))
+            _log.debug('%s %s failed:' % (self._name, test_name))
             for f in result.failures:
-                _log.debug("%s  %s" % (self._name, f.message()))
+                _log.debug('%s  %s' % (self._name, f.message()))
         elif result.type == test_expectations.SKIP:
-            _log.debug("%s %s skipped" % (self._name, test_name))
+            _log.debug('%s %s skipped' % (self._name, test_name))
         else:
-            _log.debug("%s %s passed" % (self._name, test_name))
+            _log.debug('%s %s passed' % (self._name, test_name))
 
 
 class TestShard(object):
+
     """A test shard is a named list of TestInputs."""
 
     def __init__(self, name, test_inputs):
         self.name = name
         self.test_inputs = test_inputs
         self.requires_lock = test_inputs[0].requires_lock
 
     def __repr__(self):
         return "TestShard(name='%s', test_inputs=%s, requires_lock=%s'" % (self.name, self.test_inputs, self.requires_lock)
 
     def __eq__(self, other):
         return self.name == other.name and self.test_inputs == other.test_inputs
 
 
 class Sharder(object):
+
     def __init__(self, test_split_fn, max_locked_shards):
         self._split = test_split_fn
         self._max_locked_shards = max_locked_shards
 
     def shard_tests(self, test_inputs, num_workers, fully_parallel, run_singly):
         """Groups tests into batches.
         This helps ensure that tests that depend on each other (aka bad tests!)
         continue to run together as most cross-tests dependencies tend to
         occur within the same directory.
         Return:
@@ skipping 51 matching lines @@
                 # of content_shell, it's too slow to shard them fully.
                 virtual_inputs.append(test_input)
             else:
                 unlocked_shards.append(TestShard('.', [test_input]))
 
         locked_virtual_shards, unlocked_virtual_shards = self._shard_by_directory(virtual_inputs)
 
         # The locked shards still need to be limited to self._max_locked_shards in order to not
         # overload the http server for the http tests.
         return (self._resize_shards(locked_virtual_shards + locked_shards, self._max_locked_shards, 'locked_shard'),
-            unlocked_virtual_shards + unlocked_shards)
+                unlocked_virtual_shards + unlocked_shards)
 
     def _shard_by_directory(self, test_inputs):
         """Returns two lists of shards, each shard containing all the files in a directory.
 
         This is the default mode, and gets as much parallelism as we can while
         minimizing flakiness caused by inter-test dependencies."""
         locked_shards = []
         unlocked_shards = []
         unlocked_slow_shards = []
         tests_by_dir = {}
@@ skipping 54 matching lines @@
         def split_at(seq, index):
             return (seq[:index], seq[index:])
 
         num_old_per_new = divide_and_round_up(len(old_shards), max_new_shards)
         new_shards = []
         remaining_shards = old_shards
         while remaining_shards:
             some_shards, remaining_shards = split_at(remaining_shards, num_old_per_new)
             new_shards.append(TestShard('%s_%d' % (shard_name_prefix, len(new_shards) + 1), extract_and_flatten(some_shards)))
         return new_shards