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
|
# Copyright 2018 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
""" Module for Mobly controller management."""
import collections
import copy
import logging
import yaml
from mobly import expects
from mobly import records
from mobly import signals
def verify_controller_module(module):
"""Verifies a module object follows the required interface for
controllers.
The interface is explained in the docstring of
`base_test.BaseTestClass.register_controller`.
Args:
module: An object that is a controller module. This is usually
imported with import statements or loaded by importlib.
Raises:
ControllerError: if the module does not match the Mobly controller
interface, or one of the required members is null.
"""
required_attributes = ('create', 'destroy', 'MOBLY_CONTROLLER_CONFIG_NAME')
for attr in required_attributes:
if not hasattr(module, attr):
raise signals.ControllerError(
'Module %s missing required controller module attribute'
' %s.' % (module.__name__, attr))
if not getattr(module, attr):
raise signals.ControllerError(
'Controller interface %s in %s cannot be null.' %
(attr, module.__name__))
class ControllerManager:
"""Manages the controller objects for Mobly tests.
This manages the life cycles and info retrieval of all controller objects
used in a test.
Attributes:
controller_configs: dict, controller configs provided by the user via
test bed config.
"""
def __init__(self, class_name, controller_configs):
# Controller object management.
self._controller_objects = collections.OrderedDict(
) # controller_name: objects
self._controller_modules = {} # controller_name: module
self._class_name = class_name
self.controller_configs = controller_configs
def register_controller(self, module, required=True, min_number=1):
"""Loads a controller module and returns its loaded devices.
This is to be used in a mobly test class.
Args:
module: A module that follows the controller module interface.
required: A bool. If True, failing to register the specified
controller module raises exceptions. If False, the objects
failed to instantiate will be skipped.
min_number: An integer that is the minimum number of controller
objects to be created. Default is one, since you should not
register a controller module without expecting at least one
object.
Returns:
A list of controller objects instantiated from controller_module, or
None if no config existed for this controller and it was not a
required controller.
Raises:
ControllerError:
* The controller module has already been registered.
* The actual number of objects instantiated is less than the
* `min_number`.
* `required` is True and no corresponding config can be found.
* Any other error occurred in the registration process.
"""
verify_controller_module(module)
# Use the module's name as the ref name
module_ref_name = module.__name__.split('.')[-1]
if module_ref_name in self._controller_objects:
raise signals.ControllerError(
'Controller module %s has already been registered. It cannot '
'be registered again.' % module_ref_name)
# Create controller objects.
module_config_name = module.MOBLY_CONTROLLER_CONFIG_NAME
if module_config_name not in self.controller_configs:
if required:
raise signals.ControllerError('No corresponding config found for %s' %
module_config_name)
logging.warning(
'No corresponding config found for optional controller %s',
module_config_name)
return None
try:
# Make a deep copy of the config to pass to the controller module,
# in case the controller module modifies the config internally.
original_config = self.controller_configs[module_config_name]
controller_config = copy.deepcopy(original_config)
objects = module.create(controller_config)
except Exception:
logging.exception(
'Failed to initialize objects for controller %s, abort!',
module_config_name)
raise
if not isinstance(objects, list):
raise signals.ControllerError(
'Controller module %s did not return a list of objects, abort.' %
module_ref_name)
# Check we got enough controller objects to continue.
actual_number = len(objects)
if actual_number < min_number:
module.destroy(objects)
raise signals.ControllerError(
'Expected to get at least %d controller objects, got %d.' %
(min_number, actual_number))
# Save a shallow copy of the list for internal usage, so tests can't
# affect internal registry by manipulating the object list.
self._controller_objects[module_ref_name] = copy.copy(objects)
logging.debug('Found %d objects for controller %s', len(objects),
module_config_name)
self._controller_modules[module_ref_name] = module
return objects
def unregister_controllers(self):
"""Destroy controller objects and clear internal registry.
This will be called after each test class.
"""
# TODO(xpconanfan): actually record these errors instead of just
# logging them.
for name, module in self._controller_modules.items():
logging.debug('Destroying %s.', name)
with expects.expect_no_raises('Exception occurred destroying %s.' % name):
module.destroy(self._controller_objects[name])
self._controller_objects = collections.OrderedDict()
self._controller_modules = {}
def _create_controller_info_record(self, controller_module_name):
"""Creates controller info record for a particular controller type.
Info is retrieved from all the controller objects spawned from the
specified module, using the controller module's `get_info` function.
Args:
controller_module_name: string, the name of the controller module
to retrieve info from.
Returns:
A records.ControllerInfoRecord object.
"""
module = self._controller_modules[controller_module_name]
controller_info = None
try:
controller_info = module.get_info(
copy.copy(self._controller_objects[controller_module_name]))
except AttributeError:
logging.warning(
'No optional debug info found for controller '
'%s. To provide it, implement `get_info`.', controller_module_name)
try:
yaml.dump(controller_info)
except TypeError:
logging.warning(
'The info of controller %s in class "%s" is not '
'YAML serializable! Coercing it to string.', controller_module_name,
self._class_name)
controller_info = str(controller_info)
return records.ControllerInfoRecord(self._class_name,
module.MOBLY_CONTROLLER_CONFIG_NAME,
controller_info)
def get_controller_info_records(self):
"""Get the info records for all the controller objects in the manager.
New info records for each controller object are created for every call
so the latest info is included.
Returns:
List of records.ControllerInfoRecord objects. Each opject conatins
the info of a type of controller
"""
info_records = []
for controller_module_name in self._controller_objects.keys():
with expects.expect_no_raises(
'Failed to collect controller info from %s' % controller_module_name):
record = self._create_controller_info_record(controller_module_name)
if record:
info_records.append(record)
return info_records
|
