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..0574815d3601266099696217fd1c1932dcd5a572
--- /dev/null
+++ b/appengine/findit/waterfall/process_flake_swarming_task_result_pipeline.py
@@ -0,0 +1,77 @@
+# 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 model.flake.flake_swarming_task_result import FlakeSwarmingTaskResult
+from waterfall.process_base_swarming_task_result_pipeline import (
+    ProcessBaseSwarmingTaskResultPipeline as PBSTRP)
+
+
+
+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):
+    """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.

[stgao, 2016/07/09 00:04:34]

style nit: indent.

[caiw, 2016/07/14 00:59:45]

Done.

+
+    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.
+    """
+    mfa = MasterFlakeAnalysis.Get(master_name, builder_name, step_name)
+    mfa.build_numbers.append(build_number)
+
+    tests_statuses = defaultdict(lambda: defaultdict(int))
+    successes = 0
+    tries = 0
+
+    if output_json:
+      for iteration in output_json.get('per_iteration_data'):
+        test_name = ""
+        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
+          mfa.success_rates.append(successes * 1.0/tries)

[stgao, 2016/07/09 00:04:34]

This might have to be out of both for loops, because each test will show up in
all iterations.

You may want to check one of a Swarming run result (i.e.
https://chromium-swarm.appspot.com/user/task/2fdbf62cc33ed210) for the data
format.

[caiw, 2016/07/14 00:59:45]

Done.

+      result = FlakeSwarmingTaskResult.Create(
+          master_name, builder_name, step_name,
+          build_number, test_name)
+      result.tries = tries
+      result.successes = successes
+      result.put()
+      mfa.put()
+    return tests_statuses
+
+  @ndb.transactional
+  def _GetSwarmingTask(self, master_name, builder_name,
+                       step_name, build_number):
+    # Get the appropriate kind of Swarming Task (Flake)
+    swarming_task = FlakeSwarmingTask.Get(
+        master_name, builder_name, step_name, build_number)
+    return swarming_task

[stgao, 2016/07/09 00:04:33]

nit: merge into one line.

Does the read have to be in a transaction? What's the consideration behind that?