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
|
"""Content configuration."""
from __future__ import annotations
import os
import pickle
import typing as t
from .constants import (
CONTROLLER_PYTHON_VERSIONS,
SUPPORTED_PYTHON_VERSIONS,
)
from .compat.packaging import (
PACKAGING_IMPORT_ERROR,
SpecifierSet,
Version,
)
from .compat.yaml import (
YAML_IMPORT_ERROR,
yaml_load,
)
from .io import (
open_binary_file,
read_text_file,
)
from .util import (
ApplicationError,
display,
str_to_version,
)
from .data import (
data_context,
)
from .config import (
EnvironmentConfig,
ContentConfig,
ModulesConfig,
)
MISSING = object()
def parse_modules_config(data: t.Any) -> ModulesConfig:
"""Parse the given dictionary as module config and return it."""
if not isinstance(data, dict):
raise Exception('config must be type `dict` not `%s`' % type(data))
python_requires = data.get('python_requires', MISSING)
if python_requires == MISSING:
raise KeyError('python_requires is required')
return ModulesConfig(
python_requires=python_requires,
python_versions=parse_python_requires(python_requires),
controller_only=python_requires == 'controller',
)
def parse_content_config(data: t.Any) -> ContentConfig:
"""Parse the given dictionary as content config and return it."""
if not isinstance(data, dict):
raise Exception('config must be type `dict` not `%s`' % type(data))
# Configuration specific to modules/module_utils.
modules = parse_modules_config(data.get('modules', {}))
# Python versions supported by the controller, combined with Python versions supported by modules/module_utils.
# Mainly used for display purposes and to limit the Python versions used for sanity tests.
python_versions = tuple(version for version in SUPPORTED_PYTHON_VERSIONS
if version in CONTROLLER_PYTHON_VERSIONS or version in modules.python_versions)
# True if Python 2.x is supported.
py2_support = any(version for version in python_versions if str_to_version(version)[0] == 2)
return ContentConfig(
modules=modules,
python_versions=python_versions,
py2_support=py2_support,
)
def load_config(path: str) -> t.Optional[ContentConfig]:
"""Load and parse the specified config file and return the result or None if loading/parsing failed."""
if YAML_IMPORT_ERROR:
raise ApplicationError('The "PyYAML" module is required to parse config: %s' % YAML_IMPORT_ERROR)
if PACKAGING_IMPORT_ERROR:
raise ApplicationError('The "packaging" module is required to parse config: %s' % PACKAGING_IMPORT_ERROR)
value = read_text_file(path)
try:
yaml_value = yaml_load(value)
except Exception as ex: # pylint: disable=broad-except
display.warning('Ignoring config "%s" due to a YAML parsing error: %s' % (path, ex))
return None
try:
config = parse_content_config(yaml_value)
except Exception as ex: # pylint: disable=broad-except
display.warning('Ignoring config "%s" due a config parsing error: %s' % (path, ex))
return None
display.info('Loaded configuration: %s' % path, verbosity=1)
return config
def get_content_config(args: EnvironmentConfig) -> ContentConfig:
"""
Parse and return the content configuration (if any) for the current collection.
For ansible-core, a default configuration is used.
Results are cached.
"""
if args.host_path:
args.content_config = deserialize_content_config(os.path.join(args.host_path, 'config.dat'))
if args.content_config:
return args.content_config
collection_config_path = 'tests/config.yml'
config = None
if data_context().content.collection and os.path.exists(collection_config_path):
config = load_config(collection_config_path)
if not config:
config = parse_content_config(dict(
modules=dict(
python_requires='default',
),
))
if not config.modules.python_versions:
raise ApplicationError('This collection does not declare support for modules/module_utils on any known Python version.\n'
'Ansible supports modules/module_utils on Python versions: %s\n'
'This collection provides the Python requirement: %s' % (
', '.join(SUPPORTED_PYTHON_VERSIONS), config.modules.python_requires))
args.content_config = config
return config
def parse_python_requires(value: t.Any) -> tuple[str, ...]:
"""Parse the given 'python_requires' version specifier and return the matching Python versions."""
if not isinstance(value, str):
raise ValueError('python_requires must must be of type `str` not type `%s`' % type(value))
versions: tuple[str, ...]
if value == 'default':
versions = SUPPORTED_PYTHON_VERSIONS
elif value == 'controller':
versions = CONTROLLER_PYTHON_VERSIONS
else:
specifier_set = SpecifierSet(value)
versions = tuple(version for version in SUPPORTED_PYTHON_VERSIONS if specifier_set.contains(Version(version)))
return versions
def serialize_content_config(args: EnvironmentConfig, path: str) -> None:
"""Serialize the content config to the given path. If the config has not been loaded, an empty config will be serialized."""
with open_binary_file(path, 'wb') as config_file:
pickle.dump(args.content_config, config_file)
def deserialize_content_config(path: str) -> ContentConfig:
"""Deserialize content config from the path."""
with open_binary_file(path) as config_file:
return pickle.load(config_file)
|
