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
|
# Copyright 2016 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.
from builtins import str
import copy
import io
import pprint
import os
import yaml
from mobly import keys
from mobly import utils
# An environment variable defining the base location for Mobly logs.
ENV_MOBLY_LOGPATH = 'MOBLY_LOGPATH'
_DEFAULT_LOG_PATH = '/tmp/logs/mobly/'
class MoblyConfigError(Exception):
"""Raised when there is a problem in test configuration file."""
def _validate_test_config(test_config):
"""Validates the raw configuration loaded from the config file.
Making sure the required key 'TestBeds' is present.
"""
required_key = keys.Config.key_testbed.value
if required_key not in test_config:
raise MoblyConfigError('Required key %s missing in test config.' %
required_key)
def _validate_testbed_name(name):
"""Validates the name of a test bed.
Since test bed names are used as part of the test run id, it needs to meet
certain requirements.
Args:
name: The test bed's name specified in config file.
Raises:
MoblyConfigError: The name does not meet any criteria.
"""
if not name:
raise MoblyConfigError("Test bed names can't be empty.")
name = str(name)
for char in name:
if char not in utils.valid_filename_chars:
raise MoblyConfigError('Char "%s" is not allowed in test bed names.' %
char)
def _validate_testbed_configs(testbed_configs):
"""Validates the testbed configurations.
Args:
testbed_configs: A list of testbed configuration dicts.
Raises:
MoblyConfigError: Some parts of the configuration is invalid.
"""
seen_names = set()
# Cross checks testbed configs for resource conflicts.
for config in testbed_configs:
# Check for conflicts between multiple concurrent testbed configs.
# No need to call it if there's only one testbed config.
name = config[keys.Config.key_testbed_name.value]
_validate_testbed_name(name)
# Test bed names should be unique.
if name in seen_names:
raise MoblyConfigError('Duplicate testbed name %s found.' % name)
seen_names.add(name)
def load_test_config_file(test_config_path, tb_filters=None):
"""Processes the test configuration file provied by user.
Loads the configuration file into a dict, unpacks each testbed
config into its own dict, and validate the configuration in the
process.
Args:
test_config_path: Path to the test configuration file.
tb_filters: A subset of test bed names to be pulled from the config
file. If None, then all test beds will be selected.
Returns:
A list of test configuration dicts to be passed to
test_runner.TestRunner.
"""
configs = _load_config_file(test_config_path)
if tb_filters:
tbs = []
for tb in configs[keys.Config.key_testbed.value]:
if tb[keys.Config.key_testbed_name.value] in tb_filters:
tbs.append(tb)
if len(tbs) != len(tb_filters):
raise MoblyConfigError(
'Expect to find %d test bed configs, found %d. Check if'
' you have the correct test bed names.' % (len(tb_filters), len(tbs)))
configs[keys.Config.key_testbed.value] = tbs
mobly_params = configs.get(keys.Config.key_mobly_params.value, {})
# Decide log path.
log_path = mobly_params.get(keys.Config.key_log_path.value, _DEFAULT_LOG_PATH)
if ENV_MOBLY_LOGPATH in os.environ:
log_path = os.environ[ENV_MOBLY_LOGPATH]
log_path = utils.abs_path(log_path)
# Validate configs
_validate_test_config(configs)
_validate_testbed_configs(configs[keys.Config.key_testbed.value])
# Transform config dict from user-facing key mapping to internal config object.
test_configs = []
for original_bed_config in configs[keys.Config.key_testbed.value]:
test_run_config = TestRunConfig()
test_run_config.testbed_name = original_bed_config[
keys.Config.key_testbed_name.value]
# Deprecated, use testbed_name
test_run_config.test_bed_name = test_run_config.testbed_name
test_run_config.log_path = log_path
test_run_config.controller_configs = original_bed_config.get(
keys.Config.key_testbed_controllers.value, {})
test_run_config.user_params = original_bed_config.get(
keys.Config.key_testbed_test_params.value, {})
test_configs.append(test_run_config)
return test_configs
def _load_config_file(path):
"""Loads a test config file.
The test config file has to be in YAML format.
Args:
path: A string that is the full path to the config file, including the
file name.
Returns:
A dict that represents info in the config file.
"""
with io.open(utils.abs_path(path), 'r', encoding='utf-8') as f:
conf = yaml.safe_load(f)
return conf
class TestRunConfig:
"""The data class that holds all the information needed for a test run.
Attributes:
log_path: string, specifies the root directory for all logs written by
a test run.
test_bed_name: [Deprecated, use 'testbed_name' instead]
string, the name of the test bed used by a test run.
testbed_name: string, the name of the test bed used by a test run.
controller_configs: dict, configs used for instantiating controller
objects.
user_params: dict, all the parameters to be consumed by the test logic.
summary_writer: records.TestSummaryWriter, used to write elements to
the test result summary file.
test_class_name_suffix: string, suffix to append to the class name for
reporting. This is used for differentiating the same class
executed with different parameters in a suite.
"""
def __init__(self):
self.log_path = _DEFAULT_LOG_PATH
# Deprecated, use 'testbed_name'
self.test_bed_name = None
self.testbed_name = None
self.controller_configs = {}
self.user_params = {}
self.summary_writer = None
self.test_class_name_suffix = None
def copy(self):
"""Returns a deep copy of the current config.
"""
return copy.deepcopy(self)
def __str__(self):
content = dict(self.__dict__)
content.pop('summary_writer')
return pprint.pformat(content)
|
