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
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
|
"""Unit tests for the resmokelib.testing.executor module."""
import logging
import threading
import unittest
import mock
from buildscripts.resmokelib import errors
from buildscripts.resmokelib.testing import job
from buildscripts.resmokelib.testing import queue_element
from buildscripts.resmokelib.testing.fixtures import interface as _fixtures
from buildscripts.resmokelib.utils import queue as _queue
# pylint: disable=missing-docstring,protected-access
class TestJob(unittest.TestCase):
TESTS = ["jstests/core/and.js", "jstests/core/or.js"]
@staticmethod
def mock_testcase(test_name):
testcase = mock.Mock()
testcase.test_name = test_name
testcase.REGISTERED_NAME = "js_test"
testcase.logger = logging.getLogger("job_unittest")
return testcase
@staticmethod
def mock_interrupt_flag():
interrupt_flag = mock.Mock()
interrupt_flag.is_set = lambda: False
return interrupt_flag
@staticmethod
def get_suite_options(num_repeat_tests=None, time_repeat_tests_secs=None,
num_repeat_tests_min=None, num_repeat_tests_max=None):
suite_options = mock.Mock()
suite_options.num_repeat_tests = num_repeat_tests
suite_options.time_repeat_tests_secs = time_repeat_tests_secs
suite_options.num_repeat_tests_min = num_repeat_tests_min
suite_options.num_repeat_tests_max = num_repeat_tests_max
return suite_options
@staticmethod
def queue_tests(tests, queue, queue_elem_type, suite_options):
for test in tests:
queue_elem = queue_elem_type(TestJob.mock_testcase(test), {}, suite_options)
queue.put(queue_elem)
@staticmethod
def expected_run_num(time_repeat_tests_secs, test_time_secs):
"""Return the number of times a test is expected to run."""
return time_repeat_tests_secs / test_time_secs
def test__run_num_repeat(self):
num_repeat_tests = 1
queue = _queue.Queue()
suite_options = self.get_suite_options(num_repeat_tests=num_repeat_tests)
mock_time = MockTime(1)
job_object = UnitJob(suite_options)
self.queue_tests(self.TESTS, queue, queue_element.QueueElem, suite_options)
job_object._get_time = mock_time.time
job_object._run(queue, self.mock_interrupt_flag())
self.assertEqual(job_object.total_test_num, num_repeat_tests * len(self.TESTS))
for test in self.TESTS:
self.assertEqual(job_object.tests[test], num_repeat_tests)
def test__run_time_repeat_time_no_min_max(self):
increment = 1
time_repeat_tests_secs = 10
expected_tests_run = self.expected_run_num(time_repeat_tests_secs, increment)
queue = _queue.Queue()
suite_options = self.get_suite_options(time_repeat_tests_secs=time_repeat_tests_secs)
mock_time = MockTime(increment)
job_object = UnitJob(suite_options)
self.queue_tests(self.TESTS, queue, queue_element.QueueElemRepeatTime, suite_options)
job_object._get_time = mock_time.time
job_object._run(queue, self.mock_interrupt_flag())
self.assertEqual(job_object.total_test_num, expected_tests_run * len(self.TESTS))
for test in self.TESTS:
self.assertEqual(job_object.tests[test], expected_tests_run)
def test__run_time_repeat_time_no_min(self):
increment = 1
time_repeat_tests_secs = 10
num_repeat_tests_max = 100
expected_tests_run = self.expected_run_num(time_repeat_tests_secs, increment)
queue = _queue.Queue()
suite_options = self.get_suite_options(time_repeat_tests_secs=time_repeat_tests_secs,
num_repeat_tests_max=num_repeat_tests_max)
mock_time = MockTime(increment)
job_object = UnitJob(suite_options)
self.queue_tests(self.TESTS, queue, queue_element.QueueElemRepeatTime, suite_options)
job_object._get_time = mock_time.time
job_object._run(queue, self.mock_interrupt_flag())
self.assertLess(job_object.total_test_num, num_repeat_tests_max * len(self.TESTS))
for test in self.TESTS:
self.assertEqual(job_object.tests[test], expected_tests_run)
def test__run_time_repeat_time_no_max(self):
increment = 1
time_repeat_tests_secs = 10
num_repeat_tests_min = 1
expected_tests_run = self.expected_run_num(time_repeat_tests_secs, increment)
queue = _queue.Queue()
suite_options = self.get_suite_options(time_repeat_tests_secs=time_repeat_tests_secs,
num_repeat_tests_min=num_repeat_tests_min)
mock_time = MockTime(increment)
job_object = UnitJob(suite_options)
self.queue_tests(self.TESTS, queue, queue_element.QueueElemRepeatTime, suite_options)
job_object._get_time = mock_time.time
job_object._run(queue, self.mock_interrupt_flag())
self.assertGreater(job_object.total_test_num, num_repeat_tests_min * len(self.TESTS))
for test in self.TESTS:
self.assertEqual(job_object.tests[test], expected_tests_run)
def test__run_time_repeat_time(self):
increment = 1
time_repeat_tests_secs = 10
num_repeat_tests_min = 1
num_repeat_tests_max = 100
expected_tests_run = self.expected_run_num(time_repeat_tests_secs, increment)
queue = _queue.Queue()
suite_options = self.get_suite_options(time_repeat_tests_secs=time_repeat_tests_secs,
num_repeat_tests_min=num_repeat_tests_min,
num_repeat_tests_max=num_repeat_tests_max)
mock_time = MockTime(increment)
job_object = UnitJob(suite_options)
self.queue_tests(self.TESTS, queue, queue_element.QueueElemRepeatTime, suite_options)
job_object._get_time = mock_time.time
job_object._run(queue, self.mock_interrupt_flag())
self.assertGreater(job_object.total_test_num, num_repeat_tests_min * len(self.TESTS))
self.assertLess(job_object.total_test_num, num_repeat_tests_max * len(self.TESTS))
for test in self.TESTS:
self.assertEqual(job_object.tests[test], expected_tests_run)
def test__run_time_repeat_min(self):
increment = 1
time_repeat_tests_secs = 2
num_repeat_tests_min = 3
num_repeat_tests_max = 100
queue = _queue.Queue()
suite_options = self.get_suite_options(time_repeat_tests_secs=time_repeat_tests_secs,
num_repeat_tests_min=num_repeat_tests_min,
num_repeat_tests_max=num_repeat_tests_max)
mock_time = MockTime(increment)
job_object = UnitJob(suite_options)
self.queue_tests(self.TESTS, queue, queue_element.QueueElemRepeatTime, suite_options)
job_object._get_time = mock_time.time
job_object._run(queue, self.mock_interrupt_flag())
self.assertEqual(job_object.total_test_num, num_repeat_tests_min * len(self.TESTS))
for test in self.TESTS:
self.assertEqual(job_object.tests[test], num_repeat_tests_min)
def test__run_time_repeat_max(self):
increment = 1
time_repeat_tests_secs = 30
num_repeat_tests_min = 1
num_repeat_tests_max = 10
expected_time_repeat_tests = self.expected_run_num(time_repeat_tests_secs, increment)
queue = _queue.Queue()
suite_options = self.get_suite_options(time_repeat_tests_secs=time_repeat_tests_secs,
num_repeat_tests_min=num_repeat_tests_min,
num_repeat_tests_max=num_repeat_tests_max)
mock_time = MockTime(increment)
job_object = UnitJob(suite_options)
self.queue_tests(self.TESTS, queue, queue_element.QueueElemRepeatTime, suite_options)
job_object._get_time = mock_time.time
job_object._run(queue, self.mock_interrupt_flag())
self.assertEqual(job_object.total_test_num, num_repeat_tests_max * len(self.TESTS))
for test in self.TESTS:
self.assertEqual(job_object.tests[test], num_repeat_tests_max)
self.assertLess(job_object.tests[test], expected_time_repeat_tests)
class MockTime(object):
"""Class to mock time.time."""
def __init__(self, increment):
"""Initialize with an increment which simulates a time increment."""
self._time = 0
self._increment = increment
def time(self):
"""Simulate time.time by incrementing for every invocation."""
cur_time = self._time
self._time += self._increment
return cur_time
class UnitJob(job.Job): # pylint: disable=too-many-instance-attributes
def __init__(self, suite_options): #pylint: disable=super-init-not-called
self.job_num = 0
self.logger = logging.getLogger("job_unittest")
self.fixture = None
self.hooks = []
self.report = None
self.archival = None
self.suite_options = suite_options
self.test_queue_logger = logging.getLogger("job_unittest")
self.total_test_num = 0
self.tests = {}
def _execute_test(self, test):
self.total_test_num += 1
if test.test_name not in self.tests:
self.tests[test.test_name] = 0
self.tests[test.test_name] += 1
class TestFixtureSetupAndTeardown(unittest.TestCase):
"""Test cases for error handling around setup_fixture() and teardown_fixture()."""
def setUp(self):
logger = logging.getLogger("job_unittest")
self.__job_object = job.Job(job_num=0, logger=logger, fixture=None, hooks=[], report=None,
archival=None, suite_options=None, test_queue_logger=logger)
# Initialize the Job instance such that its setup_fixture() and teardown_fixture() methods
# always indicate success. The settings for these mocked method will be changed in the
# individual test cases below.
self.__job_object.setup_fixture = mock.Mock(return_value=True)
self.__job_object.teardown_fixture = mock.Mock(return_value=True)
def __assert_when_run_tests(self, setup_succeeded=True, teardown_succeeded=True):
queue = _queue.Queue()
interrupt_flag = threading.Event()
setup_flag = threading.Event()
teardown_flag = threading.Event()
self.__job_object(queue, interrupt_flag, setup_flag, teardown_flag)
self.assertEqual(setup_succeeded, not interrupt_flag.is_set())
self.assertEqual(teardown_succeeded, not teardown_flag.is_set())
def test_setup_and_teardown_both_succeed(self):
self.__assert_when_run_tests()
def test_setup_returns_failure(self):
self.__job_object.setup_fixture.return_value = False
self.__assert_when_run_tests(setup_succeeded=False)
def test_setup_raises_logging_config_exception(self):
self.__job_object.setup_fixture.side_effect = errors.LoggerRuntimeConfigError(
"Logging configuration error intentionally raised in unit test")
self.__assert_when_run_tests(setup_succeeded=False)
def test_setup_raises_unexpected_exception(self):
self.__job_object.setup_fixture.side_effect = Exception(
"Generic error intentionally raised in unit test")
self.__assert_when_run_tests(setup_succeeded=False)
def test_teardown_returns_failure(self):
self.__job_object.teardown_fixture.return_value = False
self.__assert_when_run_tests(teardown_succeeded=False)
def test_teardown_raises_logging_config_exception(self):
self.__job_object.teardown_fixture.side_effect = errors.LoggerRuntimeConfigError(
"Logging configuration error intentionally raised in unit test")
self.__assert_when_run_tests(teardown_succeeded=False)
def test_teardown_raises_unexpected_exception(self):
self.__job_object.teardown_fixture.side_effect = Exception(
"Generic error intentionally raised in unit test")
self.__assert_when_run_tests(teardown_succeeded=False)
class TestNoOpFixtureSetupAndTeardown(unittest.TestCase):
"""Test cases for NoOpFixture handling in setup_fixture() and teardown_fixture()."""
def setUp(self):
logger = logging.getLogger("job_unittest")
self.__noop_fixture = _fixtures.NoOpFixture(logger=logger, job_num=0)
self.__noop_fixture.setup = mock.Mock()
self.__noop_fixture.teardown = mock.Mock()
test_report = mock.Mock()
test_report.find_test_info().status = "pass"
self.__job_object = job.Job(job_num=0, logger=logger, fixture=self.__noop_fixture, hooks=[],
report=test_report, archival=None, suite_options=None,
test_queue_logger=logger)
def test_setup_called_for_noop_fixture(self):
self.assertTrue(self.__job_object.setup_fixture())
self.__noop_fixture.setup.assert_called_once_with()
def test_teardown_called_for_noop_fixture(self):
self.assertTrue(self.__job_object.teardown_fixture())
self.__noop_fixture.teardown.assert_called_once_with(finished=True)
|
