Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use failed_step method in grail

Best Python code snippet using grail_python

detect_first_failure_pipeline.py

Source:detect_first_failure_pipeline.py

1# Copyright 2014 The Chromium Authors. All rights reserved.2# Use of this source code is governed by a BSD-style license that can be3# found in the LICENSE file.4import base645import json6from common import constants7from common.waterfall import failure_type8from gae_libs.http.http_client_appengine import HttpClientAppengine9from gae_libs.pipeline_wrapper import BasePipeline10from gae_libs.pipeline_wrapper import pipeline11from model.wf_analysis import WfAnalysis12from model.wf_step import WfStep13from waterfall import build_util14from waterfall import buildbot15from waterfall import swarming_util16_MAX_BUILDS_TO_CHECK = 2017_NON_FAILURE_STATUS = ['SUCCESS', 'SKIPPED', 'UNKNOWN']18_PRE_TEST_PREFIX = 'PRE_'19def _RemoveAllPrefixes(test):20  """Remove prefixes from test names.21  Args:22    test (str): A test's name, eg: 'suite1.PRE_test1'.23  Returns:24    base_test (str): A base test name, eg: 'suite1.test1'.25  """26  test_name_start = max(test.find('.'), 0)27  if test_name_start == 0:28    return test29  test_suite = test[: test_name_start]30  test_name = test[test_name_start + 1 :]31  pre_position = test_name.find(_PRE_TEST_PREFIX)32  while pre_position == 0:33    test_name = test_name[len(_PRE_TEST_PREFIX):]34    pre_position = test_name.find(_PRE_TEST_PREFIX)35  base_test = '%s.%s' % (test_suite, test_name)36  return base_test37def _GetFailureType(build_info):38  if not build_info.failed_steps:39    return failure_type.UNKNOWN40  # TODO(robertocn): Consider also bailing out of tests with infra failures.41  if constants.COMPILE_STEP_NAME in build_info.failed_steps:42    if build_info.result == buildbot.EXCEPTION:43      return failure_type.INFRA44    return failure_type.COMPILE45  # TODO(http://crbug.com/602733): differentiate test steps from infra ones.46  return failure_type.TEST47class DetectFirstFailurePipeline(BasePipeline):48  """A pipeline to detect first failure of each step.49  TODO(stgao): do test-level detection for gtest.50  """51  def _ExtractBuildInfo(self, master_name, builder_name, build_number):52    """Returns a BuildInfo instance for the specified build."""53    build = build_util.DownloadBuildData(54        master_name, builder_name, build_number)55    if build is None:  # pragma: no cover56      raise pipeline.Retry('Too many download from %s' % master_name)57    if not build.data:  # pragma: no cover58      return None59    build_info = buildbot.ExtractBuildInfo(60        master_name, builder_name, build_number, build.data)61    if not build.completed:62      build.start_time = build_info.build_start_time63      build.completed = build_info.completed64      build.result = build_info.result65      build.put()66    analysis = WfAnalysis.Get(master_name, builder_name, build_number)67    if analysis and not analysis.build_start_time:68      analysis.build_start_time = build_info.build_start_time69      analysis.put()70    return build_info71  def _SaveBlamelistAndChromiumRevisionIntoDict(self, build_info, builds):72    """Saves blame list and chromium revision info for each build.73    Args:74      build_info (BuildInfo): a BuildInfo instance which contains blame list and75          chromium revision.76      builds (dict): to which the blame list and chromium revision is saved. It77          will be updated and looks like:78          {79            555 : {80              'chromium_revision': 'a_git_hash',81              'blame_list': ['git_hash1', 'git_hash2'],82            },83          }84    """85    builds[build_info.build_number] = {86        'chromium_revision': build_info.chromium_revision,87        'blame_list': build_info.blame_list88    }89  def _CreateADictOfFailedSteps(self, build_info):90    """ Returns a dict with build number for failed steps.91    Args:92      failed_steps (list): a list of failed steps.93    Returns:94      A dict like this:95      {96        'step_name': {97          'current_failure': 555,98          'first_failure': 553,99        },100      }101    """102    failed_steps = dict()103    for step_name in build_info.failed_steps:104      failed_steps[step_name] = {105          'current_failure': build_info.build_number,106          'first_failure': build_info.build_number,107      }108    return failed_steps109  def _CheckForFirstKnownFailure(self, master_name, builder_name, build_number,110                                 failed_steps, builds):111    """Checks for first known failures of the given failed steps.112    Args:113      master_name (str): master of the failed build.114      builder_name (str): builder of the failed build.115      build_number (int): builder number of the current failed build.116      failed_steps (dict): the failed steps of the current failed build. It will117          be updated with build numbers for 'first_failure' and 'last_pass' of118          each failed step.119      builds (dict): a dict to save blame list and chromium revision.120    """121    # Look back for first known failures.122    earliest_build_number = max(0, build_number - 1 - _MAX_BUILDS_TO_CHECK)123    for n in range(build_number - 1, earliest_build_number - 1, -1):124      # Extraction should stop when when we reach to the first build125      build_info = self._ExtractBuildInfo(master_name, builder_name, n)126      if not build_info:  # pragma: no cover127        # Failed to extract the build information, bail out.128        return129      self._SaveBlamelistAndChromiumRevisionIntoDict(build_info, builds)130      if build_info.result == buildbot.SUCCESS:131        for step_name in failed_steps:132          if 'last_pass' not in failed_steps[step_name]:133            failed_steps[step_name]['last_pass'] = build_info.build_number134        # All steps passed, so stop looking back.135        return136      else:137        # If a step is not run due to some bot exception, we are not sure138        # whether the step could pass or not. So we only check failed/passed139        # steps here.140        for step_name in build_info.failed_steps:141          if (step_name in failed_steps and142              not 'last_pass' in failed_steps[step_name]):143            failed_steps[step_name]['first_failure'] = build_info.build_number144        for step_name in failed_steps:145          if (step_name in build_info.passed_steps and146              'last_pass' not in failed_steps[step_name]):147            failed_steps[step_name]['last_pass'] = build_info.build_number148        if all('last_pass' in step_info for step_info in failed_steps.values()):149          # All failed steps passed in this build cycle.150          return151  def _ConcatenateTestLog(self, string1, string2):152    """Concatenates the base64 encoded log.153    Tests if one string is a substring of another,154        if yes, returns the longer string,155        otherwise, returns the concatenation.156    Args:157      string1: base64-encoded string.158      string2: base64-encoded string.159    Returns:160      base64-encoded string.161    """162    str1 = base64.b64decode(string1)163    str2 = base64.b64decode(string2)164    if str2 in str1:165      return string1166    elif str1 in str2:167      return string2168    else:169      return base64.b64encode(str1 + str2)170  def _InitiateTestLevelFirstFailureAndSaveLog(171      self, json_data, step, failed_step=None):172    """Parses the json data and saves all the reliable failures to the step."""173    failed_test_log = {}174    if failed_step:175      failed_step['tests'] = {}176    for iteration in json_data.get('per_iteration_data'):177      for test_name in iteration.keys():178        is_reliable_failure = True179        if (any(test['status'] in _NON_FAILURE_STATUS180            for test in iteration[test_name])):181          # Ignore the test if any of the attempts didn't fail.182          # If a test is skipped, that means it was not run at all.183          # Treats it as success since the status cannot be determined.184          is_reliable_failure = False185        if is_reliable_failure:186          if failed_step:187            # Adds the test to failed_step.188            failed_step['tests'][test_name] = {189                'current_failure': failed_step['current_failure'],190                'first_failure': failed_step['current_failure'],191                'base_test_name': _RemoveAllPrefixes(test_name)192            }193            if failed_step.get('last_pass'):194              failed_step['tests'][test_name]['last_pass'] = (195                  failed_step['last_pass'])196          # Stores the output to the step's log_data later.197          failed_test_log[test_name] = ''198          for test in iteration[test_name]:199            failed_test_log[test_name] = self._ConcatenateTestLog(200                failed_test_log[test_name], test.get(201                    'output_snippet_base64', ''))202    step.log_data = json.dumps(failed_test_log) if failed_test_log else 'flaky'203    step.put()204    if failed_step and not failed_step['tests']:  # All flaky.205      del failed_step['tests']206      return False207    return True208  def _StartTestLevelCheckForFirstFailure(209      self, master_name, builder_name, build_number, step_name, failed_step,210      http_client):211    """Downloads test results and initiates first failure info at test level."""212    list_isolated_data = failed_step['list_isolated_data']213    result_log = swarming_util.RetrieveShardedTestResultsFromIsolatedServer(214        list_isolated_data, http_client)215    if (not result_log or not result_log.get('per_iteration_data') or216        result_log['per_iteration_data'] == 'invalid'):  # pragma: no cover217      return False218    step = WfStep.Get(master_name, builder_name, build_number, step_name)219    return self._InitiateTestLevelFirstFailureAndSaveLog(220        result_log, step, failed_step)221  def _GetSameStepFromBuild(222      self, master_name, builder_name, build_number, step_name,223      http_client):224    """Downloads swarming test results for a step from previous build."""225    step = WfStep.Get(226        master_name, builder_name, build_number, step_name)227    if step and step.isolated and step.log_data:228      # Test level log has been saved for this step.229      return step230    # Sends request to swarming server for isolated data.231    step_isolated_data = swarming_util.GetIsolatedDataForStep(232        master_name, builder_name, build_number, step_name,233        http_client)234    if not step_isolated_data:  # pragma: no cover235      return None236    result_log = swarming_util.RetrieveShardedTestResultsFromIsolatedServer(237        step_isolated_data, http_client)238    if (not result_log or not result_log.get('per_iteration_data') or239        result_log['per_iteration_data'] == 'invalid'):  # pragma: no cover240      return None241    step = WfStep.Create(242        master_name, builder_name, build_number, step_name)243    step.isolated = True244    self._InitiateTestLevelFirstFailureAndSaveLog(result_log, step)245    return step246  def _UpdateFirstFailureInfoForStep(self, current_build_number, failed_step):247    """Updates first_failure etc. for the step after the check for tests."""248    earliest_test_first_failure = current_build_number249    earliest_test_last_pass = current_build_number - 1250    for failed_test in failed_step['tests'].itervalues():251      # Iterates through all failed tests to prepare data for step level update.252      if not failed_test.get('last_pass'):253        # The test failed throughout checking range,254        # and there is no last_pass info for step.255        # last_pass not found.256        earliest_test_last_pass = -1257      earliest_test_first_failure = min(258          failed_test['first_failure'], earliest_test_first_failure)259      if (failed_test.get('last_pass') and260          failed_test['last_pass'] < earliest_test_last_pass):261        earliest_test_last_pass = failed_test['last_pass']262    # Updates Step level first failure info and last_pass info.263    failed_step['first_failure'] = max(264        earliest_test_first_failure, failed_step['first_failure'])265    if ((not failed_step.get('last_pass') and earliest_test_last_pass >= 0) or266        (failed_step.get('last_pass') and267         earliest_test_last_pass > failed_step['last_pass'])):268      failed_step['last_pass'] = earliest_test_last_pass269  def _UpdateFirstFailureOnTestLevel(270      self, master_name, builder_name, current_build_number, step_name,271      failed_step, http_client):272    """Iterates backwards through builds to get first failure at test level."""273    farthest_first_failure = failed_step['first_failure']274    if failed_step.get('last_pass'):275      farthest_first_failure = failed_step['last_pass'] + 1276    unfinished_tests = failed_step['tests'].keys()277    for build_number in range(278        current_build_number - 1, max(farthest_first_failure - 1, 0), -1):279      # Checks back until farthest_first_failure or build 1, don't use build 0280      # since there might be some abnormalities in build 0.281      step = self._GetSameStepFromBuild(282          master_name, builder_name, build_number, step_name,283          http_client)284      if not step or not step.log_data:  # pragma: no cover285        raise pipeline.Retry(286            'Failed to get swarming test results for a previous build.')287      failed_test_log = (288          {} if step.log_data == 'flaky' else json.loads(step.log_data))289      test_checking_list = unfinished_tests[:]290      for test_name in test_checking_list:291        if failed_test_log.get(test_name):292          failed_step['tests'][test_name]['first_failure'] = build_number293        else:294          # Last pass for this test has been found.295          # TODO(chanli): Handle cases where the test is not run at all.296          failed_step['tests'][test_name]['last_pass'] = build_number297          unfinished_tests.remove(test_name)298      if not unfinished_tests:299        break300    self._UpdateFirstFailureInfoForStep(current_build_number, failed_step)301  def _UpdateFailureInfoBuilds(self, failed_steps, builds):302    """Deletes builds that are before the farthest last_pass."""303    build_numbers_in_builds = builds.keys()304    farthest_last_pass = -1305    for failed_step in failed_steps.itervalues():306      if not failed_step.get('last_pass'):307        return308      if (farthest_last_pass < 0 or309          farthest_last_pass > failed_step['last_pass']):310        farthest_last_pass = failed_step['last_pass']311    for build_number in build_numbers_in_builds:312      if int(build_number) < farthest_last_pass:313        del builds[build_number]314  def _CheckFirstKnownFailureForSwarmingTests(315      self, master_name, builder_name, build_number, failed_steps, builds):316    """Uses swarming test results to update first failure info at test level."""317    http_client = HttpClientAppengine()318    # Identifies swarming tests and saves isolated data to them.319    result = swarming_util.GetIsolatedDataForFailedBuild(320        master_name, builder_name, build_number, failed_steps,321        http_client)322    if not result:323      return324    for step_name, failed_step in failed_steps.iteritems():325      if not failed_step.get('list_isolated_data'):  # Non-swarming step.326        continue  # pragma: no cover.327      # Checks tests in one step and updates failed_step info if swarming.328      result = self._StartTestLevelCheckForFirstFailure(329          master_name, builder_name, build_number, step_name,330          failed_step, http_client)331      if result:  # pragma: no cover332        # Iterates backwards to get a more precise failed_steps info.333        self._UpdateFirstFailureOnTestLevel(334            master_name, builder_name, build_number, step_name,335            failed_step, http_client)336    self._UpdateFailureInfoBuilds(failed_steps, builds)337  # Arguments number differs from overridden method - pylint: disable=W0221338  def run(self, master_name, builder_name, build_number):339    """340    Args:341      master_name (str): the master name of a build.342      builder_name (str): the builder name of a build.343      build_number (int): the build number of a build.344    Returns:345      A dict in the following form:346      {347        "master_name": "chromium.gpu",348        "builder_name": "GPU Linux Builder"349        "build_number": 25410,350        "failed": true,351        "failed_steps": {352          "compile": {353            "last_pass": 25408,354            "current_failure": 25410,355            "first_failure": 25409356          }357        },358        "builds": {359          "25408": {360            "chromium_revision": "474ab324d17d2cd198d3fb067cabc10a775a8df7"361            "blame_list": [362              "474ab324d17d2cd198d3fb067cabc10a775a8df7"363            ],364          },365          "25409": {366            "chromium_revision": "33c6f11de20c5b229e102c51237d96b2d2f1be04"367            "blame_list": [368              "9d5ebc5eb14fc4b3823f6cfd341da023f71f49dd",369              ...370            ],371          },372          "25410": {373            "chromium_revision": "4bffcd598dd89e0016208ce9312a1f477ff105d1"374            "blame_list": [375              "b98e0b320d39a323c81cc0542e6250349183a4df",376              ...377            ],378          }379        }380      }381    """382    build_info = self._ExtractBuildInfo(master_name, builder_name, build_number)383    if not build_info:  # pragma: no cover384      raise pipeline.Retry('Failed to extract build info.')385    build_failure_type = _GetFailureType(build_info)386    failure_info = {387        'failed': True,388        'master_name': master_name,389        'builder_name': builder_name,390        'build_number': build_number,391        'chromium_revision': build_info.chromium_revision,392        'builds': {},393        'failed_steps': {},394        'failure_type': build_failure_type,395        'parent_mastername': build_info.parent_mastername,396        'parent_buildername': build_info.parent_buildername,397    }398    if build_info.result == buildbot.SUCCESS or not build_info.failed_steps:399      failure_info['failed'] = False400      return failure_info401    builds = dict()402    self._SaveBlamelistAndChromiumRevisionIntoDict(build_info, builds)403    failed_steps = self._CreateADictOfFailedSteps(build_info)404    # Checks first failed builds for each failed step.405    self._CheckForFirstKnownFailure(406        master_name, builder_name, build_number, failed_steps, builds)407    if build_failure_type == failure_type.TEST:408      # Checks first failed builds for each failed test.409      self._CheckFirstKnownFailureForSwarmingTests(410          master_name, builder_name, build_number, failed_steps, builds)411    failure_info['builds'] = builds412    failure_info['failed_steps'] = failed_steps413    analysis = WfAnalysis.Get(master_name, builder_name, build_number)414    analysis.not_passed_steps = build_info.not_passed_steps415    analysis.build_failure_type = build_failure_type416    analysis.put()...

