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
|
"""Loading, parsing and storing of completion configurations."""
from __future__ import annotations
import abc
import dataclasses
import os
import typing as t
from .constants import (
CONTROLLER_PYTHON_VERSIONS,
SUPPORTED_PYTHON_VERSIONS,
)
from .util import (
ANSIBLE_TEST_DATA_ROOT,
cache,
read_lines_without_comments,
)
from .data import (
data_context,
)
@dataclasses.dataclass(frozen=True)
class CompletionConfig(metaclass=abc.ABCMeta):
"""Base class for completion configuration."""
name: str
@property
@abc.abstractmethod
def is_default(self):
"""True if the completion entry is only used for defaults, otherwise False."""
@dataclasses.dataclass(frozen=True)
class PosixCompletionConfig(CompletionConfig, metaclass=abc.ABCMeta):
"""Base class for completion configuration of POSIX environments."""
@property
@abc.abstractmethod
def supported_pythons(self): # type: () -> t.List[str]
"""Return a list of the supported Python versions."""
@abc.abstractmethod
def get_python_path(self, version): # type: (str) -> str
"""Return the path of the requested Python version."""
def get_default_python(self, controller): # type: (bool) -> str
"""Return the default Python version for a controller or target as specified."""
context_pythons = CONTROLLER_PYTHON_VERSIONS if controller else SUPPORTED_PYTHON_VERSIONS
version = [python for python in self.supported_pythons if python in context_pythons][0]
return version
@property
def controller_supported(self): # type: () -> bool
"""True if at least one Python version is provided which supports the controller, otherwise False."""
return any(version in CONTROLLER_PYTHON_VERSIONS for version in self.supported_pythons)
@dataclasses.dataclass(frozen=True)
class PythonCompletionConfig(PosixCompletionConfig, metaclass=abc.ABCMeta):
"""Base class for completion configuration of Python environments."""
python: str = ''
python_dir: str = '/usr/bin'
@property
def supported_pythons(self): # type: () -> t.List[str]
"""Return a list of the supported Python versions."""
versions = self.python.split(',') if self.python else []
versions = [version for version in versions if version in SUPPORTED_PYTHON_VERSIONS]
return versions
def get_python_path(self, version): # type: (str) -> str
"""Return the path of the requested Python version."""
return os.path.join(self.python_dir, f'python{version}')
@dataclasses.dataclass(frozen=True)
class RemoteCompletionConfig(CompletionConfig):
"""Base class for completion configuration of remote environments provisioned through Ansible Core CI."""
provider: t.Optional[str] = None
arch: t.Optional[str] = None
@property
def platform(self):
"""The name of the platform."""
return self.name.partition('/')[0]
@property
def version(self):
"""The version of the platform."""
return self.name.partition('/')[2]
@property
def is_default(self):
"""True if the completion entry is only used for defaults, otherwise False."""
return not self.version
def __post_init__(self):
if not self.provider:
raise Exception(f'Remote completion entry "{self.name}" must provide a "provider" setting.')
if not self.arch:
raise Exception(f'Remote completion entry "{self.name}" must provide a "arch" setting.')
@dataclasses.dataclass(frozen=True)
class InventoryCompletionConfig(CompletionConfig):
"""Configuration for inventory files."""
def __init__(self): # type: () -> None
super().__init__(name='inventory')
@property
def is_default(self): # type: () -> bool
"""True if the completion entry is only used for defaults, otherwise False."""
return False
@dataclasses.dataclass(frozen=True)
class PosixSshCompletionConfig(PythonCompletionConfig):
"""Configuration for a POSIX host reachable over SSH."""
def __init__(self, user, host): # type: (str, str) -> None
super().__init__(
name=f'{user}@{host}',
python=','.join(SUPPORTED_PYTHON_VERSIONS),
)
@property
def is_default(self): # type: () -> bool
"""True if the completion entry is only used for defaults, otherwise False."""
return False
@dataclasses.dataclass(frozen=True)
class DockerCompletionConfig(PythonCompletionConfig):
"""Configuration for Docker containers."""
image: str = ''
seccomp: str = 'default'
placeholder: bool = False
@property
def is_default(self):
"""True if the completion entry is only used for defaults, otherwise False."""
return False
def __post_init__(self):
if not self.image:
raise Exception(f'Docker completion entry "{self.name}" must provide an "image" setting.')
if not self.supported_pythons and not self.placeholder:
raise Exception(f'Docker completion entry "{self.name}" must provide a "python" setting.')
@dataclasses.dataclass(frozen=True)
class NetworkRemoteCompletionConfig(RemoteCompletionConfig):
"""Configuration for remote network platforms."""
collection: str = ''
connection: str = ''
placeholder: bool = False
def __post_init__(self):
if not self.placeholder:
super().__post_init__()
@dataclasses.dataclass(frozen=True)
class PosixRemoteCompletionConfig(RemoteCompletionConfig, PythonCompletionConfig):
"""Configuration for remote POSIX platforms."""
placeholder: bool = False
def __post_init__(self):
if not self.placeholder:
super().__post_init__()
if not self.supported_pythons:
if self.version and not self.placeholder:
raise Exception(f'POSIX remote completion entry "{self.name}" must provide a "python" setting.')
else:
if not self.version:
raise Exception(f'POSIX remote completion entry "{self.name}" is a platform default and cannot provide a "python" setting.')
@dataclasses.dataclass(frozen=True)
class WindowsRemoteCompletionConfig(RemoteCompletionConfig):
"""Configuration for remote Windows platforms."""
TCompletionConfig = t.TypeVar('TCompletionConfig', bound=CompletionConfig)
def load_completion(name, completion_type): # type: (str, t.Type[TCompletionConfig]) -> t.Dict[str, TCompletionConfig]
"""Load the named completion entries, returning them in dictionary form using the specified completion type."""
lines = read_lines_without_comments(os.path.join(ANSIBLE_TEST_DATA_ROOT, 'completion', '%s.txt' % name), remove_blank_lines=True)
if data_context().content.collection:
context = 'collection'
else:
context = 'ansible-core'
items = {name: data for name, data in [parse_completion_entry(line) for line in lines] if data.get('context', context) == context}
for item in items.values():
item.pop('context', None)
item.pop('placeholder', None)
completion = {name: completion_type(name=name, **data) for name, data in items.items()}
return completion
def parse_completion_entry(value): # type: (str) -> t.Tuple[str, t.Dict[str, str]]
"""Parse the given completion entry, returning the entry name and a dictionary of key/value settings."""
values = value.split()
name = values[0]
data = {kvp[0]: kvp[1] if len(kvp) > 1 else '' for kvp in [item.split('=', 1) for item in values[1:]]}
return name, data
def filter_completion(
completion, # type: t.Dict[str, TCompletionConfig]
controller_only=False, # type: bool
include_defaults=False, # type: bool
): # type: (...) -> t.Dict[str, TCompletionConfig]
"""Return the given completion dictionary, filtering out configs which do not support the controller if controller_only is specified."""
if controller_only:
completion = {name: config for name, config in completion.items() if isinstance(config, PosixCompletionConfig) and config.controller_supported}
if not include_defaults:
completion = {name: config for name, config in completion.items() if not config.is_default}
return completion
@cache
def docker_completion(): # type: () -> t.Dict[str, DockerCompletionConfig]
"""Return docker completion entries."""
return load_completion('docker', DockerCompletionConfig)
@cache
def remote_completion(): # type: () -> t.Dict[str, PosixRemoteCompletionConfig]
"""Return remote completion entries."""
return load_completion('remote', PosixRemoteCompletionConfig)
@cache
def windows_completion(): # type: () -> t.Dict[str, WindowsRemoteCompletionConfig]
"""Return windows completion entries."""
return load_completion('windows', WindowsRemoteCompletionConfig)
@cache
def network_completion(): # type: () -> t.Dict[str, NetworkRemoteCompletionConfig]
"""Return network completion entries."""
return load_completion('network', NetworkRemoteCompletionConfig)
|
