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
|
#!/usr/bin/env python3
# Copyright 2021 the V8 project authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Test integrating the sequence processor into a simple test pipeline.
"""
import os
import sys
import unittest
# Needed because the test runner contains relative imports.
TOOLS_PATH = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(TOOLS_PATH)
from testrunner.testproc import base
from testrunner.testproc.loader import LoadProc
from testrunner.testproc.sequence import SequenceProc
class FakeExecutionProc(base.TestProc):
"""Simulates the pipeline sink consuming and running the tests.
Test execution is simulated for each test by calling run().
"""
def __init__(self):
super(FakeExecutionProc, self).__init__()
self.tests = []
def next_test(self, test):
self.tests.append(test)
return True
def run(self):
test = self.tests.pop()
self._send_result(test, test.n)
class FakeResultObserver(base.TestProcObserver):
"""Observer to track all results sent back through the pipeline."""
def __init__(self):
super(FakeResultObserver, self).__init__()
self.tests = set([])
def _on_result_for(self, test, result):
self.tests.add(test.n)
class FakeTest(object):
"""Simple test representation to differentiate light/heavy tests."""
def __init__(self, n, is_heavy):
self.n = n
self.is_heavy = is_heavy
self.keep_output = False
class TestSequenceProc(unittest.TestCase):
def _test(self, tests, batch_size, max_heavy):
# Set up a simple processing pipeline:
# Loader -> observe results -> sequencer -> execution.
loader = LoadProc(iter(tests), initial_batch_size=batch_size)
results = FakeResultObserver()
sequence_proc = SequenceProc(max_heavy)
execution = FakeExecutionProc()
loader.connect_to(results)
results.connect_to(sequence_proc)
sequence_proc.connect_to(execution)
# Fill the execution queue (with the number of tests potentially
# executed in parallel).
loader.load_initial_tests()
# Simulate the execution test by test.
while execution.tests:
# Assert the invariant of maximum heavy tests executed simultaneously.
self.assertLessEqual(
sum(int(test.is_heavy) for test in execution.tests), max_heavy)
# As in the real pipeline, running a test and returning its result
# will add another test into the pipeline.
execution.run()
# Ensure that all tests are processed and deliver results.
self.assertEqual(set(test.n for test in tests), results.tests)
def test_wrong_usage(self):
with self.assertRaises(Exception):
SequenceProc(0)
def test_no_tests(self):
self._test([], 1, 1)
def test_large_batch_light(self):
self._test([
FakeTest(0, False),
FakeTest(1, False),
FakeTest(2, False),
], 4, 1)
def test_small_batch_light(self):
self._test([
FakeTest(0, False),
FakeTest(1, False),
FakeTest(2, False),
], 2, 1)
def test_large_batch_heavy(self):
self._test([
FakeTest(0, True),
FakeTest(1, True),
FakeTest(2, True),
], 4, 1)
def test_small_batch_heavy(self):
self._test([
FakeTest(0, True),
FakeTest(1, True),
FakeTest(2, True),
], 2, 1)
def test_large_batch_mixed(self):
self._test([
FakeTest(0, True),
FakeTest(1, False),
FakeTest(2, True),
FakeTest(3, False),
], 4, 1)
def test_small_batch_mixed(self):
self._test([
FakeTest(0, True),
FakeTest(1, False),
FakeTest(2, True),
FakeTest(3, False),
], 2, 1)
def test_large_batch_more_heavy(self):
self._test([
FakeTest(0, True),
FakeTest(1, True),
FakeTest(2, True),
FakeTest(3, False),
FakeTest(4, True),
FakeTest(5, True),
FakeTest(6, False),
], 4, 2)
def test_small_batch_more_heavy(self):
self._test([
FakeTest(0, True),
FakeTest(1, True),
FakeTest(2, True),
FakeTest(3, False),
FakeTest(4, True),
FakeTest(5, True),
FakeTest(6, False),
], 2, 2)
if __name__ == '__main__':
unittest.main()
|
