[Findit] Adding central config test class for unit tests

appengine/findit/waterfall/test/swarming_util_test.py

 # Copyright 2015 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 collections
 import json
 import os
 import urllib
 import zlib
 
-from testing_utils import testing
-
 from common.retry_http_client import RetryHttpClient
+from model.wf_config import FinditConfig
 from model.wf_step import WfStep
 from waterfall import swarming_util
 from waterfall import waterfall_config
 from waterfall.swarming_task_request import SwarmingTaskRequest
-
-
-_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
-}
+from waterfall.test import wf_testcase
 
 
 class SwarmingHttpClient(RetryHttpClient):
 
   def __init__(self):
     self.get_responses = dict()
     self.post_responses = dict()
 
   def _GetData(self, data_type, file_name=None):
     file_name_map = {
@@ skipping 11 matching lines @@
   def _SetResponseForGetRequestIsolated(self, url, file_hash):
     self.get_responses[url] = self._GetData('isolated', file_hash)
 
   def _SetResponseForGetRequestSwarmingList(
       self, master_name, builder_name, build_number, step_name=None):
     if builder_name == 'download_failed':
       return
 
     url = ('https://%s/_ah/api/swarming/v1/tasks/'
            'list?tags=%s&tags=%s&tags=%s') % (
-               _MOCK_SWARMING_SETTINGS.get('server_host'),
+               FinditConfig().Get().swarming_settings.get('server_host'),
                urllib.quote('master:%s' % master_name),
                urllib.quote('buildername:%s' % builder_name),
                urllib.quote('buildnumber:%d' % build_number))
 
     if step_name:
       url += '&tags=%s' % urllib.quote('stepname:%s' % step_name)
       response = self._GetData('step')
     else:
       response = self._GetData('build')
 
     cursor_swarming_data = {
         'cursor': None,
         'items': [],
         'state': 'all',
         'limit': 100,
         'sort': 'created_ts'
     }
     cursor_url = ('%s&cursor=thisisacursor') % url
 
     self.get_responses[url] = response
     self.get_responses[cursor_url] = json.dumps(cursor_swarming_data)
 
   def _SetResponseForGetRequestSwarmingResult(self, task_id):
     url = ('https://%s/_ah/api/swarming/v1/task/%s/result') % (
-        _MOCK_SWARMING_SETTINGS.get('server_host'), task_id)
+        FinditConfig().Get().swarming_settings.get('server_host'), task_id)
 
     response = self._GetData('task')
     self.get_responses[url] = response
 
   def _SetResponseForPostRequest(self, isolated_hash):
     if isolated_hash == 'not found':
       response = '{"content":"eJyrrgUAAXUA+Q=="}'
     else:
       response = self._GetData('isolated', isolated_hash)
 
@@ skipping 31 matching lines @@
   def _Put(self, *_):  # pragma: no cover
     pass
 
   def SetResponse(self, method, url, content=None, status_code=200):
     self.responses[method][url] = (status_code, content)
 
   def GetRequest(self, url):
     return self.requests.get(url)
 
 
-class SwarmingUtilTest(testing.AppengineTestCase):
+class SwarmingUtilTest(wf_testcase.WaterfallTestCase):
 
   def setUp(self):
     super(SwarmingUtilTest, self).setUp()
     self.http_client = SwarmingHttpClient()
     self.logged_http_client = _LoggedHttpClient()
 
-    def _MockGetSwarmingSettings():
-      return _MOCK_SWARMING_SETTINGS
-
-    self.mock(waterfall_config, 'GetSwarmingSettings', _MockGetSwarmingSettings)
-
   def testGetSwarmingTaskRequest(self):
     task_request_json = {
         'expiration_secs': 2,
         'name': 'name',
         'parent_task_id': 'pti',
         'priority': 1,
         'properties': {
             'command': 'cmd',
             'dimensions': [{'key': 'd', 'value': 'dv'}],
             'env': [{'key': 'e', 'value': 'ev'}],
@@ skipping 203 matching lines @@
     self.assertEqual(expected_data, data)
 
   def testGetIsolatedDataForStepFailed(self):
     master_name = 'm'
     builder_name = 'download_failed'
     build_number = 223
     step_name = 's1'
 
     self.http_client._SetResponseForGetRequestSwarmingList(
         master_name, builder_name, build_number, step_name)
+
     task_ids = swarming_util.GetIsolatedDataForStep(
         master_name, builder_name, build_number, step_name,
         self.http_client)
     expected_task_ids = []
 
     self.assertEqual(expected_task_ids, task_ids)
 
   def testDownloadTestResults(self):
     isolated_data = {
         'digest': 'shard1_isolated',
@@ skipping 202 matching lines @@
     expected_failure_log = {
         'failures': ['Test1'],
         'valid': True
     }
 
     self.assertEqual(expected_failure_log, failure_log)
 
   def testGetTagValueInvalidTag(self):
     tags = ['a:1', 'b:2']
     self.assertIsNone(swarming_util.GetTagValue(tags, 'c'))