[Findit] Pipeline change to save suspected cls to data store.

appengine/findit/waterfall/identify_try_job_culprit_pipeline.py

 # Copyright 2015 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.
 
 from collections import defaultdict
+import copy
 import logging
 
 from google.appengine.ext import ndb
 
 from common.git_repository import GitRepository
 from common.http_client_appengine import HttpClientAppengine as HttpClient
 from common.pipeline_wrapper import BasePipeline
 from common.waterfall import failure_type
+from model import analysis_approach_type
 from model import analysis_status
 from model import result_status
 from model.wf_analysis import WfAnalysis
 from model.wf_try_job import WfTryJob
 from model.wf_try_job_data import WfTryJobData
 from waterfall.send_notification_for_culprit_pipeline import (
     SendNotificationForCulpritPipeline)
+from waterfall import suspected_cl_util
 
 
 GIT_REPO = GitRepository(
     'https://chromium.googlesource.com/chromium/src.git', HttpClient())
 
 
 def _GetResultAnalysisStatus(analysis, result):
   """Returns the analysis status based on existing status and try job result.
 
   Args:
@@ skipping 13 matching lines @@
   if (try_job_found_culprit and
       (old_result_status is None or
        old_result_status == result_status.NOT_FOUND_UNTRIAGED or
        old_result_status == result_status.NOT_FOUND_INCORRECT or
        old_result_status == result_status.NOT_FOUND_CORRECT)):
     return result_status.FOUND_UNTRIAGED
 
   return old_result_status
 
 
-def _GetSuspectedCLs(analysis, result):
+def _GetSuspectedCLs(analysis, try_job_type, result, culprits):
   """Returns a list of suspected CLs.
 
   Args:
     analysis: The WfAnalysis entity corresponding to this try job.
-    result: A result dict containing the culprit from the results of
-      this try job.
+    try_job_type: Try job type, COMPILE or TEST, the same with failure type.
+    result: A result dict containing the result of this try job.
+    culprits: A list of suspected CLs found by the try job.
 
   Returns:
     A combined list of suspected CLs from those already in analysis and those
     found by this try job.
   """
   suspected_cls = analysis.suspected_cls[:] if analysis.suspected_cls else []
   suspected_cl_revisions = [cl['revision'] for cl in suspected_cls]
-  culprit = result.get('culprit')
-  compile_cl_info = culprit.get('compile')
 
-  if compile_cl_info:
-    # Suspected CL is from compile failure.
-    revision = compile_cl_info.get('revision')
+  for revision, try_job_suspected_cl in culprits.iteritems():
+    suspected_cl_copy = copy.deepcopy(try_job_suspected_cl)
     if revision not in suspected_cl_revisions:
       suspected_cl_revisions.append(revision)
-      suspected_cls.append(compile_cl_info)
-    return suspected_cls
-
-  # Suspected CLs are from test failures.
-  for results in culprit.itervalues():
-    if results.get('revision'):
-      # Non swarming test failures, only have step level failure info.
-      revision = results.get('revision')
-      cl_info = {
-          'url': results.get('url'),
-          'repo_name': results.get('repo_name'),
-          'revision': results.get('revision'),
-          'commit_position': results.get('commit_position')
-      }
-      if revision not in suspected_cl_revisions:
-        suspected_cl_revisions.append(revision)
-        suspected_cls.append(cl_info)
-    else:
-      for test_cl_info in results['tests'].values():
-        revision = test_cl_info.get('revision')
-        if revision not in suspected_cl_revisions:
-          suspected_cl_revisions.append(revision)
-          suspected_cls.append(test_cl_info)
+      if try_job_type == failure_type.COMPILE:
+        failures = {'compile': []}
+      else:
+        failures = _GetTestFailureCausedByCL(
+          result.get('report', {}).get('result', {}).get(revision))
+      suspected_cl_copy['failures'] = failures
+      suspected_cl_copy['top_score'] = None
+      suspected_cls.append(suspected_cl_copy)
 
   return suspected_cls
 
 
 def _GetFailedRevisionFromResultsDict(results_dict):
   """Finds the failed revision from the given dict of revisions.
 
   Args:
     results_dict: (dict) A dict that maps revisions to their results. For
     example:
@@ skipping 52 matching lines @@
       if (not test_result['valid'] or
           test_result['status'] != 'failed'):  # pragma: no cover
         continue
 
       failed_revisions.add(revision)
 
       if step not in culprit_map:
         culprit_map[step] = {
             'tests': {}
         }
-      if (not test_result['failures'] and
-          not culprit_map[step].get('revision')):
-        # Non swarming test failures, only have step level failure info.
-        culprit_map[step]['revision'] = revision
       for failed_test in test_result['failures']:
         # Swarming tests, gets first failed revision for each test.
         if failed_test not in culprit_map[step]['tests']:
           culprit_map[step]['tests'][failed_test] = {
               'revision': revision
           }
 
   return culprit_map, list(failed_revisions)
 
 
@@ skipping 49 matching lines @@
           send_notification_right_now)
   elif compile_suspected_cl:
     # A special case where try job didn't find any suspected cls, but
     # heuristic found a suspected_cl.
     _StartSendNotificationPipeline(
         master_name, builder_name, build_number,
         compile_suspected_cl['repo_name'], compile_suspected_cl['revision'],
         send_notification_right_now=True)
 
 
