"""Integration test configuration."""
import logging
import re
import shutil
import sys
from pathlib import Path
import simtools.version as simtools_version
from simtools.io import ascii_handler, io_handler
from simtools.utils import value_conversion
_logger = logging.getLogger(__name__)
_PREPARED_RESOURCE_PATTERN = re.compile(r"\$\{prepared:([^}]+)\}")
[docs]
class VersionError(Exception):
"""Raise if model version requested is not supported."""
[docs]
def get_list_of_test_configurations(config_files, test_resources_path=None):
"""
Return list of test configuration dictionaries or test names.
Read all configuration files for testing.
Parameters
----------
config_files: list
List of integration test configuration files.
test_resources_path: str or pathlib.Path, optional
Base directory containing the ``static``, ``generated``, and ``downloaded``
resource directories.
Returns
-------
list:
list of test names or of configuration dictionaries.
"""
_logger.debug(f"Configuration files: {config_files}")
configs = _read_configs_from_files(config_files, test_resources_path=test_resources_path)
return (
configs,
[
f"{item.get('application', 'no-app-name')}_{item.get('test_name', 'no-test-name')}"
for item in configs
],
)
def _read_configs_from_files(config_files, test_resources_path=None):
"""Read test configuration from files."""
configs = []
for config_file in config_files:
# read config file
# remove new line characters from config - otherwise issues
# with especially long file names
_dict = value_conversion.remove_substring_recursively_from_dict(
ascii_handler.collect_data_from_file(file_name=config_file), substring="\n"
)
configs.extend(
resolve_test_resource_paths(
_dict.get("applications", []), test_resources_path=test_resources_path
)
)
return configs
[docs]
def resolve_test_resource_paths(value, test_resources_path=None):
"""Resolve test-resource references recursively."""
return io_handler.resolve_test_resource_paths(value, test_resources_path=test_resources_path)
[docs]
def resolve_prepared_resource_paths(value, prepared_resources_path):
"""Resolve per-test prepared-resource references recursively.
Parameters
----------
value : object
Configuration value, mapping, or sequence to resolve.
prepared_resources_path : str or pathlib.Path
Directory containing files created by integration-test preparation steps.
Returns
-------
object
Configuration with ``${prepared:...}`` references replaced by absolute paths.
Raises
------
ValueError
If a prepared-resource reference escapes its preparation directory.
"""
prepared_resources_path = Path(prepared_resources_path).resolve()
if isinstance(value, dict):
return {
key: resolve_prepared_resource_paths(item, prepared_resources_path)
for key, item in value.items()
}
if isinstance(value, list):
return [resolve_prepared_resource_paths(item, prepared_resources_path) for item in value]
if not isinstance(value, str):
return value
def _resolve(match):
relative_path = Path(match.group(1))
if relative_path.is_absolute() or ".." in relative_path.parts:
raise ValueError(f"Invalid prepared-resource path: {relative_path.as_posix()}")
return str(prepared_resources_path / relative_path)
return _PREPARED_RESOURCE_PATTERN.sub(_resolve, value)
def _copy_resolved_resource_config_files(config, output_path, test_resources_path):
"""Copy resource config files to output path with test-resource paths resolved."""
if test_resources_path is None:
return
test_resources_path = Path(test_resources_path).expanduser().resolve()
config_dir = Path(output_path) / "resolved-resource-configs"
config_dir.mkdir(parents=True, exist_ok=True)
def _resolve(value):
if isinstance(value, dict):
return {key: _resolve(item) for key, item in value.items()}
if isinstance(value, list):
return [_resolve(item) for item in value]
if not isinstance(value, str):
return value
path = Path(value).expanduser()
if (
not path.is_absolute()
or not path.exists()
or path.suffix.lower()
not in (
".json",
".yaml",
".yml",
)
):
return value
try:
path.relative_to(test_resources_path)
except ValueError:
return value
resolved_data = resolve_test_resource_paths(
ascii_handler.collect_data_from_file(path, test_resources_path=test_resources_path),
test_resources_path=test_resources_path,
)
resolved_config_file = config_dir / path.name
ascii_handler.write_data_to_file(
data=resolved_data,
output_file=resolved_config_file,
sort_keys=False,
)
return str(resolved_config_file)
config.update(_resolve(config))
def _skip_test_for_model_version(config, model_version_requested):
"""
Skip test if model version requested is not supported.
Compares full and major.minor version strings.
"""
if not (config.get("model_version_use_current") and model_version_requested):
return
model_version_config = str(config["configuration"]["model_version"])
if (
simtools_version.compare_versions(
model_version_requested,
model_version_config,
level=simtools_version.version_kind(model_version_requested),
)
!= 0
):
raise VersionError(
f"Model version requested {model_version_requested} not supported for this test"
)
def _prepare_test_options(config, output_path, model_version=None, test_resources_path=None):
"""
Prepare test configuration.
This means either to write a temporary config file
or to return a single string for single boolean options (e.g., --version).
Change output path and file to paths provided with output_path.
Parameters
----------
config: dict
Dictionary with the configuration for the application test.
output_path: str
Output path.
model_version: str
Model versions (default: use those given in config files)
Returns
-------
config_file: str
Path to the temporary configuration file.
config_string: str
Command line configuration as single string.
config_file_model_version: str
Configuration file model version
"""
if len(config) == 1 and next(iter(config.values())) is True:
return None, "--" + next(iter(config.keys())).lower(), None
tmp_config_file = output_path / "tmp_config.yml"
config_file_model_version = config.get("model_version")
if isinstance(config_file_model_version, list):
config_file_model_version = [str(item) for item in config_file_model_version]
elif config_file_model_version is not None:
config_file_model_version = str(config_file_model_version)
if model_version and "model_version" in config:
config.update({"model_version": model_version})
_relocate_output_paths(config, output_path)
_copy_resolved_resource_config_files(config, output_path, test_resources_path)
_logger.info(f"Writing config file: {tmp_config_file}")
ascii_handler.write_data_to_file(data=config, output_file=tmp_config_file, sort_keys=False)
return tmp_config_file, None, config_file_model_version
def _relocate_output_paths(config, output_path):
"""Relocate configured output directories and paths beneath them together."""
output_roots = [
Path(config[key]) for key in ("output_path", "grid_output_path") if key in config
]
for key, value in config.items():
if not isinstance(value, str | Path):
continue
path = Path(value)
if path.is_absolute():
continue
if any(path.is_relative_to(root) for root in output_roots):
config[key] = str(Path(output_path) / path)
[docs]
def get_application_command(app, config_file=None, config_string=None):
"""
Return the command to run the application with the given config file.
Parameters
----------
app: str
Name of the application.
config_file: str
Configuration file.
config_string: str
Configuration string (e.g., '--version')
Returns
-------
str: command to run the application test.
"""
if "simtools-" in app:
cmd = app
if shutil.which(app) is None:
module_name = app.removeprefix("simtools-").replace("-", "_")
cmd = f"{sys.executable} src/simtools/applications/{module_name}.py"
else:
cmd = f"{sys.executable} src/simtools/applications/{app}.py"
if config_string:
return f"{cmd} {config_string}"
if config_file is not None:
return f"{cmd} --config {config_file}"
return cmd
[docs]
def create_tmp_output_path(tmp_test_directory, config):
"""
Create temporary output path.
Parameters
----------
tmp_test_directory: str
Temporary directory.
config: dict
Configuration dictionary.
Returns
-------
str: path to the temporary output directory.
"""
try:
tmp_output_path = Path(tmp_test_directory).joinpath(
config["application"] + "-" + config["test_name"]
)
except KeyError as exc:
raise KeyError(f"No application defined in configuration {config}.") from exc
tmp_output_path.mkdir(parents=True, exist_ok=True)
_logger.info(f"Temporary output path: {tmp_output_path}")
return tmp_output_path
