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
|
# Copyright 2018 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.
from .result import SKIPPED
"""
Pipeline
Test processors are chained together and communicate with each other by
calling previous/next processor in the chain.
----next_test()----> ----next_test()---->
Proc1 Proc2 Proc3
<---result_for()---- <---result_for()----
For every next_test there is exactly one result_for call.
If processor ignores the test it has to return SkippedResult.
If it created multiple subtests for one test and wants to pass all of them to
the previous processor it can enclose them in GroupedResult.
Subtests
When test processor needs to modify the test or create some variants of the
test it creates subtests and sends them to the next processor.
Each subtest has:
- procid - globally unique id that should contain id of the parent test and
some suffix given by test processor, e.g. its name + subtest type.
- processor - which created it
- origin - pointer to the parent (sub)test
"""
DROP_RESULT = 0
DROP_OUTPUT = 1
DROP_PASS_OUTPUT = 2
DROP_PASS_STDOUT = 3
def get_reduce_result_function(requirement):
if requirement == DROP_RESULT:
return lambda _: None
if requirement == DROP_OUTPUT:
def f(result):
result.output = None
return result
return f
if requirement == DROP_PASS_OUTPUT:
def f(result):
if not result.has_unexpected_output:
result.output = None
return result
return f
if requirement == DROP_PASS_STDOUT:
def f(result):
if not result.has_unexpected_output:
result.output.stdout = None
result.output.stderr = None
return result
return f
class TestProc(object):
def __init__(self):
self._prev_proc = None
self._next_proc = None
self._requirement = DROP_RESULT
self._prev_requirement = None
self._reduce_result = lambda result: result
def connect_to(self, next_proc):
"""Puts `next_proc` after itself in the chain."""
next_proc._prev_proc = self
self._next_proc = next_proc
def remove_from_chain(self):
if self._prev_proc:
self._prev_proc._next_proc = self._next_proc
if self._next_proc:
self._next_proc._prev_proc = self._prev_proc
def setup(self, requirement=DROP_RESULT):
"""
Method called by previous processor or processor pipeline creator to let
the processors know what part of the result can be ignored.
"""
self._prev_requirement = requirement
if self._next_proc:
self._next_proc.setup(max(requirement, self._requirement))
if self._prev_requirement < self._requirement:
self._reduce_result = get_reduce_result_function(self._prev_requirement)
def next_test(self, test):
"""
Method called by previous processor whenever it produces new test.
This method shouldn't be called by anyone except previous processor.
"""
raise NotImplementedError()
def result_for(self, test, result):
"""
Method called by next processor whenever it has result for some test.
This method shouldn't be called by anyone except next processor.
"""
raise NotImplementedError()
def heartbeat(self):
if self._prev_proc:
self._prev_proc.heartbeat()
### Communication
def _send_test(self, test):
"""Helper method for sending test to the next processor."""
self._next_proc.next_test(test)
def _send_result(self, test, result):
"""Helper method for sending result to the previous processor."""
result = self._reduce_result(result)
self._prev_proc.result_for(test, result)
class TestProcObserver(TestProc):
"""Processor used for observing the data."""
def __init__(self):
super(TestProcObserver, self).__init__()
def next_test(self, test):
self._on_next_test(test)
self._send_test(test)
def result_for(self, test, result):
self._on_result_for(test, result)
self._send_result(test, result)
def heartbeat(self):
self._on_heartbeat()
super(TestProcObserver, self).heartbeat()
def _on_next_test(self, test):
"""Method called after receiving test from previous processor but before
sending it to the next one."""
pass
def _on_result_for(self, test, result):
"""Method called after receiving result from next processor but before
sending it to the previous one."""
pass
def _on_heartbeat(self):
pass
class TestProcProducer(TestProc):
"""Processor for creating subtests."""
def __init__(self, name):
super(TestProcProducer, self).__init__()
self._name = name
def next_test(self, test):
self._next_test(test)
def result_for(self, subtest, result):
self._result_for(subtest.origin, subtest, result)
### Implementation
def _next_test(self, test):
raise NotImplementedError()
def _result_for(self, test, subtest, result):
"""
result_for method extended with `subtest` parameter.
Args
test: test used by current processor to create the subtest.
subtest: test for which the `result` is.
result: subtest execution result created by the output processor.
"""
raise NotImplementedError()
### Managing subtests
def _create_subtest(self, test, subtest_id, **kwargs):
"""Creates subtest with subtest id <processor name>-`subtest_id`."""
return test.create_subtest(self, '%s-%s' % (self._name, subtest_id),
**kwargs)
class TestProcFilter(TestProc):
"""Processor for filtering tests."""
def next_test(self, test):
if self._filter(test):
self._send_result(test, SKIPPED)
else:
self._send_test(test)
def result_for(self, test, result):
self._send_result(test, result)
def _filter(self, test):
"""Returns whether test should be filtered out."""
raise NotImplementedError()
|
