# Copyright (C) 2012 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 # 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 errno import fnmatch import hashlib import json import logging import re from collections import OrderedDict, defaultdict from typing import Collection, Mapping, NamedTuple, Optional, Sequence, Set, Tuple from blinkpy.web_tests.layout_package.json_results_generator import convert_times_trie_to_flat_paths from blinkpy.web_tests.models.typ_types import ResultType from blinkpy.web_tests.port.base import Port _log = logging.getLogger(__name__) class WebTestFinder(object): def __init__(self, port, options): self._port = port self._options = options self._filesystem = self._port.host.filesystem self.WEB_TESTS_DIRECTORIES = ('src', 'third_party', 'blink', 'web_tests') def find_tests( self, args, test_lists=None, filter_files=None, inverted_filter_files=None, fastest_percentile=None, filters=None, ): filters = filters or [] paths = self._strip_test_dir_prefixes(args) if test_lists: new_paths = self._read_test_list_files( test_lists, self._port.TEST_PATH_SEPARATOR) new_paths = [ self._strip_test_dir_prefix(new_path) for new_path in new_paths ] paths += new_paths all_tests = [] if not paths or fastest_percentile: all_tests = self._port.tests(None) path_tests = [] if paths: path_tests = self._port.tests(paths) test_files = None running_all_tests = False if fastest_percentile: times_trie = self._times_trie() if times_trie: fastest_tests = self._fastest_tests(times_trie, all_tests, fastest_percentile) test_files = list(set(fastest_tests).union(path_tests)) else: _log.warning( 'Running all the tests the first time to generate timing data.' ) test_files = all_tests running_all_tests = True elif paths: test_files = path_tests else: test_files = all_tests running_all_tests = True all_filters = [] if filters: all_filters = [f.split('::') for f in filters] if filter_files: file_filters = self._read_filter_files( filter_files, self._port.TEST_PATH_SEPARATOR) all_filters = all_filters + file_filters if inverted_filter_files: file_filters = self._read_filter_files( inverted_filter_files, self._port.TEST_PATH_SEPARATOR) all_filters = all_filters + self._invert_filters(file_filters) test_files = filter_tests(test_files, all_filters) # de-dupe the test list and paths here before running them. test_files = list(OrderedDict.fromkeys(test_files)) paths = list(OrderedDict.fromkeys(paths)) return (paths, test_files, running_all_tests) def _times_trie(self): times_ms_path = self._port.bot_test_times_path() if self._filesystem.exists(times_ms_path): return json.loads(self._filesystem.read_text_file(times_ms_path)) else: return {} # The following line should run the fastest 50% of tests *and* # the css3/flexbox tests. It should *not* run the fastest 50% # of the css3/flexbox tests. # # run_web_tests.py --fastest=50 css3/flexbox def _fastest_tests(self, times_trie, all_tests, fastest_percentile): times = convert_times_trie_to_flat_paths(times_trie) # Ignore tests with a time==0 because those are skipped tests. sorted_times = sorted([test for (test, time) in times.items() if time], key=lambda t: (times[t], t)) clamped_percentile = max(0, min(100, fastest_percentile)) number_of_tests_to_return = int( len(sorted_times) * clamped_percentile / 100) fastest_tests = set(sorted_times[:number_of_tests_to_return]) # Don't try to run tests in the times_trie that no longer exist, fastest_tests = fastest_tests.intersection(all_tests) # For fastest tests, include any tests not in the times_ms.json so that # new tests get run in the fast set. unaccounted_tests = set(all_tests) - set(times.keys()) # Using a set to dedupe here means that --order=None won't work, but that's # ok because --fastest already runs in an arbitrary order. return list(fastest_tests.union(unaccounted_tests)) def _strip_test_dir_prefixes(self, paths): return [self._strip_test_dir_prefix(path) for path in paths if path] def _strip_test_dir_prefix(self, path): # Remove src/third_party/blink/web_tests/ from the front of the test path, # or any subset of these. for i in range(len(self.WEB_TESTS_DIRECTORIES)): # Handle both "web_tests/foo/bar.html" and "web_tests\foo\bar.html" if # the filesystem uses '\\' as a directory separator for separator in (self._port.TEST_PATH_SEPARATOR, self._filesystem.sep): directory_prefix = separator.join( self.WEB_TESTS_DIRECTORIES[i:]) + separator if path.startswith(directory_prefix): return path[len(directory_prefix):] return path def _read_filter_files(self, filenames, test_path_separator): fs = self._filesystem filters = [] for filename in filenames: file_lines = [] try: if test_path_separator != fs.sep: filename = filename.replace(test_path_separator, fs.sep) file_contents = fs.read_text_file(filename).split('\n') for line in file_contents: line = self._strip_comments(line) if not line: continue file_lines.append(line) filters.append(file_lines) except IOError as error: if error.errno == errno.ENOENT: _log.critical('') _log.critical('--test-launcher-filter-file "%s" not found', filename) raise return filters def _invert_filters(self, filters): inverted_filters = [] for terms in filters: inverted_filter = [] for term in terms: if term.startswith('-'): inverted_filter.append(term[1:]) elif term.startswith('+'): inverted_filter.append('-' + term[1:]) else: inverted_filter.append('-' + term) inverted_filters.append(inverted_filter) return inverted_filters def _read_test_list_files(self, filenames, test_path_separator): fs = self._filesystem positive_matches = [] for filename in filenames: try: if test_path_separator != fs.sep: filename = filename.replace(test_path_separator, fs.sep) file_contents = fs.read_text_file(filename).split('\n') for line in file_contents: line = self._strip_comments(line) if not line: continue if line[0] == '-': _log.debug( 'test-list %s contains a negative filter %s' % (filename, line)) positive_matches.append(line) except IOError as error: if error.errno == errno.ENOENT: _log.critical('') _log.critical('--test-list file "%s" not found', filename) raise return positive_matches @staticmethod def _strip_comments(line): commentIndex = line.find('#') if commentIndex == -1: commentIndex = len(line) line = re.sub(r'\s+', ' ', line[:commentIndex].strip()) if line == '': return None else: return line def skip_tests(self, paths, all_tests_list, expectations): """Given a list of tests, returns the ones that should be skipped. A test may be skipped for many reasons, depending on the expectation files and options selected. The most obvious is SKIP entries in TestExpectations, but we also e.g. skip idlharness tests on MSAN/ASAN due to https://crbug.com/856601. Note that for programmatically added SKIPs, this function will modify the input expectations to include the SKIP expectation (but not write it to disk) Args: paths: the paths passed on the command-line to run_web_tests.py all_tests_list: all tests that we are considering running expectations: parsed TestExpectations data Returns: a set of tests that should be skipped (not run). """ all_tests = set(all_tests_list) tests_to_skip = set() tests_always_skipped = set() for test in all_tests: # Manual tests and virtual tests skipped by platform config are # always skipped and not affected by the --skip parameter if (self._port.skipped_due_to_manual_test(test) or self._port.virtual_test_skipped_due_to_platform_config( test) or self._port.skipped_due_to_exclusive_virtual_tests(test)): tests_always_skipped.update({test}) continue # We always skip idlharness tests for MSAN/ASAN, even when running # with --no-expectations (https://crbug.com/856601). Note we will # run the test anyway if it is explicitly specified on the command # line; paths are removed from the skip list after this loop. if self._options.enable_sanitizer and Port.is_wpt_idlharness_test( test): tests_to_skip.update({test}) continue if self._options.no_expectations: # do not skip anything from TestExpectations continue expected_results = expectations.get_expectations(test).results if ResultType.Skip in expected_results: tests_to_skip.update({test}) if self._options.skip_timeouts and ResultType.Timeout in expected_results: tests_to_skip.update({test}) if self._options.skip_failing_tests and ResultType.Failure in expected_results: tests_to_skip.update({test}) if self._options.skipped == 'only': tests_to_skip = all_tests - tests_to_skip elif self._options.skipped == 'ignore': tests_to_skip = set() elif self._options.skipped != 'always': # make sure we're explicitly running any tests passed on the command line; equivalent to 'default'. tests_to_skip -= set(paths) tests_to_skip.update(tests_always_skipped) return tests_to_skip def split_into_chunks(self, test_names): """split into a list to run and a set to skip, based on --shard_index and --total_shards.""" if self._options.shard_index is None and self._options.total_shards is None: return test_names if self._options.shard_index is None: raise ValueError( 'Must provide --shard-index or GTEST_SHARD_INDEX when sharding.' ) if self._options.total_shards is None: raise ValueError( 'Must provide --total-shards or GTEST_TOTAL_SHARDS when sharding.' ) if self._options.shard_index >= self._options.total_shards: raise ValueError( 'Shard index (%d) should be less than total shards (%d)!' % (self._options.shard_index, self._options.total_shards)) return self._split_into_chunks(test_names, self._options.shard_index, self._options.total_shards) @staticmethod def _split_into_chunks(test_names, index, count): tests_and_indices = [ (test_name, int(hashlib.sha256(test_name.encode('utf-8')).hexdigest(), 16) % count) for test_name in test_names ] tests_to_run = [ test_name for test_name, test_index in tests_and_indices if test_index == index ] _log.debug('chunk %d of %d contains %d tests of %d', index, count, len(tests_to_run), len(test_names)) return tests_to_run def filter_tests(tests, filters): """Returns a filtered list of tests to run. The test-filtering semantics are documented in https://bit.ly/chromium-test-runner-api and https://bit.ly/chromium-test-list-format, but are as follows: A test should be run only if it would be run when every flag (ie filter) is evaluated individually. A test should be skipped if it would be skipped if any flag (filter) was evaluated individually. Each filter is a list of potentially glob expressions, with each expression optionally prefixed by a "-" or "+". If the glob starts with a "-", it is a negative term, otherwise it is a positive term. If multiple filter terms in a flag (filter) match a given test name, the longest match takes priority (longest match wins). The order of the filters should not matter. It is an error to have multiple expressions of the same length that conflict. Globbing is fairly limited; "?" is not allowed, and unescaped "*" must only appear at the end of the glob. If multiple globs match a test, the longest match wins. If both globs are the same length, an error is raised. A test will be run only if it passes every filter. """ for terms in filters: # Although `FilterTrie` also happens to support exact terms, it's # slower than simply checking for membership in a set of test names, so # partition the terms into exact/glob. exact_pos_terms, exact_neg_terms, glob_terms = _extract_terms(terms) include_by_default = not exact_pos_terms and not any( include for _, include in glob_terms) glob_trie = FilterTrie.from_terms(glob_terms) filtered_tests = [] for test in tests: if test in exact_neg_terms: continue if test in exact_pos_terms: filtered_tests.append(test) continue include = glob_trie.should_include( test.split(Port.TEST_PATH_SEPARATOR)) if include is None: include = include_by_default if include: filtered_tests.append(test) tests = filtered_tests return tests ParsedTerm = Tuple[str, bool] class FilterTrie(NamedTuple): children: Mapping[str, 'FilterTrie'] # `None` indicates no term terminates at this node. include: Optional[bool] = None @classmethod def from_terms(cls, terms: Collection[ParsedTerm]) -> 'FilterTrie': include_for_trie, child_terms_by_next_part = None, defaultdict(list) for test, include in terms: next_part, _, rest = test.partition(Port.TEST_PATH_SEPARATOR) if next_part: child_terms_by_next_part[next_part].append((rest, include)) else: assert include_for_trie is None, 'contradictory terms' include_for_trie = include # Pre-sort the children so that `should_include()` honors the # longest-glob-wins rule. children = OrderedDict() for next_part in sorted(child_terms_by_next_part, key=_remove_glob, reverse=True): children[next_part] = cls.from_terms( child_terms_by_next_part[next_part]) return cls(children, include_for_trie) def should_include(self, test_parts: Sequence[str]) -> Optional[bool]: """Determine whether a test should run or not. Arguments: test_parts: The path components of the test. URL query parameters should be attached to the last component. Returns: Whether the test should be included. If `None`, there's no definitive answer because no terms matched. """ if not test_parts: # Exact match found when all path parts have been exhausted. return self.include next_part, *rest = test_parts if child := self.children.get(next_part): # Look for the rest of an exact match or more specific glob, which # always beats this node's glob (or that of one of its ancestors). if (include := child.should_include(rest)) is not None: return include # If there's no exact match for a child directory/test, there can # still be a glob match at this directory. for maybe_glob, child in self.children.items(): if maybe_glob.endswith('*') and next_part.startswith( _remove_glob(maybe_glob)): assert child.include is not None, 'glob must terminate term' return child.include return None def _extract_terms( terms: Collection[str]) -> Tuple[Set[str], Set[str], Set[ParsedTerm]]: """Extract terms of the filter into exact +/- terms and glob terms. The +/- prefix char for exact terms will also be stripped as they are no longer needed to identify the term as a positive or negative filter. Any prefix in globbed terms (terms with a *) will preserve the sign. """ exact_pos_terms, exact_neg_terms, glob_terms = set(), set(), set() for term in terms: test, include = _strip_sign(term) if not test: raise ValueError(f'Empty filter entry {term!r}') if _has_illegal_wildcard(test): raise ValueError(f'Bad test filter {term!r} specified; ' 'unescaped wildcards are only allowed at the end') if test.endswith('*'): glob_terms.add((test, include)) elif include: exact_pos_terms.add(test) else: exact_neg_terms.add(test) contradictory_terms = (exact_pos_terms & exact_neg_terms) | { test for test, include in glob_terms if (test, not include) in glob_terms } if test := next(iter(contradictory_terms), None): raise ValueError( f'Both "+{test}" and "-{test}" specified in test filter') return exact_pos_terms, exact_neg_terms, glob_terms def _strip_sign(term: str) -> ParsedTerm: if term.startswith('-'): return term[len('-'):], False elif term.startswith('+'): return term[len('+'):], True return term, True def _remove_glob(test: str) -> str: return test[:-len('*')] if test.endswith('*') else test def _has_illegal_wildcard(term: str) -> bool: # Remove any legal wildcard and add an initial dummy character, which fails # terms starting with `*` that aren't just `*`. term = ' ' + _remove_glob(term) return any(prev != '\\' and char == '*' for prev, char in zip(term[:-1], term[1:]))