src/v8/tools/testrunner/local/testsuite.py

# Copyright 2012 the V8 project authors. 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 in the documentation and/or other materials provided
#       with the distribution.
#     * Neither the name of Google Inc. nor the names of its
#       contributors may be used to endorse or promote products derived
#       from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


import fnmatch
import imp
import itertools
import os
from contextlib import contextmanager

from . import command
from . import statusfile
from . import utils
from ..objects.testcase import TestCase
from .variants import ALL_VARIANTS, ALL_VARIANT_FLAGS


STANDARD_VARIANT = set(["default"])


class VariantsGenerator(object):
  def __init__(self, variants):
    self._all_variants = [v for v in variants if v in ALL_VARIANTS]
    self._standard_variant = [v for v in variants if v in STANDARD_VARIANT]

  def gen(self, test):
    """Generator producing (variant, flags, procid suffix) tuples."""
    flags_set = self._get_flags_set(test)
    for n, variant in enumerate(self._get_variants(test)):
      yield (variant, flags_set[variant][0], n)

  def _get_flags_set(self, test):
    return ALL_VARIANT_FLAGS

  def _get_variants(self, test):
    if test.only_standard_variant:
      return self._standard_variant
    return self._all_variants


class TestCombiner(object):
  def get_group_key(self, test):
    """To indicate what tests can be combined with each other we define a group
    key for each test. Tests with the same group key can be combined. Test
    without a group key (None) is not combinable with any other test.
    """
    raise NotImplementedError()

  def combine(self, name, tests):
    """Returns test combined from `tests`. Since we identify tests by their
    suite and name, `name` parameter should be unique within one suite.
    """
    return self._combined_test_class()(name, tests)

  def _combined_test_class(self):
    raise NotImplementedError()


class TestLoader(object):
  """Base class for loading TestSuite tests after applying test suite
  transformations."""

  def __init__(self, suite, test_class, test_config, test_root):
    self.suite = suite
    self.test_class = test_class
    self.test_config = test_config
    self.test_root = test_root
    self.test_count_estimation = len(list(self._list_test_filenames()))

  def _list_test_filenames(self):
    """Implemented by the subclassed TestLoaders to list filenames.

    Filenames are expected to be sorted and are deterministic."""
    raise NotImplementedError

  def _should_filter_by_name(self, name):
    return False

  def _should_filter_by_test(self, test):
    return False

  def _filename_to_testname(self, filename):
    """Hook for subclasses to write their own filename transformation
    logic before the test creation."""
    return filename

  # TODO: not needed for every TestLoader, extract it into a subclass.
  def _path_to_name(self, path):
    if utils.IsWindows():
      return path.replace(os.path.sep, "/")

    return path

  def _create_test(self, path, suite, **kwargs):
    """Converts paths into test objects using the given options"""
    return self.test_class(
      suite, path, self._path_to_name(path), self.test_config, **kwargs)

  def list_tests(self):
    """Loads and returns the test objects for a TestSuite"""
    # TODO: detect duplicate tests.
    for filename in self._list_test_filenames():
      if self._should_filter_by_name(filename):
        continue

      testname = self._filename_to_testname(filename)
      case = self._create_test(testname, self.suite)
      if self._should_filter_by_test(case):
        continue

      yield case


