[Findit] Adding central config test class for unit tests

appengine/findit/waterfall/test/swarming_tasks_to_try_job_pipeline_test.py

 # 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.
 
 import json
 
-from testing_utils import testing
-
 from common import buildbucket_client
 from common.git_repository import GitRepository
 from model.wf_try_job import WfTryJob
 from pipeline_wrapper import pipeline_handlers
 from waterfall import swarming_util
 from waterfall import trigger_swarming_task_pipeline
-from waterfall import waterfall_config
 from waterfall.swarming_task_request import SwarmingTaskRequest
 from waterfall.swarming_tasks_to_try_job_pipeline import (
     SwarmingTasksToTryJobPipeline)
+from waterfall.test import wf_testcase
 from waterfall.try_job_type import TryJobType
 
 
 _ISOLATED_SERVER = 'https://isolateserver.appspot.com'
 _ISOLATED_STORAGE_URL = 'isolateserver.storage.googleapis.com'
 _SAMPLE_FAILURE_LOG = {
     'per_iteration_data': [
         {
             'TestSuite1.test1': [
                 {
@@ skipping 46 matching lines @@
             'TestSuite1.test3': [
                 {
                     'status': 'FAILURE',
                     'other_info': 'N/A'
                 }
             ]
         }
     ]
 }
 
-_MOCK_SWARMING_SETTINGS = {
-    'task_timeout_hours': 23,
-    'server_query_interval_seconds': 60,
-    'iterations_to_rerun': 10,
-    'server_host': 'chromium-swarm.appspot.com',
-    'default_request_priority': 150,
-    'isolated_storage_url': 'isolateserver.storage.googleapis.com',
-    'isolated_server': 'https://isolateserver.appspot.com',
-    'request_expiration_hours': 20
-}
 
-_MOCK_TRY_JOB_SETTINGS = {
-    'server_query_interval_seconds': 60,
-    'job_timeout_hours': 5,
-    'allowed_response_error_times': 1
-}
-
-
-class SwarmingTasksToTryJobPipelineTest(testing.AppengineTestCase):
+class SwarmingTasksToTryJobPipelineTest(wf_testcase.WaterfallTestCase):
   app_module = pipeline_handlers._APP
 
-  def _MockGetTrybotForWaterfallBuilder(self, *_):
-    def MockedGetTrybotForWaterfallBuilder(*_):
-      return 'linux_chromium_variable', 'master.tryserver.chromium.linux'
-    self.mock(waterfall_config, 'GetTrybotForWaterfallBuilder',
-              MockedGetTrybotForWaterfallBuilder)
-
   def _MockTriggerTryJobs(self, responses):
     def MockedTriggerTryJobs(*_):
       try_job_results = []
       for response in responses:
         if response.get('error'):  # pragma: no cover
           try_job_results.append((
               buildbucket_client.BuildbucketError(response['error']), None))
         else:
           try_job_results.append((
               None, buildbucket_client.BuildbucketBuild(response['build'])))
       return try_job_results
     self.mock(buildbucket_client, 'TriggerTryJobs', MockedTriggerTryJobs)
 
-  def _MockGetSwarmingSettings(self):
-    def _GetMockSwarmingSettings():
-      return _MOCK_SWARMING_SETTINGS
-    self.mock(waterfall_config, 'GetSwarmingSettings', _GetMockSwarmingSettings)
-
-  def _MockGetTryJobSettings(self):
-    def _GetMockTryJobSettings():
-      return _MOCK_TRY_JOB_SETTINGS
-    self.mock(waterfall_config, 'GetTryJobSettings', _GetMockTryJobSettings)
-
   def _MockGetTryJobs(self, build_id):
     def MockedGetTryJobs(*_):
       data = {
           '1': {
               'build': {
                   'id': '1',
                   'url': 'url',
                   'status': 'COMPLETED',
                   'result_details_json': json.dumps({
                       'properties': {
@@ skipping 79 matching lines @@
 
     responses = [
         {
             'build': {
                 'id': '1',
                 'url': 'url',
                 'status': 'SCHEDULED',
             }
         }
     ]
-    self._MockGetTrybotForWaterfallBuilder(master_name, builder_name)
     self._MockTriggerTryJobs(responses)
     self._MockGetTryJobs('1')
     self._MockGetChangeLog('rev2')
-    self._MockGetTryJobSettings()
 
     WfTryJob.Create(master_name, builder_name, build_number).put()
 
     root_pipeline = SwarmingTasksToTryJobPipeline(
         master_name, builder_name, build_number, 'rev1', 'rev2', ['rev2'],
         TryJobType.COMPILE)
     root_pipeline.start()
     self.execute_queued_tasks()
 
     try_job = WfTryJob.Get(master_name, builder_name, build_number)
@@ skipping 24 matching lines @@
 
   def testSuccessfullyScheduleNewTryJobForTest(self):
     master_name = 'm'
     builder_name = 'b'
     build_number = 1
     targeted_tests = {
         'a_test': ['TestSuite1.test1', 'TestSuite1.test3'],
         'b_test': [],  # Non-swarming test.
     }
 
-    self._MockGetSwarmingSettings()
-    self._MockGetTryJobSettings()
-
     # Mocks for TriggerSwarmingTaskPipeline.
     def MockedDownloadSwarmingTaskData(*_):
       return [{'task_id': '1'}, {'task_id': '2'}]
     self.mock(swarming_util, 'ListSwarmingTasksDataByTags',
               MockedDownloadSwarmingTaskData)
 
     def MockedGetSwarmingTaskRequest(ref_task_id, *_):
       self.assertEqual('1', ref_task_id)
       return SwarmingTaskRequest.Deserialize({
           'expiration_secs': 3600,
@@ skipping 58 matching lines @@
     # Mocks for try job pipelines.
     responses = [
         {
             'build': {
                 'id': '2',
                 'url': 'url',
                 'status': 'SCHEDULED',
             }
         }
     ]
-    self._MockGetTrybotForWaterfallBuilder(master_name, builder_name)
     self._MockTriggerTryJobs(responses)
     self._MockGetTryJobs('2')
     self._MockGetChangeLog('rev1')
 
     WfTryJob.Create(master_name, builder_name, build_number).put()
 
     root_pipeline = SwarmingTasksToTryJobPipeline(
         master_name, builder_name, build_number, 'rev0', 'rev1', ['rev1'],
         TryJobType.TEST, None, targeted_tests)
     root_pipeline.start()
@@ skipping 32 matching lines @@
                             'commit_position': '1',
                             'review_url': 'url_1'
                         }
                     }
                 }
             }
         }
     ]
 
     self.assertEqual(expected_try_job_results, try_job.test_results)