third_party/v8/tools/testrunner/testproc/combiner.py

# Copyright 2018 the V8 project authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

# for py2/py3 compatibility
from __future__ import print_function

from collections import defaultdict
import time

from . import base
from ..objects import testcase
from ..outproc import base as outproc

class CombinerProc(base.TestProc):
  def __init__(self, rng, min_group_size, max_group_size, count):
    """
    Args:
      rng: random number generator
      min_group_size: minimum number of tests to combine
      max_group_size: maximum number of tests to combine
      count: how many tests to generate. 0 means infinite running
    """
    super(CombinerProc, self).__init__()

    self._rng = rng
    self._min_size = min_group_size
    self._max_size = max_group_size
    self._count = count

    # Index of the last generated test
    self._current_num = 0

    # {suite name: instance of TestGroups}
    self._groups = defaultdict(TestGroups)

    # {suite name: instance of TestCombiner}
    self._combiners = {}

  def setup(self, requirement=base.DROP_RESULT):
    # Combiner is not able to pass results (even as None) to the previous
    # processor.
    assert requirement == base.DROP_RESULT
    self._next_proc.setup(base.DROP_RESULT)

  def next_test(self, test):
    group_key = self._get_group_key(test)
    if not group_key:
      # Test not suitable for combining
      return False

    self._groups[test.suite.name].add_test(group_key, test)
    return True

  def _get_group_key(self, test):
    combiner =  self._get_combiner(test.suite)
    if not combiner:
      print ('>>> Warning: There is no combiner for %s testsuite' %
             test.suite.name)
      return None
    return combiner.get_group_key(test)

  def result_for(self, test, result):
    self._send_next_test()

  def generate_initial_tests(self, num=1):
    for _ in range(0, num):
      self._send_next_test()

  def _send_next_test(self):
    if self.is_stopped:
      return False

    if self._count and self._current_num >= self._count:
      return False

    combined_test = self._create_new_test()
    if not combined_test:
      # Not enough tests
      return False

    return self._send_test(combined_test)

  def _create_new_test(self):
    suite, combiner = self._select_suite()
    groups = self._groups[suite]

    max_size = self._rng.randint(self._min_size, self._max_size)
    sample = groups.sample(self._rng, max_size)
    if not sample:
      return None

    self._current_num += 1
    return combiner.combine('%s-%d' % (suite, self._current_num), sample)

  def _select_suite(self):
    """Returns pair (suite name, combiner)."""
    selected = self._rng.randint(0, len(self._groups) - 1)
    for n, suite in enumerate(self._groups):
      if n == selected:
        return suite, self._combiners[suite]

  def _get_combiner(self, suite):
    combiner = self._combiners.get(suite.name)
    if not combiner:
      combiner = suite.get_test_combiner()
      self._combiners[suite.name] = combiner
    return combiner


class TestGroups(object):
  def __init__(self):
    self._groups = defaultdict(list)
    self._keys = []

  def add_test(self, key, test):
    self._groups[key].append(test)
    self._keys.append(key)

  def sample(self, rng, max_size):
    # Not enough tests
    if not self._groups:
      return None

    group_key = rng.choice(self._keys)
    tests = self._groups[group_key]
    return [rng.choice(tests) for _ in range(0, max_size)]