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
|
# This file is part of cloud-init. See LICENSE file for license information.
"""Used to collect data from platforms during tests."""
from functools import partial
import os
from cloudinit import util as c_util
from tests.cloud_tests import (config, LOG, setup_image, util)
from tests.cloud_tests.stage import (PlatformComponent, run_stage, run_single)
from tests.cloud_tests import (platforms, images, snapshots, instances)
def collect_script(instance, base_dir, script, script_name):
"""Collect script data.
@param instance: instance to run script on
@param base_dir: base directory for output data
@param script: script contents
@param script_name: name of script to run
@return_value: None, may raise errors
"""
LOG.debug('running collect script: %s', script_name)
(out, err, exit) = instance.run_script(
script, rcs=range(0, 256),
description='collect: {}'.format(script_name))
c_util.write_file(os.path.join(base_dir, script_name), out)
def collect_test_data(args, snapshot, os_name, test_name):
"""Collect data for test case.
@param args: cmdline arguments
@param snapshot: instantiated snapshot
@param test_name: name or path of test to run
@return_value: tuple of results and fail count
"""
res = ({}, 1)
# load test config
test_name = config.path_to_name(test_name)
test_config = config.load_test_config(test_name)
user_data = test_config['cloud_config']
test_scripts = test_config['collect_scripts']
test_output_dir = os.sep.join(
(args.data_dir, snapshot.platform_name, os_name, test_name))
# if test is not enabled, skip and return 0 failures
if not test_config.get('enabled', False):
LOG.warning('test config %s is not enabled, skipping', test_name)
return ({}, 0)
# if testcase requires a feature flag that the image does not support,
# skip the testcase with a warning
req_features = test_config.get('required_features', [])
if any(feature not in snapshot.features for feature in req_features):
LOG.warn('test config %s requires features not supported by image, '
'skipping.\nrequired features: %s\nsupported features: %s',
test_name, req_features, snapshot.features)
return ({}, 0)
# if there are user data overrides required for this test case, apply them
overrides = snapshot.config.get('user_data_overrides', {})
if overrides:
LOG.debug('updating user data for collect with: %s', overrides)
user_data = util.update_user_data(user_data, overrides)
# create test instance
component = PlatformComponent(
partial(instances.get_instance, snapshot, user_data,
block=True, start=False, use_desc=test_name))
LOG.info('collecting test data for test: %s', test_name)
with component as instance:
start_call = partial(run_single, 'boot instance', partial(
instance.start, wait=True, wait_for_cloud_init=True))
collect_calls = [partial(run_single, 'script {}'.format(script_name),
partial(collect_script, instance,
test_output_dir, script, script_name))
for script_name, script in test_scripts.items()]
res = run_stage('collect for test: {}'.format(test_name),
[start_call] + collect_calls)
return res
def collect_snapshot(args, image, os_name):
"""Collect data for snapshot of image.
@param args: cmdline arguments
@param image: instantiated image with set up complete
@return_value tuple of results and fail count
"""
res = ({}, 1)
component = PlatformComponent(partial(snapshots.get_snapshot, image))
LOG.debug('creating snapshot for %s', os_name)
with component as snapshot:
LOG.info('collecting test data for os: %s', os_name)
res = run_stage(
'collect test data for {}'.format(os_name),
[partial(collect_test_data, args, snapshot, os_name, test_name)
for test_name in args.test_config])
return res
def collect_image(args, platform, os_name):
"""Collect data for image.
@param args: cmdline arguments
@param platform: instantiated platform
@param os_name: name of distro to collect for
@return_value: tuple of results and fail count
"""
res = ({}, 1)
os_config = config.load_os_config(
platform.platform_name, os_name, require_enabled=True,
feature_overrides=args.feature_override)
LOG.debug('os config: %s', os_config)
component = PlatformComponent(
partial(images.get_image, platform, os_config))
LOG.info('acquiring image for os: %s', os_name)
with component as image:
res = run_stage('set up and collect data for os: {}'.format(os_name),
[partial(setup_image.setup_image, args, image)] +
[partial(collect_snapshot, args, image, os_name)],
continue_after_error=False)
return res
def collect_platform(args, platform_name):
"""Collect data for platform.
@param args: cmdline arguments
@param platform_name: platform to collect for
@return_value: tuple of results and fail count
"""
res = ({}, 1)
platform_config = config.load_platform_config(
platform_name, require_enabled=True)
platform_config['data_dir'] = args.data_dir
LOG.debug('platform config: %s', platform_config)
component = PlatformComponent(
partial(platforms.get_platform, platform_name, platform_config))
LOG.info('setting up platform: %s', platform_name)
with component as platform:
res = run_stage('collect for platform: {}'.format(platform_name),
[partial(collect_image, args, platform, os_name)
for os_name in args.os_name])
return res
def collect(args):
"""Entry point for collection.
@param args: cmdline arguments
@return_value: fail count
"""
(res, failed) = run_stage(
'collect data', [partial(collect_platform, args, platform_name)
for platform_name in args.platform])
LOG.debug('collect stages: %s', res)
if args.result:
util.merge_results({'collect_stages': res}, args.result)
return failed
# vi: ts=4 expandtab
|