class GenericTestLoader(TestLoader):
  """Generic TestLoader implementing the logic for listing filenames"""
  @property
  def excluded_files(self):
    return set()

  @property
  def excluded_dirs(self):
    return set()

  @property
  def excluded_suffixes(self):
    return set()

  @property
  def test_dirs(self):
    return [self.test_root]

  @property
  def extensions(self):
    return []

  def __find_extension(self, filename):
    for extension in self.extensions:
      if filename.endswith(extension):
        return extension

    return False

  def _should_filter_by_name(self, filename):
    if not self.__find_extension(filename):
      return True

    for suffix in self.excluded_suffixes:
      if filename.endswith(suffix):
        return True

    if os.path.basename(filename) in self.excluded_files:
      return True

    return False

  def _filename_to_testname(self, filename):
    extension = self.__find_extension(filename)
    if not extension:
      return filename

    return filename[:-len(extension)]

  def _to_relpath(self, abspath, test_root):
    return os.path.relpath(abspath, test_root)

  def _list_test_filenames(self):
    for test_dir in sorted(self.test_dirs):
      test_root = os.path.join(self.test_root, test_dir)
      for dirname, dirs, files in os.walk(test_root, followlinks=True):
        dirs.sort()
        for dir in dirs:
          if dir in self.excluded_dirs or dir.startswith('.'):
            dirs.remove(dir)

        files.sort()
        for filename in files:
          abspath = os.path.join(dirname, filename)

          yield self._to_relpath(abspath, test_root)


class JSTestLoader(GenericTestLoader):
  @property
  def extensions(self):
    return [".js", ".mjs"]


class TestGenerator(object):
  def __init__(self, test_count_estimate, slow_tests, fast_tests):
    self.test_count_estimate = test_count_estimate
    self.slow_tests = slow_tests
    self.fast_tests = fast_tests
    self._rebuild_iterator()

  def _rebuild_iterator(self):
    self._iterator = itertools.chain(self.slow_tests, self.fast_tests)

  def __iter__(self):
    return self

  def __next__(self):
    return self.next()

  def next(self):
    return next(self._iterator)

  def merge(self, test_generator):
    self.test_count_estimate += test_generator.test_count_estimate
    self.slow_tests = itertools.chain(
      self.slow_tests, test_generator.slow_tests)
    self.fast_tests = itertools.chain(
      self.fast_tests, test_generator.fast_tests)
    self._rebuild_iterator()


@contextmanager
def _load_testsuite_module(name, root):
  f = None
  try:
    (f, pathname, description) = imp.find_module("testcfg", [root])
    yield imp.load_module(name + "_testcfg", f, pathname, description)
  finally:
    if f:
      f.close()

class TestSuite(object):
  @staticmethod
  def Load(root, test_config, framework_name):
    name = root.split(os.path.sep)[-1]
    with _load_testsuite_module(name, root) as module:
      return module.GetSuite(name, root, test_config, framework_name)

  def __init__(self, name, root, test_config, framework_name):
    self.name = name  # string
    self.root = root  # string containing path
    self.test_config = test_config
    self.framework_name = framework_name  # name of the test runner impl
    self.tests = None  # list of TestCase objects
    self.statusfile = None

    self._test_loader = self._test_loader_class()(
      self, self._test_class(), self.test_config, self.root)

  def status_file(self):
    return "%s/%s.status" % (self.root, self.name)

  @property
  def _test_loader_class(self):
    raise NotImplementedError

  def ListTests(self):
    return self._test_loader.list_tests()

  def __initialize_test_count_estimation(self):
    # Retrieves a single test to initialize the test generator.
    next(iter(self.ListTests()), None)

  def __calculate_test_count(self):
    self.__initialize_test_count_estimation()
    return self._test_loader.test_count_estimation

  def load_tests_from_disk(self, statusfile_variables):
    self.statusfile = statusfile.StatusFile(
      self.status_file(), statusfile_variables)

    test_count = self.__calculate_test_count()
    slow_tests = (test for test in self.ListTests() if test.is_slow)
    fast_tests = (test for test in self.ListTests() if not test.is_slow)
    return TestGenerator(test_count, slow_tests, fast_tests)

  def get_variants_gen(self, variants):
    return self._variants_gen_class()(variants)

  def _variants_gen_class(self):
    return VariantsGenerator

  def test_combiner_available(self):
    return bool(self._test_combiner_class())

  def get_test_combiner(self):
    cls = self._test_combiner_class()
    if cls:
      return cls()
    return None

  def _test_combiner_class(self):
    """Returns Combiner subclass. None if suite doesn't support combining
    tests.
    """
    return None

  def _test_class(self):
    raise NotImplementedError