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
|
# 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.
"""
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
class TestProc(object):
def __init__(self):
self._prev_proc = None
self._next_proc = None
self._stopped = False
self._requirement = DROP_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 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.
Returns a boolean value to signal whether the test was loaded into the
execution queue successfully or not.
"""
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()
def stop(self):
if not self._stopped:
self._stopped = True
if self._prev_proc:
self._prev_proc.stop()
if self._next_proc:
self._next_proc.stop()
@property
def is_stopped(self):
return self._stopped
### Communication
def notify_previous(self, event):
self._on_event(event)
if self._prev_proc:
self._prev_proc.notify_previous(event)
def _on_event(self, event):
"""Called when processors to the right signal events, e.g. termination.
Args:
event: A text describing the signalled event.
"""
pass
def _send_test(self, test):
"""Helper method for sending test to the next processor."""
return self._next_proc.next_test(test)
def _send_result(self, test, result):
"""Helper method for sending result to the previous processor."""
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)
return 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):
return 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):
return False
return 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()
|
