1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227
|
# Copyright 2018 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import json
import logging
import unittest
from unittest import mock
from unittest.mock import call
from core import results_dashboard
class ResultsDashboardTest(unittest.TestCase):
def setUp(self):
self.dummy_token_generator = lambda: 'Arthur-Merlin'
self.perf_data = {'foo': 1, 'bar': 2}
self.dashboard_url = 'https://chromeperf.appspot.com'
self.dummy_token = {'token': 'dummy1234', 'file': 'path/to/file'}
def testRetryForSendResultRetryException(self):
def raise_retry_exception(url, histogramset_json, token_generator_callback):
del url, histogramset_json # unused
del token_generator_callback # unused
raise results_dashboard.SendResultsRetryException('Should retry')
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch(
'core.results_dashboard._SendHistogramJson',
side_effect=raise_retry_exception) as m:
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
self.assertFalse(upload_result)
self.assertEqual(m.call_count, 5)
self.assertEqual(
sleep_mock.mock_calls,
[call(15), call(30),
call(60), call(120),
call(240)])
def testNoRetryForSendResultFatalException(self):
def raise_retry_exception(url, histogramset_json, token_generator_callback):
del url, histogramset_json # unused
del token_generator_callback # unused
raise results_dashboard.SendResultsFatalException('Do not retry')
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch(
'core.results_dashboard._SendHistogramJson',
side_effect=raise_retry_exception) as m:
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
self.assertFalse(upload_result)
self.assertEqual(m.call_count, 1)
self.assertFalse(sleep_mock.mock_calls)
def testNoRetryForSuccessfulSendResult(self):
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch('core.results_dashboard._SendHistogramJson') as m:
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
self.assertTrue(upload_result)
self.assertEqual(m.call_count, 1)
self.assertFalse(sleep_mock.mock_calls)
def testNoRetryAfterSucessfulSendResult(self):
counter = [0]
def raise_retry_exception_first_two_times(url, histogramset_json,
token_generator_callback):
del url, histogramset_json # unused
del token_generator_callback # unused
counter[0] += 1
if counter[0] <= 2:
raise results_dashboard.SendResultsRetryException('Please retry')
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch(
'core.results_dashboard._SendHistogramJson',
side_effect=raise_retry_exception_first_two_times) as m:
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
self.assertTrue(upload_result)
self.assertEqual(m.call_count, 3)
self.assertEqual(sleep_mock.mock_calls, [call(15), call(30)])
def testSendHistogramJsonRetryException(self):
def request(*args, **kwargs):
del args, kwargs
content = json.dumps(self.dummy_token)
class Response(object):
status = 403
reason = 'dummy auth error'
return Response(), content
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch.object(
results_dashboard.httplib2.Http, 'request', new=request):
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
self.assertFalse(upload_result)
self.assertEqual(
sleep_mock.mock_calls,
[call(15), call(30),
call(60), call(120),
call(240)])
def testSendHistogramJsonTypeWarning(self):
def request(*args, **kwargs):
del args, kwargs
content = 'non json response'
class Response(object):
status = 200
return Response(), content
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch.object(
results_dashboard.httplib2.Http, 'request', new=request):
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
# upload should still be successful, despite failing to fetch token.
self.assertTrue(upload_result)
self.assertFalse(sleep_mock.mock_calls)
def testSendHistogramJsonValueWarning(self):
def request(*args, **kwargs):
del args, kwargs
content = json.dumps(None)
class Response(object):
status = 200
return Response(), content
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch.object(
results_dashboard.httplib2.Http, 'request', new=request):
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
# upload should still be successful, despite failing to fetch token.
self.assertTrue(upload_result)
self.assertFalse(sleep_mock.mock_calls)
def testSuccessfulSendHistogramJson(self):
def request(*args, **kwargs):
del args, kwargs
content = json.dumps(self.dummy_token)
class Response(object):
status = 200
return Response(), content
with mock.patch('core.results_dashboard.time.sleep') as sleep_mock:
with mock.patch.object(
results_dashboard.httplib2.Http, 'request', new=request):
with mock.patch.object(logging, 'info') as log_mock:
upload_result = results_dashboard.SendResults(
self.perf_data,
'dummy_benchmark',
self.dashboard_url,
send_as_histograms=True,
token_generator_callback=self.dummy_token_generator,
num_retries=5)
self.assertTrue(upload_result)
self.assertFalse(sleep_mock.mock_calls)
# check token id is logged to info.
found_token = False
for log_call in log_mock.call_args_list:
log = log_call[0][0] % tuple(log_call[0][1:])
if 'dummy1234' in log:
found_token = True
break
self.assertTrue(
found_token, msg='Upload completion token not found in logs.')
|