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

appengine/findit/waterfall/identify_try_job_culprit_pipeline.py

Index: appengine/findit/waterfall/identify_try_job_culprit_pipeline.py
diff --git a/appengine/findit/waterfall/identify_try_job_culprit_pipeline.py b/appengine/findit/waterfall/identify_try_job_culprit_pipeline.py
index e59c99796299dd9c27f06438fab9edfa1bc48c54..f9d3038cdeab5b0c0f313b4e409ad9e8909ae0b6 100644
--- a/appengine/findit/waterfall/identify_try_job_culprit_pipeline.py
+++ b/appengine/findit/waterfall/identify_try_job_culprit_pipeline.py
@@ -11,9 +11,11 @@ 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_suspected_cl import WfSuspectedCL
 from model.wf_try_job import WfTryJob
 from model.wf_try_job_data import WfTryJobData
 from waterfall.send_notification_for_culprit_pipeline import (
@@ -51,51 +53,60 @@ def _GetResultAnalysisStatus(analysis, result):
   return old_result_status
 
 
-def _GetSuspectedCLs(analysis, result):
-  """Returns a list of suspected CLs.
+def _GetTryJobSuspectedCLs(result):
+  """Returns all suspected CLs found by the try job.
 
   Args:
-    analysis: The WfAnalysis entity corresponding to this try job.
     result: A result dict containing the culprit from the results of
       this try job.
 
   Returns:
-    A combined list of suspected CLs from those already in analysis and those
-    found by this try job.
+    A list of suspected CLs 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]
+  suspected_cls = []
+  if not result or not result.get('culprit'):
+    return 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')
-    if revision not in suspected_cl_revisions:
-      suspected_cl_revisions.append(revision)
-      suspected_cls.append(compile_cl_info)
+    suspected_cls.append(compile_cl_info)
     return suspected_cls
 
   # Suspected CLs are from test failures.
+  suspected_cl_revisions = []

[lijeffrey, 2016/08/10 08:12:39]

Move the test suspected_cls logic to a separate function for testability

[chanli, 2016/08/11 23:39:36]

Done.

   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')
-      }
+    for test_cl_info in results['tests'].values():

[lijeffrey, 2016/08/10 08:12:39]

nit: if using itervalues() above, why not use it here too?

[chanli, 2016/08/11 23:39:36]

Done.

+      revision = test_cl_info['revision']
       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)
+        suspected_cls.append(test_cl_info)
+
+  return suspected_cls
+
+
+def _GetSuspectedCLs(analysis, try_job_suspected_cls):
+  """Returns a list of suspected CLs.
+
+  Args:
+    analysis: The WfAnalysis entity corresponding to this try job.
+    is_compile_try_job: True if the try job is for compile failure,
+      otherwise False.
+    try_job_suspected_cls: 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]
+
+  for try_job_suspected_cl in try_job_suspected_cls:
+    if try_job_suspected_cl['revision'] not in suspected_cl_revisions:
+      suspected_cl_revisions.append(try_job_suspected_cl['revision'])
+      suspected_cls.append(try_job_suspected_cl)
 
   return suspected_cls
 
@@ -168,10 +179,6 @@ def _GetCulpritsForTestsFromResultsDict(blame_list, test_results):
         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']:
@@ -248,11 +255,6 @@ class IdentifyTryJobCulpritPipeline(BasePipeline):
   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])
@@ -261,12 +263,9 @@ class IdentifyTryJobCulpritPipeline(BasePipeline):
     """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
@@ -318,26 +317,57 @@ class IdentifyTryJobCulpritPipeline(BasePipeline):
       try_job_result.put()
 
     @ndb.transactional
-    def UpdateWfAnalysisWithTryJobResult():
+    def UpdateWfAnalysisWithTryJobResult(try_job_suspected_cls):
       if not culprits:
         return
-
       # Update analysis result and suspected CLs with results of this try job if
       # culprits were found.
       analysis = WfAnalysis.Get(master_name, builder_name, build_number)
       updated_result_status = _GetResultAnalysisStatus(analysis, result)
-      updated_suspected_cls = _GetSuspectedCLs(analysis, result)
-
+      updated_suspected_cls = _GetSuspectedCLs(analysis, try_job_suspected_cls)
       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()
 
+    @ndb.transactional
+    def CreateOrUpdateSuspectedCL(repo_name, revision, commit_position):

[lijeffrey, 2016/08/10 08:12:39]

this looks almost the same as _SaveSuspectedCL in identify_culprit_pipeline, is
there any way they can be refactored into a common module?

[lijeffrey, 2016/08/10 08:12:39]

nit: Why not just name this UpdateSuspectedCL?

[chanli, 2016/08/11 23:39:36]

Done.

[chanli, 2016/08/11 23:39:36]

Done.

+        suspected_cl = WfSuspectedCL.Get(repo_name, revision)

[lijeffrey, 2016/08/10 08:12:39]

suspected_cl = WfSuspectedCl.Get(...) or WfSuspectedCL.Create(...)

[chanli, 2016/08/11 23:39:36]

Done.

+        if not suspected_cl:
+          suspected_cl = WfSuspectedCL.Create(
+              repo_name, revision, commit_position)
+
+        if suspected_cl.approach == analysis_approach_type.HEURISTIC:
+          suspected_cl.approach = analysis_approach_type.BOTH
+        elif suspected_cl.approach is None:  #pragma: no cover
+          suspected_cl.approach = analysis_approach_type.TRY_JOB
+
+        suspected_cl.failure_type = suspected_cl.failure_type or try_job_type
+
+        build_info = [master_name, builder_name, build_number]
+        if build_info not in suspected_cl.builds:
+          suspected_cl.builds.append(build_info)
+
+        suspected_cl.put()
+
+    def CreateOrUpdateSuspectedCLs(try_job_suspected_cls):
+      if not culprits:
+        return
+
+      # Creates or updates each suspected_cl.
+      for culprit in try_job_suspected_cls:
+        CreateOrUpdateSuspectedCL(
+            culprit['repo_name'], culprit['revision'],
+            culprit.get('commit_position'))
+
+    try_job_suspected_cls = _GetTryJobSuspectedCLs(result)
     # Store try-job results.
     UpdateTryJobResult()
     # Add try-job results to WfAnalysis.
-    UpdateWfAnalysisWithTryJobResult()
+    UpdateWfAnalysisWithTryJobResult(try_job_suspected_cls)
+    # Updates suspected_cl
+    CreateOrUpdateSuspectedCLs(try_job_suspected_cls)
 
     _NotifyCulprits(master_name, builder_name, build_number, culprits)
     return result.get('culprit') if result else None