-
Notifications
You must be signed in to change notification settings - Fork 5
/
block.py
executable file
·164 lines (133 loc) · 7.28 KB
/
block.py
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
"""Describe TestBlock class."""
# pylint: disable=dangerous-default-value,too-many-arguments
from itertools import count
from rotest.common.utils import get_class_fields
from rotest.common.config import ROTEST_WORK_DIR
from rotest.core.flow_component import (AbstractFlowComponent, MODE_OPTIONAL,
MODE_FINALLY, MODE_CRITICAL,
BlockInput, BlockOutput)
assert MODE_FINALLY
assert MODE_CRITICAL
assert MODE_OPTIONAL
class TestBlock(AbstractFlowComponent):
"""Define TestBlock, which is a part of a test.
Defines tests that are parts of a greater test. The block is dependent on
the containing :class:`rotest.core.flow.TestFlow` to lock and pass
resources to it.
Test blocks can only be skipped because of test related issues
(e.g. no resources, test-flow stopped due to failure, intentional skip)
and can't be skipped on account of 'run_delta' (passed in previous runs),
tags filtering, etc.
Declaring 'inputs': assign class fields to instances of BlockInput to
ask for values for the block (values are passed via common, parametrize,
previous blocks passing them as outputs, or as requested resources).
You can pass a default value to BlockInput to assign if non is supplied
(making it an optional input).
Declaring 'outputs': assign class fields to instances of BlockOutput to
share values from the instance (self) to the parent and siblings.
the block automatically shares the declared outputs after teardown.
In case the blocks under a flow don't 'connect' properly (a block doesn't
have its declared output in self.__dict__ or a block doesn't get all its
inputs from) an error would be raised before the tests start.
Test authors should subclass TestBlock for their own tests and override
'inputs' tuple with the names of the fields required for the run of the
block, and override 'mode' to state the type of the block.
Attributes:
resources (tuple): list of the required resources. each item is a
tuple of (resource_name, resource type, parameters dictionary),
you can use :func:`rotest.core..request` to create the tuple.
identifier (number): unique id of the test.
data (rotest.core.models._data.Data): contain information
about a test run.
logger (logging.Logger): test logger.
save_state (bool): a flag to determine if storing the states of
resources is required.
force_initialize (bool): a flag to determine if the resources will be
initialized even if their validation succeeds.
config (AttrDict): dictionary of configurations.
enable_debug (bool): whether to enable entering ipdb debugging mode
upon any exception in a test statement.
resource_manager (ClientResourceManager): client resource manager.
skip_init (bool): True to skip resources initialize and validation.
mode (number): running mode code. available modes are:
CRITICAL: stop test flow on failure or error.
FINALLY: always run this block, regardless of the others' result.
OPTIONAL: don't stop test flow on failure (but do so on error),
failure in this type of block still fails the test-flow.
TAGS (list): list of tags by which the test may be filtered.
IS_COMPLEX (bool): if this test is complex (may contain sub-tests).
"""
IS_COMPLEX = False
def __init__(self, indexer=count(), base_work_dir=ROTEST_WORK_DIR,
save_state=True, force_initialize=False, config=None,
parent=None, run_data=None, enable_debug=True,
resource_manager=None, skip_init=False, is_main=True):
super(TestBlock, self).__init__(parent=parent,
config=config,
indexer=indexer,
is_main=is_main,
run_data=run_data,
skip_init=skip_init,
save_state=save_state,
enable_debug=enable_debug,
base_work_dir=base_work_dir,
force_initialize=force_initialize,
resource_manager=resource_manager)
self.addCleanup(self._share_outputs)
self._set_parameters(override_previous=False, **self.__class__.common)
@classmethod
def get_name(cls):
"""Return test name.
You can override this class method and use values from 'common' to
create a more indicative name for the test.
Returns:
str. test name.
"""
class_name = cls.common.get(cls.COMPONENT_NAME_PARAMETER, cls.__name__)
method_name = cls.get_test_method_name()
return '.'.join((class_name, method_name))
@classmethod
def get_inputs(cls):
"""Return a dict of all the input instances of this block.
Returns:
dict. block's inputs (name: input placeholder instance).
"""
return dict(get_class_fields(cls, BlockInput))
@classmethod
def get_outputs(cls):
"""Return a dict of all the input instances of this block.
Returns:
dict. block's inputs (name: input placeholder instance).
"""
return dict(get_class_fields(cls, BlockOutput))
def _share_outputs(self):
"""Share all the declared outputs of the block."""
outputs_dict = {}
for output_name in self.get_outputs():
if output_name not in self.__dict__:
self.logger.warn("Block %r didn't create output %r",
self.data.name, output_name)
outputs_dict[output_name] = getattr(self, output_name)
self.share_data(**outputs_dict)
def validate_inputs(self, extra_inputs=[]):
"""Validate that all the required inputs of the blocks were passed.
All names under the 'inputs' list must be attributes of the test-block
when it begins to run, otherwise the block would raise an exception.
Args:
extra_inputs (list): fields the component would get from its parent
or siblings.
Raises:
AttributeError: not all inputs were passed to the block.
"""
required_inputs = [name
for (name, value) in self.get_inputs().iteritems()
if not value.is_optional()]
required_inputs.extend(self._pipes.itervalues())
missing_inputs = [input_name for input_name in required_inputs
if (input_name not in self.__dict__ and
input_name not in extra_inputs and
input_name not in self._pipes)]
if len(missing_inputs) > 0:
raise AttributeError("Block %r under %r is missing mandatory "
"inputs %s" %
(self.data.name, self.parent, missing_inputs))