Waterfall components of regression range finder.

appengine/findit/waterfall/process_flake_swarming_task_result_pipeline.py

Index: appengine/findit/waterfall/process_flake_swarming_task_result_pipeline.py
diff --git a/appengine/findit/waterfall/process_flake_swarming_task_result_pipeline.py b/appengine/findit/waterfall/process_flake_swarming_task_result_pipeline.py
new file mode 100644
index 0000000000000000000000000000000000000000..7325add9ee3b7d2c4b0715ef956a30dbd994f4ff
--- /dev/null
+++ b/appengine/findit/waterfall/process_flake_swarming_task_result_pipeline.py
@@ -0,0 +1,91 @@
+# Copyright 2016 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 datetime
+import logging
+import time
+
+from google.appengine.ext import ndb
+
+from common.http_client_appengine import HttpClientAppengine as HttpClient
+from common.pipeline_wrapper import BasePipeline
+
+from model import analysis_status
+from model.flake.master_flake_analysis import MasterFlakeAnalysis
+from model.flake.flake_swarming_task import FlakeSwarmingTask
+from waterfall.process_base_swarming_task_result_pipeline import (
+    ProcessBaseSwarmingTaskResultPipeline as PBSTRP)

[lijeffrey, 2016/07/27 22:38:23]

I know this is a long name but it's probably better just to spell it out for
consistency with the other parts of the code and for readability

[caiw, 2016/07/27 23:51:33]

Done.

+
+
+class ProcessFlakeSwarmingTaskResultPipeline(PBSTRP):
+  """A pipeline for monitoring swarming task and processing task result.
+
+  This pipeline waits for result for a swarming task and processes the result to
+  generate a dict for statuses for each test run.
+  """
+  # Arguments number differs from overridden method - pylint: disable=W0221
+  def _CheckTestsRunStatuses(self, output_json, master_name,
+                             builder_name, build_number, step_name,
+                             master_build_number, test_name):
+    """Checks result status for each test run and saves the numbers accordingly.
+
+    Args:
+      output_json (dict): A dict of all test results in the swarming task.
+      master_name (string): Name of master of swarming rerun.
+      builder_name (dict): Name of builder of swarming rerun.
+      build_number (int): Build Number of swarming rerun.
+      step_name (dict): Name of step of swarming rerun.
+      master_build_number (int): Build number of corresponding mfa
+      test_name (string): Name of test of swarming rerun
+
+    Returns:
+      tests_statuses (dict): A dict of different statuses for each test.
+
+    Currently for each test, we are saving number of total runs,
+    number of succeeded runs and number of failed runs.
+    """
+
+    master_flake_analysis = MasterFlakeAnalysis.Get(master_name, builder_name,
+                                  master_build_number,
+                                  step_name, test_name)
+    flake_swarming_task = FlakeSwarmingTask.Get(
+        master_name, builder_name, build_number, step_name, test_name)
+
+    tests_statuses = defaultdict(lambda: defaultdict(int))
+
+    if not output_json:

[lijeffrey, 2016/07/27 22:38:23]

it looks like you may be able to bail out even sooner before. This way you save
2 datastore reads.

test_statuses = defaultdict(...)

if not output_json:
  return test_statuses

master_flake_analysis = ...
flake_swarming_task = ...

[caiw, 2016/07/27 23:51:33]

Done.

+      return tests_statuses
+
+    successes = 0
+    tries = 0
+    for iteration in output_json.get('per_iteration_data'):
+      for test_name, tests in iteration.iteritems():
+        tries += 1
+        tests_statuses[test_name]['total_run'] += len(tests)
+        for test in tests:
+          if test['status'] == 'SUCCESS':
+            successes += 1
+          tests_statuses[test_name][test['status']] += 1
+    master_flake_analysis.build_numbers.append(build_number)

[lijeffrey, 2016/07/27 22:38:23]

nit: empty line before master_flake_analysis.build_numbers.append(...) for
readability

[caiw, 2016/07/27 23:51:33]

Done.

+    master_flake_analysis.success_rates.append(successes * 1.0/tries)
+    flake_swarming_task.tries = tries
+    flake_swarming_task.successes = successes
+    flake_swarming_task.put()
+    master_flake_analysis.put()
+    return tests_statuses
+
+  def _GetArgs(self, master_name, builder_name, build_number,
+               step_name, *args):
+    master_build_number = args[0]
+    test_name = args[1]
+    return (master_name, builder_name, build_number, step_name,
+            master_build_number, test_name)
+
+  # Unused Argument - pylint: disable=W0612,W0613
+  def _GetSwarmingTask(self, master_name, builder_name, build_number,
+                       step_name, master_build_number, test_name):
+    # Get the appropriate kind of Swarming Task (Flake)

[lijeffrey, 2016/07/27 22:38:23]

nit: comment ends with .

You should be able to run gpylint <path to python file> that will point these
out for you

[caiw, 2016/07/27 23:51:33]

Done.

+    return FlakeSwarmingTask.Get(master_name, builder_name,
+                                 build_number, step_name, test_name)