123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562 |
- # Copyright 2020 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- """Methods related to test expectations/expectation files."""
- from __future__ import print_function
- import collections
- import datetime
- import logging
- import os
- import re
- import subprocess
- import sys
- from typing import Iterable, List, Optional, Set, Tuple, Union
- import six
- from typ import expectations_parser
- from unexpected_passes_common import data_types
- from unexpected_passes_common import result_output
- FINDER_DISABLE_COMMENT_BASE = 'finder:disable'
- FINDER_ENABLE_COMMENT_BASE = 'finder:enable'
- FINDER_COMMENT_SUFFIX_GENERAL = '-general'
- FINDER_COMMENT_SUFFIX_STALE = '-stale'
- FINDER_COMMENT_SUFFIX_UNUSED = '-unused'
- ALL_FINDER_SUFFIXES = frozenset([
- FINDER_COMMENT_SUFFIX_GENERAL,
- FINDER_COMMENT_SUFFIX_STALE,
- FINDER_COMMENT_SUFFIX_UNUSED,
- ])
- FINDER_DISABLE_COMMENT_GENERAL = (FINDER_DISABLE_COMMENT_BASE +
- FINDER_COMMENT_SUFFIX_GENERAL)
- FINDER_DISABLE_COMMENT_STALE = (FINDER_DISABLE_COMMENT_BASE +
- FINDER_COMMENT_SUFFIX_STALE)
- FINDER_DISABLE_COMMENT_UNUSED = (FINDER_DISABLE_COMMENT_BASE +
- FINDER_COMMENT_SUFFIX_UNUSED)
- FINDER_ENABLE_COMMENT_GENERAL = (FINDER_ENABLE_COMMENT_BASE +
- FINDER_COMMENT_SUFFIX_GENERAL)
- FINDER_ENABLE_COMMENT_STALE = (FINDER_ENABLE_COMMENT_BASE +
- FINDER_COMMENT_SUFFIX_STALE)
- FINDER_ENABLE_COMMENT_UNUSED = (FINDER_ENABLE_COMMENT_BASE +
- FINDER_COMMENT_SUFFIX_UNUSED)
- FINDER_DISABLE_COMMENTS = frozenset([
- FINDER_DISABLE_COMMENT_GENERAL, FINDER_DISABLE_COMMENT_STALE,
- FINDER_DISABLE_COMMENT_UNUSED
- ])
- FINDER_ENABLE_COMMENTS = frozenset([
- FINDER_ENABLE_COMMENT_GENERAL,
- FINDER_ENABLE_COMMENT_STALE,
- FINDER_ENABLE_COMMENT_UNUSED,
- ])
- ALL_FINDER_COMMENTS = frozenset(FINDER_DISABLE_COMMENTS
- | FINDER_ENABLE_COMMENTS)
- GIT_BLAME_REGEX = re.compile(
- r'^[\w\s]+\(.+(?P<date>\d\d\d\d-\d\d-\d\d)[^\)]+\)(?P<content>.*)$',
- re.DOTALL)
- EXPECTATION_LINE_REGEX = re.compile(r'^.*\[ .* \] .* \[ \w* \].*$', re.DOTALL)
- # pylint: disable=useless-object-inheritance
- class RemovalType(object):
- STALE = FINDER_COMMENT_SUFFIX_STALE
- UNUSED = FINDER_COMMENT_SUFFIX_UNUSED
- class Expectations(object):
- def CreateTestExpectationMap(
- self, expectation_files: Optional[Union[str, List[str]]],
- tests: Optional[Iterable[str]],
- grace_period: int) -> data_types.TestExpectationMap:
- """Creates an expectation map based off a file or list of tests.
- Args:
- expectation_files: A filepath or list of filepaths to expectation files to
- read from, or None. If a filepath is specified, |tests| must be None.
- tests: An iterable of strings containing test names to check. If
- specified, |expectation_file| must be None.
- grace_period: An int specifying how many days old an expectation must
- be in order to be parsed, i.e. how many days old an expectation must
- be before it is a candidate for removal/modification.
- Returns:
- A data_types.TestExpectationMap, although all its BuilderStepMap contents
- will be empty.
- """
- def AddContentToMap(content: str, ex_map: data_types.TestExpectationMap,
- expectation_file_name: str) -> None:
- list_parser = expectations_parser.TaggedTestListParser(content)
- expectations_for_file = ex_map.setdefault(
- expectation_file_name, data_types.ExpectationBuilderMap())
- logging.debug('Parsed %d expectations', len(list_parser.expectations))
- for e in list_parser.expectations:
- if 'Skip' in e.raw_results:
- continue
- # Expectations that only have a Pass expectation (usually used to
- # override a broader, failing expectation) are not handled by the
- # unexpected pass finder, so ignore those.
- if e.raw_results == ['Pass']:
- continue
- expectation = data_types.Expectation(e.test, e.tags, e.raw_results,
- e.reason)
- assert expectation not in expectations_for_file
- expectations_for_file[expectation] = data_types.BuilderStepMap()
- logging.info('Creating test expectation map')
- assert expectation_files or tests
- assert not (expectation_files and tests)
- expectation_map = data_types.TestExpectationMap()
- if expectation_files:
- if not isinstance(expectation_files, list):
- expectation_files = [expectation_files]
- for ef in expectation_files:
- # Normalize to '/' as the path separator.
- expectation_file_name = os.path.normpath(ef).replace(os.path.sep, '/')
- content = self._GetNonRecentExpectationContent(expectation_file_name,
- grace_period)
- AddContentToMap(content, expectation_map, expectation_file_name)
- else:
- expectation_file_name = ''
- content = '# results: [ RetryOnFailure ]\n'
- for t in tests:
- content += '%s [ RetryOnFailure ]\n' % t
- AddContentToMap(content, expectation_map, expectation_file_name)
- return expectation_map
- def _GetNonRecentExpectationContent(self, expectation_file_path: str,
- num_days: int) -> str:
- """Gets content from |expectation_file_path| older than |num_days| days.
- Args:
- expectation_file_path: A string containing a filepath pointing to an
- expectation file.
- num_days: An int containing how old an expectation in the given
- expectation file must be to be included.
- Returns:
- The contents of the expectation file located at |expectation_file_path|
- as a string with any recent expectations removed.
- """
- num_days = datetime.timedelta(days=num_days)
- content = ''
- # `git blame` output is normally in the format:
- # revision optional_filename (author date time timezone lineno) line_content
- # The --porcelain option is meant to be more machine readable, but is much
- # more difficult to parse for what we need to do here. In order to
- # guarantee that the filename won't be included in the output (by default,
- # it will be shown if there is content from a renamed file), pass -c to
- # use the same format as `git annotate`, which is:
- # revision (author date time timezone lineno)line_content
- # (Note the lack of space between the ) and the content).
- cmd = ['git', 'blame', '-c', expectation_file_path]
- with open(os.devnull, 'w') as devnull:
- blame_output = subprocess.check_output(cmd,
- stderr=devnull).decode('utf-8')
- for line in blame_output.splitlines(True):
- match = GIT_BLAME_REGEX.match(line)
- assert match
- date = match.groupdict()['date']
- line_content = match.groupdict()['content']
- if EXPECTATION_LINE_REGEX.match(line):
- if six.PY2:
- date_parts = date.split('-')
- date = datetime.date(year=int(date_parts[0]),
- month=int(date_parts[1]),
- day=int(date_parts[2]))
- else:
- date = datetime.date.fromisoformat(date)
- date_diff = datetime.date.today() - date
- if date_diff > num_days:
- content += line_content
- else:
- logging.debug('Omitting expectation %s because it is too new',
- line_content.rstrip())
- else:
- content += line_content
- return content
- def RemoveExpectationsFromFile(self,
- expectations: List[data_types.Expectation],
- expectation_file: str,
- removal_type: str) -> Set[str]:
- """Removes lines corresponding to |expectations| from |expectation_file|.
- Ignores any lines that match but are within a disable block or have an
- inline disable comment.
- Args:
- expectations: A list of data_types.Expectations to remove.
- expectation_file: A filepath pointing to an expectation file to remove
- lines from.
- removal_type: A RemovalType enum corresponding to the type of expectations
- being removed.
- Returns:
- A set of strings containing URLs of bugs associated with the removed
- expectations.
- """
- with open(expectation_file) as f:
- input_contents = f.read()
- output_contents = ''
- in_disable_block = False
- disable_block_reason = ''
- disable_block_suffix = ''
- removed_urls = set()
- for line in input_contents.splitlines(True):
- # Auto-add any comments or empty lines
- stripped_line = line.strip()
- if _IsCommentOrBlankLine(stripped_line):
- output_contents += line
- # Only allow one enable/disable per line.
- assert len([c for c in ALL_FINDER_COMMENTS if c in line]) <= 1
- # Handle disable/enable block comments.
- if _LineContainsDisableComment(line):
- if in_disable_block:
- raise RuntimeError(
- 'Invalid expectation file %s - contains a disable comment "%s" '
- 'that is in another disable block.' %
- (expectation_file, stripped_line))
- in_disable_block = True
- disable_block_reason = _GetDisableReasonFromComment(line)
- disable_block_suffix = _GetFinderCommentSuffix(line)
- if _LineContainsEnableComment(line):
- if not in_disable_block:
- raise RuntimeError(
- 'Invalid expectation file %s - contains an enable comment "%s" '
- 'that is outside of a disable block.' %
- (expectation_file, stripped_line))
- in_disable_block = False
- continue
- current_expectation = self._CreateExpectationFromExpectationFileLine(
- line, expectation_file)
- # Add any lines containing expectations that don't match any of the given
- # expectations to remove.
- if any(e for e in expectations if e == current_expectation):
- # Skip any expectations that match if we're in a disable block or there
- # is an inline disable comment.
- if in_disable_block and _DisableSuffixIsRelevant(
- disable_block_suffix, removal_type):
- output_contents += line
- logging.info(
- 'Would have removed expectation %s, but inside a disable block '
- 'with reason %s', stripped_line, disable_block_reason)
- elif _LineContainsRelevantDisableComment(line, removal_type):
- output_contents += line
- logging.info(
- 'Would have removed expectation %s, but it has an inline disable '
- 'comment with reason %s',
- stripped_line.split('#')[0], _GetDisableReasonFromComment(line))
- else:
- bug = current_expectation.bug
- if bug:
- # It's possible to have multiple whitespace-separated bugs per
- # expectation, so treat each one separately.
- removed_urls |= set(bug.split())
- else:
- output_contents += line
- with open(expectation_file, 'w') as f:
- f.write(output_contents)
- return removed_urls
- def _CreateExpectationFromExpectationFileLine(self, line: str,
- expectation_file: str
- ) -> data_types.Expectation:
- """Creates a data_types.Expectation from |line|.
- Args:
- line: A string containing a single line from an expectation file.
- expectation_file: A filepath pointing to an expectation file |line| came
- from.
- Returns:
- A data_types.Expectation containing the same information as |line|.
- """
- header = self._GetExpectationFileTagHeader(expectation_file)
- single_line_content = header + line
- list_parser = expectations_parser.TaggedTestListParser(single_line_content)
- assert len(list_parser.expectations) == 1
- typ_expectation = list_parser.expectations[0]
- return data_types.Expectation(typ_expectation.test, typ_expectation.tags,
- typ_expectation.raw_results,
- typ_expectation.reason)
- def _GetExpectationFileTagHeader(self, expectation_file: str) -> str:
- """Gets the tag header used for expectation files.
- Args:
- expectation_file: A filepath pointing to an expectation file to get the
- tag header from.
- Returns:
- A string containing an expectation file header, i.e. the comment block at
- the top of the file defining possible tags and expected results.
- """
- raise NotImplementedError()
- def ModifySemiStaleExpectations(
- self, stale_expectation_map: data_types.TestExpectationMap) -> Set[str]:
- """Modifies lines from |stale_expectation_map| in |expectation_file|.
- Prompts the user for each modification and provides debug information since
- semi-stale expectations cannot be blindly removed like fully stale ones.
- Args:
- stale_expectation_map: A data_types.TestExpectationMap containing stale
- expectations.
- file_handle: An optional open file-like object to output to. If not
- specified, stdout will be used.
- Returns:
- A set of strings containing URLs of bugs associated with the modified
- (manually modified by the user or removed by the script) expectations.
- """
- expectations_to_remove = []
- expectations_to_modify = []
- modified_urls = set()
- for expectation_file, e, builder_map in (
- stale_expectation_map.IterBuilderStepMaps()):
- with open(expectation_file) as infile:
- file_contents = infile.read()
- line, line_number = self._GetExpectationLine(e, file_contents,
- expectation_file)
- expectation_str = None
- if not line:
- logging.error(
- 'Could not find line corresponding to semi-stale expectation for '
- '%s with tags %s and expected results %s', e.test, e.tags,
- e.expected_results)
- expectation_str = '[ %s ] %s [ %s ]' % (' '.join(
- e.tags), e.test, ' '.join(e.expected_results))
- else:
- expectation_str = '%s (approx. line %d)' % (line, line_number)
- str_dict = result_output.ConvertBuilderMapToPassOrderedStringDict(
- builder_map)
- print('\nSemi-stale expectation:\n%s' % expectation_str)
- result_output.RecursivePrintToFile(str_dict, 1, sys.stdout)
- response = _WaitForUserInputOnModification()
- if response == 'r':
- expectations_to_remove.append(e)
- elif response == 'm':
- expectations_to_modify.append(e)
- # It's possible that the user will introduce a typo while manually
- # modifying an expectation, which will cause a parser error. Catch that
- # now and give them chances to fix it so that they don't lose all of their
- # work due to an early exit.
- while True:
- try:
- with open(expectation_file) as infile:
- file_contents = infile.read()
- _ = expectations_parser.TaggedTestListParser(file_contents)
- break
- except expectations_parser.ParseError as error:
- logging.error('Got parser error: %s', error)
- logging.error(
- 'This probably means you introduced a typo, please fix it.')
- _WaitForAnyUserInput()
- modified_urls |= self.RemoveExpectationsFromFile(expectations_to_remove,
- expectation_file,
- RemovalType.STALE)
- for e in expectations_to_modify:
- modified_urls |= set(e.bug.split())
- return modified_urls
- def _GetExpectationLine(self, expectation: data_types.Expectation,
- file_contents: str, expectation_file: str
- ) -> Union[Tuple[None, None], Tuple[str, int]]:
- """Gets the line and line number of |expectation| in |file_contents|.
- Args:
- expectation: A data_types.Expectation.
- file_contents: A string containing the contents read from an expectation
- file.
- expectation_file: A string containing the path to the expectation file
- that |file_contents| came from.
- Returns:
- A tuple (line, line_number). |line| is a string containing the exact line
- in |file_contents| corresponding to |expectation|. |line_number| is an int
- corresponding to where |line| is in |file_contents|. |line_number| may be
- off if the file on disk has changed since |file_contents| was read. If a
- corresponding line cannot be found, both |line| and |line_number| are
- None.
- """
- # We have all the information necessary to recreate the expectation line and
- # line number can be pulled during the initial expectation parsing. However,
- # the information we have is not necessarily in the same order as the
- # text file (e.g. tag ordering), and line numbers can change pretty
- # dramatically between the initial parse and now due to stale expectations
- # being removed. So, parse this way in order to improve the user experience.
- file_lines = file_contents.splitlines()
- for line_number, line in enumerate(file_lines):
- if _IsCommentOrBlankLine(line.strip()):
- continue
- current_expectation = self._CreateExpectationFromExpectationFileLine(
- line, expectation_file)
- if expectation == current_expectation:
- return line, line_number + 1
- return None, None
- def FindOrphanedBugs(self, affected_urls: Iterable[str]) -> Set[str]:
- """Finds cases where expectations for bugs no longer exist.
- Args:
- affected_urls: An iterable of affected bug URLs, as returned by functions
- such as RemoveExpectationsFromFile.
- Returns:
- A set containing a subset of |affected_urls| who no longer have any
- associated expectations in any expectation files.
- """
- seen_bugs = set()
- expectation_files = self.GetExpectationFilepaths()
- for ef in expectation_files:
- with open(ef) as infile:
- contents = infile.read()
- for url in affected_urls:
- if url in seen_bugs:
- continue
- if url in contents:
- seen_bugs.add(url)
- return set(affected_urls) - seen_bugs
- def GetExpectationFilepaths(self) -> List[str]:
- """Gets all the filepaths to expectation files of interest.
- Returns:
- A list of strings, each element being a filepath pointing towards an
- expectation file.
- """
- raise NotImplementedError()
- def _WaitForAnyUserInput() -> None:
- """Waits for any user input.
- Split out for testing purposes.
- """
- _get_input('Press any key to continue')
- def _WaitForUserInputOnModification() -> str:
- """Waits for user input on how to modify a semi-stale expectation.
- Returns:
- One of the following string values:
- i - Expectation should be ignored and left alone.
- m - Expectation will be manually modified by the user.
- r - Expectation should be removed by the script.
- """
- valid_inputs = ['i', 'm', 'r']
- prompt = ('How should this expectation be handled? (i)gnore/(m)anually '
- 'modify/(r)emove: ')
- response = _get_input(prompt).lower()
- while response not in valid_inputs:
- print('Invalid input, valid inputs are %s' % (', '.join(valid_inputs)))
- response = _get_input(prompt).lower()
- return response
- def _LineContainsDisableComment(line: str) -> bool:
- return FINDER_DISABLE_COMMENT_BASE in line
- def _LineContainsEnableComment(line: str) -> bool:
- return FINDER_ENABLE_COMMENT_BASE in line
- def _GetFinderCommentSuffix(line: str) -> str:
- """Gets the suffix of the finder comment on the given line.
- Examples:
- 'foo # finder:disable' -> ''
- 'foo # finder:disable-stale some_reason' -> '-stale'
- """
- target_str = None
- if _LineContainsDisableComment(line):
- target_str = FINDER_DISABLE_COMMENT_BASE
- elif _LineContainsEnableComment(line):
- target_str = FINDER_ENABLE_COMMENT_BASE
- else:
- raise RuntimeError('Given line %s did not have a finder comment.' % line)
- line = line[line.find(target_str):]
- line = line.split()[0]
- suffix = line.replace(target_str, '')
- assert suffix in ALL_FINDER_SUFFIXES
- return suffix
- def _LineContainsRelevantDisableComment(line: str, removal_type: str) -> bool:
- """Returns whether the given line contains a relevant disable comment.
- Args:
- line: A string containing the line to check.
- removal_type: A RemovalType enum corresponding to the type of expectations
- being removed.
- Returns:
- A bool denoting whether |line| contains a relevant disable comment given
- |removal_type|.
- """
- if FINDER_DISABLE_COMMENT_GENERAL in line:
- return True
- if FINDER_DISABLE_COMMENT_BASE + removal_type in line:
- return True
- return False
- def _DisableSuffixIsRelevant(suffix: str, removal_type: str) -> bool:
- """Returns whether the given suffix is relevant given the removal type.
- Args:
- suffix: A string containing a disable comment suffix.
- removal_type: A RemovalType enum corresponding to the type of expectations
- being removed.
- Returns:
- True if suffix is relevant and its disable request should be honored.
- """
- if suffix == FINDER_COMMENT_SUFFIX_GENERAL:
- return True
- if suffix == removal_type:
- return True
- return False
- def _GetDisableReasonFromComment(line: str) -> str:
- suffix = _GetFinderCommentSuffix(line)
- return line.split(FINDER_DISABLE_COMMENT_BASE + suffix, 1)[1].strip()
- def _IsCommentOrBlankLine(line: str) -> bool:
- return (not line or line.startswith('#'))
- def _get_input(prompt: str) -> str:
- if sys.version_info[0] == 2:
- return raw_input(prompt)
- return input(prompt)
|