You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
ansible/test/lib/ansible_test/_internal/content_config.py

180 lines
5.6 KiB
Python

"""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)