Method _test

deps/v8/tools/testrunner/testproc/sequence_test.py:64–90

(self, tests, batch_size, max_heavy)

class TestSequenceProc(unittest.TestCase):

  def _test(self, tests, batch_size, max_heavy):
    # Set up a simple processing pipeline:
    # Loader -> observe results -> sequencer -> execution.
    loader = LoadProc(iter(tests), initial_batch_size=batch_size)
    results = FakeResultObserver()
    sequence_proc = SequenceProc(max_heavy)
    execution = FakeExecutionProc()
    loader.connect_to(results)
    results.connect_to(sequence_proc)
    sequence_proc.connect_to(execution)

    # Fill the execution queue (with the number of tests potentially
    # executed in parallel).
    loader.load_initial_tests()

    # Simulate the execution test by test.
    while execution.tests:
      # Assert the invariant of maximum heavy tests executed simultaneously.
      self.assertLessEqual(
          sum(int(test.is_heavy) for test in execution.tests), max_heavy)

      # As in the real pipeline, running a test and returning its result
      # will add another test into the pipeline.
      execution.run()

    # Ensure that all tests are processed and deliver results.
    self.assertEqual(set(test.n for test in tests), results.tests)

  def test_wrong_usage(self):
    with self.assertRaises(Exception):