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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
|
"""Configuration classes."""
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import sys
from . import types as t
from .util import (
is_shippable,
docker_qualify_image,
find_python,
generate_pip_command,
get_docker_completion,
get_remote_completion,
ApplicationError,
)
from .util_common import (
CommonConfig,
)
from .metadata import (
Metadata,
)
from .data import (
data_context,
)
try:
TIntegrationConfig = t.TypeVar('TIntegrationConfig', bound='IntegrationConfig')
except AttributeError:
TIntegrationConfig = None # pylint: disable=invalid-name
class EnvironmentConfig(CommonConfig):
"""Configuration common to all commands which execute in an environment."""
def __init__(self, args, command):
"""
:type args: any
:type command: str
"""
super(EnvironmentConfig, self).__init__(args, command)
self.local = args.local is True
self.venv = args.venv
self.venv_system_site_packages = args.venv_system_site_packages
self.python = args.python if 'python' in args else None # type: str
self.docker = docker_qualify_image(args.docker) # type: str
self.docker_raw = args.docker # type: str
self.remote = args.remote # type: str
self.docker_privileged = args.docker_privileged if 'docker_privileged' in args else False # type: bool
self.docker_pull = args.docker_pull if 'docker_pull' in args else False # type: bool
self.docker_keep_git = args.docker_keep_git if 'docker_keep_git' in args else False # type: bool
self.docker_seccomp = args.docker_seccomp if 'docker_seccomp' in args else None # type: str
self.docker_memory = args.docker_memory if 'docker_memory' in args else None
if self.docker_seccomp is None:
self.docker_seccomp = get_docker_completion().get(self.docker_raw, {}).get('seccomp', 'default')
self.remote_stage = args.remote_stage # type: str
self.remote_provider = args.remote_provider # type: str
self.remote_aws_region = args.remote_aws_region # type: str
self.remote_terminate = args.remote_terminate # type: str
if self.remote_provider == 'default':
self.remote_provider = None
self.requirements = args.requirements # type: bool
if self.python == 'default':
self.python = None
actual_major_minor = '.'.join(str(i) for i in sys.version_info[:2])
self.python_version = self.python or actual_major_minor
self.python_interpreter = args.python_interpreter
self.pip_check = args.pip_check
self.delegate = self.docker or self.remote or self.venv
self.delegate_args = [] # type: t.List[str]
if self.delegate:
self.requirements = True
self.inject_httptester = args.inject_httptester if 'inject_httptester' in args else False # type: bool
self.httptester = docker_qualify_image(args.httptester if 'httptester' in args else '') # type: str
if self.get_delegated_completion().get('httptester', 'enabled') == 'disabled':
self.httptester = False
if self.get_delegated_completion().get('pip-check', 'enabled') == 'disabled':
self.pip_check = False
if args.check_python and args.check_python != actual_major_minor:
raise ApplicationError('Running under Python %s instead of Python %s as expected.' % (actual_major_minor, args.check_python))
if self.docker_keep_git:
def git_callback(files): # type: (t.List[t.Tuple[str, str]]) -> None
"""Add files from the content root .git directory to the payload file list."""
for dirpath, _dirnames, filenames in os.walk(os.path.join(data_context().content.root, '.git')):
paths = [os.path.join(dirpath, filename) for filename in filenames]
files.extend((path, os.path.relpath(path, data_context().content.root)) for path in paths)
data_context().register_payload_callback(git_callback)
@property
def python_executable(self):
"""
:rtype: str
"""
return find_python(self.python_version)
@property
def pip_command(self):
"""
:rtype: list[str]
"""
return generate_pip_command(self.python_executable)
def get_delegated_completion(self):
"""Returns a dictionary of settings specific to the selected delegation system, if any. Otherwise returns an empty dictionary.
:rtype: dict[str, str]
"""
if self.docker:
return get_docker_completion().get(self.docker_raw, {})
if self.remote:
return get_remote_completion().get(self.remote, {})
return {}
class TestConfig(EnvironmentConfig):
"""Configuration common to all test commands."""
def __init__(self, args, command):
"""
:type args: any
:type command: str
"""
super(TestConfig, self).__init__(args, command)
self.coverage = args.coverage # type: bool
self.coverage_label = args.coverage_label # type: str
self.coverage_check = args.coverage_check # type: bool
self.coverage_config_base_path = None # type: t.Optional[str]
self.include = args.include or [] # type: t.List[str]
self.exclude = args.exclude or [] # type: t.List[str]
self.require = args.require or [] # type: t.List[str]
self.changed = args.changed # type: bool
self.tracked = args.tracked # type: bool
self.untracked = args.untracked # type: bool
self.committed = args.committed # type: bool
self.staged = args.staged # type: bool
self.unstaged = args.unstaged # type: bool
self.changed_from = args.changed_from # type: str
self.changed_path = args.changed_path # type: t.List[str]
self.lint = args.lint if 'lint' in args else False # type: bool
self.junit = args.junit if 'junit' in args else False # type: bool
self.failure_ok = args.failure_ok if 'failure_ok' in args else False # type: bool
self.metadata = Metadata.from_file(args.metadata) if args.metadata else Metadata()
self.metadata_path = None
if self.coverage_check:
self.coverage = True
def metadata_callback(files): # type: (t.List[t.Tuple[str, str]]) -> None
"""Add the metadata file to the payload file list."""
config = self
if data_context().content.collection:
working_path = data_context().content.collection.directory
else:
working_path = ''
if self.metadata_path:
files.append((os.path.abspath(config.metadata_path), os.path.join(working_path, config.metadata_path)))
data_context().register_payload_callback(metadata_callback)
class ShellConfig(EnvironmentConfig):
"""Configuration for the shell command."""
def __init__(self, args):
"""
:type args: any
"""
super(ShellConfig, self).__init__(args, 'shell')
self.raw = args.raw # type: bool
if self.raw:
self.httptester = False
class SanityConfig(TestConfig):
"""Configuration for the sanity command."""
def __init__(self, args):
"""
:type args: any
"""
super(SanityConfig, self).__init__(args, 'sanity')
self.test = args.test # type: t.List[str]
self.skip_test = args.skip_test # type: t.List[str]
self.list_tests = args.list_tests # type: bool
self.allow_disabled = args.allow_disabled # type: bool
if args.base_branch:
self.base_branch = args.base_branch # str
elif is_shippable():
self.base_branch = os.environ.get('BASE_BRANCH', '') # str
if self.base_branch:
self.base_branch = 'origin/%s' % self.base_branch
else:
self.base_branch = ''
self.info_stderr = self.lint
class IntegrationConfig(TestConfig):
"""Configuration for the integration command."""
def __init__(self, args, command):
"""
:type args: any
:type command: str
"""
super(IntegrationConfig, self).__init__(args, command)
self.start_at = args.start_at # type: str
self.start_at_task = args.start_at_task # type: str
self.allow_destructive = args.allow_destructive # type: bool
self.allow_root = args.allow_root # type: bool
self.allow_disabled = args.allow_disabled # type: bool
self.allow_unstable = args.allow_unstable # type: bool
self.allow_unstable_changed = args.allow_unstable_changed # type: bool
self.allow_unsupported = args.allow_unsupported # type: bool
self.retry_on_error = args.retry_on_error # type: bool
self.continue_on_error = args.continue_on_error # type: bool
self.debug_strategy = args.debug_strategy # type: bool
self.changed_all_target = args.changed_all_target # type: str
self.changed_all_mode = args.changed_all_mode # type: str
self.list_targets = args.list_targets # type: bool
self.tags = args.tags
self.skip_tags = args.skip_tags
self.diff = args.diff
self.no_temp_workdir = args.no_temp_workdir
self.no_temp_unicode = args.no_temp_unicode
if self.get_delegated_completion().get('temp-unicode', 'enabled') == 'disabled':
self.no_temp_unicode = True
if self.list_targets:
self.explain = True
self.info_stderr = True
def get_ansible_config(self): # type: () -> str
"""Return the path to the Ansible config for the given config."""
ansible_config_relative_path = os.path.join(data_context().content.integration_path, '%s.cfg' % self.command)
ansible_config_path = os.path.join(data_context().content.root, ansible_config_relative_path)
if not os.path.exists(ansible_config_path):
# use the default empty configuration unless one has been provided
ansible_config_path = super(IntegrationConfig, self).get_ansible_config()
return ansible_config_path
class PosixIntegrationConfig(IntegrationConfig):
"""Configuration for the posix integration command."""
def __init__(self, args):
"""
:type args: any
"""
super(PosixIntegrationConfig, self).__init__(args, 'integration')
class WindowsIntegrationConfig(IntegrationConfig):
"""Configuration for the windows integration command."""
def __init__(self, args):
"""
:type args: any
"""
super(WindowsIntegrationConfig, self).__init__(args, 'windows-integration')
self.windows = args.windows # type: t.List[str]
self.inventory = args.inventory # type: str
if self.windows:
self.allow_destructive = True
class NetworkIntegrationConfig(IntegrationConfig):
"""Configuration for the network integration command."""
def __init__(self, args):
"""
:type args: any
"""
super(NetworkIntegrationConfig, self).__init__(args, 'network-integration')
self.platform = args.platform # type: t.List[str]
self.platform_collection = dict(args.platform_collection or []) # type: t.Dict[str, str]
self.platform_connection = dict(args.platform_connection or []) # type: t.Dict[str, str]
self.inventory = args.inventory # type: str
self.testcase = args.testcase # type: str
class UnitsConfig(TestConfig):
"""Configuration for the units command."""
def __init__(self, args):
"""
:type args: any
"""
super(UnitsConfig, self).__init__(args, 'units')
self.collect_only = args.collect_only # type: bool
self.num_workers = args.num_workers # type: int
self.requirements_mode = args.requirements_mode if 'requirements_mode' in args else ''
if self.requirements_mode == 'only':
self.requirements = True
elif self.requirements_mode == 'skip':
self.requirements = False
|