ci_test_failure.py

Source:ci_test_failure.py

...146    master_name(str): Name of the master of the failed build.147    builder_name(str): Name of the builder of the failed build.148    current_build_number(int): Number of the failed build.149    step_name(str): Name of one failed step the function is analyzing.150    failed_step(TestFailedStep): Known information about the failed step.151    build_numbers(list): A list of build numbers in backward order.152    http_client(FinditHttpClient): Http_client to send request.153  Returns:154    Updated failed_step with test level failure info.155  """156  farthest_first_failure = failed_step.first_failure157  if failed_step.last_pass:158    farthest_first_failure = failed_step.last_pass + 1159  unfinished_tests = failed_step.tests.keys()160  for build_number in build_numbers:161    if (build_number >= current_build_number or162        build_number < farthest_first_failure):163      continue164    # Checks back until farthest_first_failure or build 1, don't use build 0...

sample.py

Source:sample.py

...92    @property93    def trimmed_rev_fp(self):94        return self._trimmed_rev95    @property96    def failed_step(self):97        """str: step in the pipeline in which the sample failed, otherwise None."""98        return self._failed_step99    @failed_step.setter100    def failed_step(self, step):101        self._failed_step = step102    @staticmethod103    def load_samples(map_fp, in_dir, sample_type):104        """creates a list of Sample objects"""105        samples = list()106        reader = csv.DictReader(open(map_fp), delimiter='\t')107        for row in reader:108            if row['#SampleID'] == '':109                continue110            if sample_type == 'PE':111                smpl = Sample(format=sample_type, id=row['#SampleID'],112                              fwd_fp=os.path.join(in_dir, row['ForwardFastqFile']),113                              rev_fp=os.path.join(in_dir, row['ReverseFastqFile']))114            else:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.