+def _GetTestFailureCausedByCL(result):
+  if not result:
+    return None
+
+  failures = {}
+  for step_name, step_result in result.iteritems():
+    if step_result['status'] == 'failed':
+      failures[step_name] = step_result['failures']
+
+  return failures
+
+
 class IdentifyTryJobCulpritPipeline(BasePipeline):
   """A pipeline to identify culprit CL info based on try job compile results."""
 
   def _GetCulpritInfo(self, failed_revisions):
     """Gets commit_positions and review urls for revisions."""
     culprits = {}
     # TODO(lijeffrey): remove hard-coded 'chromium' when DEPS file parsing is
     # supported.
     for failed_revision in failed_revisions:
       culprits[failed_revision] = {
           'revision': failed_revision,
           'repo_name': 'chromium'
       }
       change_log = GIT_REPO.GetChangeLog(failed_revision)
       if change_log:
         culprits[failed_revision]['commit_position'] = (
             change_log.commit_position)
         culprits[failed_revision]['url'] = (
             change_log.code_review_url or change_log.commit_url)
 
     return culprits
 
   def _FindCulpritForEachTestFailure(self, blame_list, result):
     # For test failures, we need to traverse the result dict in chronological
     # order to identify the culprits for each failed step or test.
     # The earliest revision that a test failed is the culprit.
     culprit_map = defaultdict(dict)
     failed_revisions = set()
 
-    # Recipe should return culprits with the farmat as:
+    # Recipe should return culprits with the format as:
     # 'culprits': {
     #     'step1': {
     #         'test1': 'rev1',
     #         'test2': 'rev2',
     #         ...
     #     },
     #     ...
     # }
     if result['report'].get('culprits'):
       for step_name, tests in result['report']['culprits'].iteritems():
         culprit_map[step_name]['tests'] = {}
         for test_name, revision in tests.iteritems():
           culprit_map[step_name]['tests'][test_name] = {
             'revision': revision
           }
           failed_revisions.add(revision)
       return culprit_map, list(failed_revisions)
 
     return _GetCulpritsForTestsFromResultsDict(
         blame_list, result['report'].get('result'))
 
   def _UpdateCulpritMapWithCulpritInfo(self, culprit_map, culprits):
     """Fills in commit_position and review url for each failed rev in map."""
     for step_culprit in culprit_map.values():
-      if step_culprit.get('revision'):
-        culprit = culprits[step_culprit['revision']]
-        step_culprit['commit_position'] = culprit['commit_position']
-        step_culprit['url'] = culprit['url']
-        step_culprit['repo_name'] = culprit['repo_name']
       for test_culprit in step_culprit.get('tests', {}).values():
         test_revision = test_culprit['revision']
         test_culprit.update(culprits[test_revision])
 
   def _GetCulpritDataForTest(self, culprit_map):
     """Gets culprit revision for each failure for try job metadata."""
     culprit_data = {}
     for step, step_culprit in culprit_map.iteritems():
-      if step_culprit['tests']:
-        culprit_data[step] = {}
-        for test, test_culprit in step_culprit['tests'].iteritems():
-          culprit_data[step][test] = test_culprit['revision']
-      else:
-        culprit_data[step] = step_culprit['revision']
+      culprit_data[step] = {}
+      for test, test_culprit in step_culprit['tests'].iteritems():
+        culprit_data[step][test] = test_culprit['revision']
     return culprit_data
 
   # Arguments number differs from overridden method - pylint: disable=W0221
   def run(
       self, master_name, builder_name, build_number, blame_list, try_job_type,
       try_job_id, result):
     """Identifies the information for failed revisions.
 
     Please refer to try_job_result_format.md for format check.
     """
@@ skipping 39 matching lines @@
 
     @ndb.transactional
     def UpdateWfAnalysisWithTryJobResult():
       if not culprits:
         return
 
       analysis = WfAnalysis.Get(master_name, builder_name, build_number)
       # Update analysis result and suspected CLs with results of this try job if
       # culprits were found.
       updated_result_status = _GetResultAnalysisStatus(analysis, result)
-      updated_suspected_cls = _GetSuspectedCLs(analysis, result)
-
+      updated_suspected_cls = _GetSuspectedCLs(
+          analysis, try_job_type, result, culprits)
       if (analysis.result_status != updated_result_status or
           analysis.suspected_cls != updated_suspected_cls):
         analysis.result_status = updated_result_status
         analysis.suspected_cls = updated_suspected_cls
         analysis.put()
 
+    def UpdateSuspectedCLs():
+      if not culprits:
+        return
+
+      # Creates or updates each suspected_cl.
+      for culprit in culprits.values():
+        revision = culprit['revision']
+        if try_job_type == failure_type.COMPILE:
+          failures = {'compile': []}
+        else:
+          failures = _GetTestFailureCausedByCL(
+              result.get('report', {}).get('result', {}).get(revision))
+
+        suspected_cl_util.UpdateSuspectedCL(
+            culprit['repo_name'], revision, culprit.get('commit_position'),
+            analysis_approach_type.TRY_JOB, master_name, builder_name,
+            build_number, try_job_type, failures, None)
+
     # Store try-job results.
     UpdateTryJobResult()
 
     # Saves cls found by heuristic approach for later use.
     # This part must be before UpdateWfAnalysisWithTryJobResult().
     analysis = WfAnalysis.Get(master_name, builder_name, build_number)
     heuristic_cls = _GetHeuristicSuspectedCLs(analysis)
     compile_suspected_cl = (
         _GetSuspectedCLFoundByHeuristicForCompile(analysis)
         if try_job_type == failure_type.COMPILE else None)
 
     # Add try-job results to WfAnalysis.
     UpdateWfAnalysisWithTryJobResult()
 
+    # TODO (chanli): Update suspected_cl for builds in the same group with
+    # current build.
+    # Updates suspected_cl.
+    UpdateSuspectedCLs()
+
     _NotifyCulprits(master_name, builder_name, build_number, culprits,
                     heuristic_cls, compile_suspected_cl)
     return result.get('culprit') if result else None