Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(1581)

Issues Search
    My Issues | Starred     Open | Closed | All

Unified Diff: appengine/findit/waterfall/test/trigger_swarming_tasks_pipeline_test.py

Issue 2139093002: [Findit] Trigger swarming tasks after detech_first_faliure_pipeline (Closed) Base URL: https://chromium.googlesource.com/infra/infra.git@master
Patch Set: . Created 4 years, 5 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « appengine/findit/waterfall/test/swarming_tasks_to_try_job_pipeline_test.py ('k') | appengine/findit/waterfall/test/try_job_util_test.py » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: appengine/findit/waterfall/test/trigger_swarming_tasks_pipeline_test.py
diff --git a/appengine/findit/waterfall/test/trigger_swarming_tasks_pipeline_test.py b/appengine/findit/waterfall/test/trigger_swarming_tasks_pipeline_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..ca3c043b9f6b0a08a2af9b8daf47e42fbad450d9
--- /dev/null
+++ b/appengine/findit/waterfall/test/trigger_swarming_tasks_pipeline_test.py
@@ -0,0 +1,278 @@
+# 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 common.pipeline_wrapper import BasePipeline
+from common.pipeline_wrapper import pipeline_handlers
+from common.waterfall import failure_type
+from model.wf_analysis import WfAnalysis
+from waterfall import trigger_swarming_task_pipeline
+from waterfall import trigger_swarming_tasks_pipeline
+from waterfall.test import wf_testcase
+from waterfall.trigger_swarming_tasks_pipeline import (
+ TriggerSwarmingTasksPipeline)
+from waterfall.trigger_swarming_task_pipeline import TriggerSwarmingTaskPipeline
+
+
+class TriggerSwarmingTasksPipelineTest(wf_testcase.WaterfallTestCase):
+ app_module = pipeline_handlers._APP
+
+ def test_GetStepsThatNeedToTriggerSwarmingTasksNoAnalysis(self):
+ result = (
+ trigger_swarming_tasks_pipeline._GetStepsThatNeedToTriggerSwarmingTasks(
+ 'm', 'b', 1, {}))
+ self.assertEqual(result, {})
+
+ def test_GetStepsThatNeedToTriggerSwarmingTasksNoFailureResultMap(self):
+ master_name = 'm'
+ builder_name = 'b'
+ build_number = 2
+ WfAnalysis.Create(master_name, builder_name, build_number).put()
+
+ failure_info = {
+ 'failed': True,
+ 'master_name': 'm',
+ 'builder_name': 'b',
+ 'build_number': 2,
+ 'chromium_revision': None,
+ 'builds': {
+ 2: {
+ 'blame_list': [],
+ 'chromium_revision': None
+ }
+ },
+ 'failed_steps': {
+ 'abc_test': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'tests': {
+ 'Unittest2.Subtest1': {
+ 'current_failure': 2,
+ 'first_failure': 2,
+ 'last_pass': 1,
+ 'base_test_name': 'Unittest2.Subtest1'
+ },
+ 'Unittest3.Subtest2': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'base_test_name': 'Unittest3.Subtest2'
+ }
+ }
+ },
+ 'a_test': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ }
+ },
+ 'failure_type': failure_type.TEST
+ }
+
+ expected_result = {
+ 'abc_test': ['Unittest2.Subtest1']
+ }
+ expected_failure_result_map = {
+ 'abc_test': {
+ 'Unittest2.Subtest1': '%s/%s/%d' % (
+ master_name, builder_name, build_number),
+ 'Unittest3.Subtest2': '%s/%s/%d' % (
+ master_name, builder_name, 1)
+ }
+ }
+
+ result = (
+ trigger_swarming_tasks_pipeline._GetStepsThatNeedToTriggerSwarmingTasks(
+ master_name, builder_name, build_number, failure_info))
+ analysis = WfAnalysis.Get(master_name, builder_name, build_number)
+
+ self.assertEqual(result, expected_result)
+ self.assertEqual(analysis.failure_result_map, expected_failure_result_map)
+
+ def test_GetStepsThatNeedToTriggerSwarmingTasks(self):
+ master_name = 'm'
+ builder_name = 'b'
+ build_number = 2
+ analysis = WfAnalysis.Create(master_name, builder_name, build_number)
+ analysis.failure_result_map = {
+ 'a_tests': {
+ 'Unittest1.Subtest1': 'm/b/1'
+ }
+ }
+ analysis.put()
+
+ failure_info = {
+ 'failed': True,
+ 'master_name': 'm',
+ 'builder_name': 'b',
+ 'build_number': 2,
+ 'chromium_revision': None,
+ 'builds': {
+ 2: {
+ 'blame_list': [],
+ 'chromium_revision': None
+ }
+ },
+ 'failed_steps': {
+ 'abc_test': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'tests': {
+ 'Unittest2.Subtest1': {
+ 'current_failure': 2,
+ 'first_failure': 2,
+ 'last_pass': 1,
+ 'base_test_name': 'Unittest2.Subtest1'
+ },
+ 'Unittest3.Subtest2': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'base_test_name': 'Unittest3.Subtest2'
+ }
+ }
+ },
+ 'a_tests': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'tests': {
+ 'Unittest1.Subtest1': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'base_test_name': 'Unittest3.Subtest2'
+ }
+ }
+ }
+ },
+ 'failure_type': failure_type.TEST
+ }
+
+ expected_result = {
+ 'abc_test': ['Unittest2.Subtest1']
+ }
+ result = (
+ trigger_swarming_tasks_pipeline._GetStepsThatNeedToTriggerSwarmingTasks(
+ master_name, builder_name, build_number, failure_info))
+ self.assertEqual(result, expected_result)
+
+ def testTriggerSwarmingTasksPipelineNoFailureInfo(self):
+ class _MockedTriggerSwarmingTaskPipeline(BasePipeline):
+ count = 0
+ def run(self, *_): # pragma: no cover
+ _MockedTriggerSwarmingTaskPipeline.count += 1
+ self.mock(trigger_swarming_task_pipeline, 'TriggerSwarmingTaskPipeline',
+ _MockedTriggerSwarmingTaskPipeline)
+ pipeline = TriggerSwarmingTasksPipeline('m', 'b', 1, {})
+ pipeline.start()
+ self.execute_queued_tasks()
+ self.assertEqual(_MockedTriggerSwarmingTaskPipeline.count, 0)
+
+ def testTriggerSwarmingTasksPipelineCompile(self):
+ class _MockedTriggerSwarmingTaskPipeline(BasePipeline):
+ count = 0
+ def run(self, *_): # pragma: no cover
+ _MockedTriggerSwarmingTaskPipeline.count += 1
+ self.mock(trigger_swarming_task_pipeline, 'TriggerSwarmingTaskPipeline',
+ _MockedTriggerSwarmingTaskPipeline)
+ failure_info = {
+ 'failed': True,
+ 'master_name': 'm',
+ 'builder_name': 'b',
+ 'build_number': 2,
+ 'chromium_revision': None,
+ 'builds': {
+ 2: {
+ 'blame_list': [],
+ 'chromium_revision': None
+ }
+ },
+ 'failed_steps': {
+ 'compile': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ }
+ },
+ 'failure_type': failure_type.COMPILE
+ }
+ pipeline = TriggerSwarmingTasksPipeline('m', 'b', 2, failure_info)
+ pipeline.start()
+ self.execute_queued_tasks()
+ self.assertEqual(_MockedTriggerSwarmingTaskPipeline.count, 0)
+
+ def testTriggerSwarmingTasksPipeline(self):
+ master_name = 'm'
+ builder_name = 'b'
+ build_number = 2
+ analysis = WfAnalysis.Create(master_name, builder_name, build_number)
+ analysis.failure_result_map = {
+ 'a_tests': {
+ 'Unittest1.Subtest1': 'm/b/1'
+ }
+ }
+ analysis.put()
+
+ failure_info = {
+ 'failed': True,
+ 'master_name': 'm',
+ 'builder_name': 'b',
+ 'build_number': 2,
+ 'chromium_revision': None,
+ 'builds': {
+ 2: {
+ 'blame_list': [],
+ 'chromium_revision': None
+ }
+ },
+ 'failed_steps': {
+ 'abc_test': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'tests': {
+ 'Unittest2.Subtest1': {
+ 'current_failure': 2,
+ 'first_failure': 2,
+ 'last_pass': 1,
+ 'base_test_name': 'Unittest2.Subtest1'
+ },
+ 'Unittest3.Subtest2': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'base_test_name': 'Unittest3.Subtest2'
+ }
+ }
+ },
+ 'a_tests': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'tests': {
+ 'Unittest1.Subtest1': {
+ 'current_failure': 2,
+ 'first_failure': 1,
+ 'last_pass': 0,
+ 'base_test_name': 'Unittest3.Subtest2'
+ }
+ }
+ }
+ },
+ 'failure_type': failure_type.TEST
+ }
+
+ class _MockedTriggerSwarmingTaskPipeline(BasePipeline):
+ count = 0
+ def run(self, *_):
+ _MockedTriggerSwarmingTaskPipeline.count += 1
+ self.mock(trigger_swarming_task_pipeline, 'TriggerSwarmingTaskPipeline',
+ _MockedTriggerSwarmingTaskPipeline)
+ pipeline = TriggerSwarmingTasksPipeline(
+ master_name, builder_name, build_number, failure_info)
+ pipeline.start()
+ self.execute_queued_tasks()
+ self.assertEqual(_MockedTriggerSwarmingTaskPipeline.count, 1)
« no previous file with comments | « appengine/findit/waterfall/test/swarming_tasks_to_try_job_pipeline_test.py ('k') | appengine/findit/waterfall/test/try_job_util_test.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698