From 4ccf18d9a63fee2f79e8944e4aeeeb060c1f171b Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Mon, 15 Jun 2026 11:08:25 +0200 Subject: [PATCH 01/12] feat(task): add pydantic model validation in config.py --- antarest/core/config.py | 549 +++++++----------- antarest/core/logging/utils.py | 2 +- tests/core/remote/test_remote_executor.py | 4 +- .../launcher_blueprint/test_launcher_local.py | 2 +- tests/launcher/test_local_launcher.py | 6 +- tests/launcher/test_service.py | 5 +- tests/launcher/test_slurm_launcher.py | 2 +- 7 files changed, 211 insertions(+), 359 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 8467eb60f0..6a5cdeeac6 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -15,13 +15,13 @@ import platform import string import tempfile -from dataclasses import asdict, dataclass, field from enum import StrEnum from pathlib import Path -from typing import ClassVar +from typing import Any, ClassVar, Self import yaml from antares.study.version import SolverVersion +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from antarest.core.model import JSON from antarest.core.roles import RoleType @@ -42,86 +42,66 @@ class InternalMatrixFormat(StrEnum): FEATHER = "feather" -@dataclass(frozen=True) -class ExternalAuthConfig: +class ExternalAuthConfig(BaseModel): """ Sub config object dedicated to external auth service """ + model_config = ConfigDict(frozen=True) + url: str | None = None default_group_role: RoleType = RoleType.READER add_ext_groups: bool = False - group_mapping: dict[str, str] = field(default_factory=dict) - - @classmethod - def from_dict(cls, data: JSON) -> "ExternalAuthConfig": - defaults = cls() - return cls( - url=data.get("url", defaults.url), - default_group_role=( - RoleType(data["default_group_role"]) if "default_group_role" in data else defaults.default_group_role - ), - add_ext_groups=data.get("add_ext_groups", defaults.add_ext_groups), - group_mapping=data.get("group_mapping", defaults.group_mapping), - ) + group_mapping: dict[str, str] = Field(default_factory=dict) -@dataclass(frozen=True) -class SecurityConfig: +class SecurityConfig(BaseModel): """ Sub config object dedicated to security """ + model_config = ConfigDict(frozen=True) + jwt_key: str = "" admin_pwd: str = "" disabled: bool = False - external_auth: ExternalAuthConfig = ExternalAuthConfig() + external_auth: ExternalAuthConfig = Field(default_factory=ExternalAuthConfig) + @model_validator(mode="before") @classmethod - def from_dict(cls, data: JSON) -> "SecurityConfig": - defaults = cls() - return cls( - jwt_key=data.get("jwt", {}).get("key", defaults.jwt_key), - admin_pwd=data.get("login", {}).get("admin", {}).get("pwd", defaults.admin_pwd), - disabled=data.get("disabled", defaults.disabled), - external_auth=( - ExternalAuthConfig.from_dict(data["external_auth"]) - if "external_auth" in data - else defaults.external_auth - ), - ) + def _flatten_nested_keys(cls, data: Any) -> Any: + if not isinstance(data, dict): + return data + data = dict(data) + if "jwt" in data: + data.setdefault("jwt_key", data.pop("jwt", {}).get("key", "")) + if "login" in data: + data.setdefault("admin_pwd", data.pop("login", {}).get("admin", {}).get("pwd", "")) + return data -@dataclass(frozen=True) -class WorkspaceConfig: +class WorkspaceConfig(BaseModel): """ Sub config object dedicated to workspace """ - filter_in: list[str] = field(default_factory=lambda: [".*"]) - filter_out: list[str] = field(default_factory=lambda: []) - groups: list[str] = field(default_factory=lambda: []) - path: Path = Path() + model_config = ConfigDict(frozen=True) - @classmethod - def from_dict(cls, data: JSON) -> "WorkspaceConfig": - defaults = cls() - return cls( - filter_in=data.get("filter_in", defaults.filter_in), - filter_out=data.get("filter_out", defaults.filter_out), - groups=data.get("groups", defaults.groups), - path=Path(data["path"]) if "path" in data else defaults.path, - ) + filter_in: list[str] = Field(default_factory=lambda: [".*"]) + filter_out: list[str] = Field(default_factory=list) + groups: list[str] = Field(default_factory=list) + path: Path = Field(default_factory=Path) -@dataclass(frozen=True) -class DbConfig: +class DbConfig(BaseModel): """ Sub config object dedicated to db """ - db_url: str = "" - db_admin_url: str | None = None + model_config = ConfigDict(frozen=True, populate_by_name=True) + + db_url: str = Field(default="", alias="url") + db_admin_url: str | None = Field(default=None, alias="admin_url") db_connect_timeout: int = 10 pool_recycle: int | None = None pool_pre_ping: bool = False @@ -130,62 +110,40 @@ class DbConfig: pool_size: int = 5 pool_use_lifo: bool = False - @classmethod - def from_dict(cls, data: JSON) -> "DbConfig": - defaults = cls() - return cls( - db_admin_url=data.get("admin_url", defaults.db_admin_url), - db_url=data.get("url", defaults.db_url), - db_connect_timeout=data.get("db_connect_timeout", defaults.db_connect_timeout), - pool_recycle=data.get("pool_recycle", defaults.pool_recycle), - pool_pre_ping=data.get("pool_pre_ping", defaults.pool_pre_ping), - pool_use_null=data.get("pool_use_null", defaults.pool_use_null), - pool_max_overflow=data.get("pool_max_overflow", defaults.pool_max_overflow), - pool_size=data.get("pool_size", defaults.pool_size), - pool_use_lifo=data.get("pool_use_lifo", defaults.pool_use_lifo), - ) -@dataclass(frozen=True) -class V2OutputStorageConfig: +class V2OutputStorageConfig(BaseModel): """ Configuration for "new style" internal output storage """ + model_config = ConfigDict(frozen=True) + enable: bool = False archive_dir: Path = Path("./output-archives") variables_dir: Path = Path("./output-variables") - @classmethod - def from_dict(cls, data: JSON) -> "V2OutputStorageConfig": - defaults = cls() - return V2OutputStorageConfig( - enable=data.get("enable", defaults.enable), - archive_dir=Path(data.get("archive_dir", str(defaults.archive_dir))), - variables_dir=Path(data.get("variables_dir", str(defaults.variables_dir))), - ) - -@dataclass(frozen=True) -class OutOfStudyFileOutputStorageConfig: +class OutOfStudyFileOutputStorageConfig(BaseModel): """ Configuration for output storage in a single dir for all studies """ + model_config = ConfigDict(frozen=True) storage_dir: Path = Path("./outputs") @classmethod - def from_dict(cls, data: JSON) -> "OutOfStudyFileOutputStorageConfig": - defaults = cls() - storage_dir = Path(data.get("storage_dir", str(defaults.storage_dir))) - storage_dir.mkdir(parents=True, exist_ok=True) - return OutOfStudyFileOutputStorageConfig(storage_dir=storage_dir) + @model_validator(mode="after") + def validate_range(self) -> type[Self]: + self.storage_dir.mkdir(parents=True, exist_ok=True) + return self -@dataclass(frozen=True) -class OutputStorageConfig: +class OutputStorageConfig(BaseModel): """Configuration for output storage""" + model_config = ConfigDict(frozen=True) + v2: V2OutputStorageConfig = V2OutputStorageConfig() out_of_study: OutOfStudyFileOutputStorageConfig = OutOfStudyFileOutputStorageConfig() @@ -195,8 +153,8 @@ class OutputStorageConfig: def from_dict(cls, data: JSON) -> "OutputStorageConfig": defaults = cls() config = cls( - v2=V2OutputStorageConfig.from_dict(data.get("v2", {})), - out_of_study=OutOfStudyFileOutputStorageConfig.from_dict(data.get("out_of_study", {})), + v2=V2OutputStorageConfig.model_validate(data.get("v2", {})), + out_of_study=OutOfStudyFileOutputStorageConfig.model_validate(data.get("out_of_study", {})), default_storage_type=OutputStorageType(data.get("default_storage_type", defaults.default_storage_type)), ) if config.default_storage_type == OutputStorageType.V2 and not config.v2.enable: @@ -204,33 +162,28 @@ def from_dict(cls, data: JSON) -> "OutputStorageConfig": return config -@dataclass(frozen=True) -class StudyStorageConfig: +class StudyStorageConfig(BaseModel): """ Sub config object dedicated to study storage configuration (from study.storage in YAML) """ - database_mode_enabled: bool = False + model_config = ConfigDict(frozen=True) - @classmethod - def from_dict(cls, data: JSON) -> "StudyStorageConfig": - defaults = cls() - return cls( - database_mode_enabled=data.get("database_mode_enabled", defaults.database_mode_enabled), - ) + database_mode_enabled: bool = False -@dataclass(frozen=True) -class StorageConfig: +class StorageConfig(BaseModel): """ Sub config object dedicated to study module """ + model_config = ConfigDict(frozen=True) + matrixstore: Path = Path("./matrixstore") archive_dir: Path = Path("./archives") archive_format: ArchiveFormat = ArchiveFormat.ZIP tmp_dir: Path = Path(tempfile.gettempdir()) - workspaces: dict[str, WorkspaceConfig] = field(default_factory=dict) + workspaces: dict[str, WorkspaceConfig] = Field(default_factory=dict) allow_deletion: bool = False watcher_lock: bool = True watcher_lock_delay: int = 10 @@ -259,64 +212,24 @@ class StorageConfig: disk_usage_log_cron: str = "0 * * * *" disk_space_analyzer_sleeping_time: int = 300 disk_space_analyzer_cron: str = "0 1 * * *" - study_storage: StudyStorageConfig = StudyStorageConfig() - output: OutputStorageConfig = OutputStorageConfig() + study_storage: StudyStorageConfig = Field(default_factory=StudyStorageConfig) + output: OutputStorageConfig = Field(default_factory=OutputStorageConfig) @classmethod def from_dict(cls, data: JSON, desktop_mode: bool = False) -> "StorageConfig": if data.get("auto_archive_sleeping_time") and data.get("auto_archive_cron"): raise ValueError("auto_archive_sleeping_time and auto_archive_cron cannot be used together") - defaults = cls() - workspaces = ( - {key: WorkspaceConfig.from_dict(value) for key, value in data["workspaces"].items()} - if "workspaces" in data - else defaults.workspaces - ) + processed = dict(data) + # Flatten study.storage → study_storage + if "study" in processed and isinstance(processed["study"], dict): + processed.setdefault("study_storage", processed.pop("study").get("storage", {})) + # Validate workspaces and inject desktop system workspaces + workspaces = {k: WorkspaceConfig.model_validate(v) for k, v in processed.pop("workspaces", {}).items()} cls.validate_workspaces(workspaces, desktop_mode) if desktop_mode: workspaces = {**workspaces, **cls.system_workspaces()} - return cls( - matrixstore=Path(data["matrixstore"]) if "matrixstore" in data else defaults.matrixstore, - archive_dir=Path(data["archive_dir"]) if "archive_dir" in data else defaults.archive_dir, - archive_format=( - ArchiveFormat(data["archive_format"]) if "archive_format" in data else defaults.archive_format - ), - tmp_dir=Path(data["tmp_dir"]) if "tmp_dir" in data else defaults.tmp_dir, - workspaces=workspaces, - allow_deletion=data.get("allow_deletion", defaults.allow_deletion), - watcher_lock=data.get("watcher_lock", defaults.watcher_lock), - watcher_lock_delay=data.get("watcher_lock_delay", defaults.watcher_lock_delay), - download_default_expiration_timeout_minutes=( - data.get( - "download_default_expiration_timeout_minutes", - defaults.download_default_expiration_timeout_minutes, - ) - ), - matrix_gc_sleeping_time=data.get("matrix_gc_sleeping_time", defaults.matrix_gc_sleeping_time), - matrix_gc_dry_run=data.get("matrix_gc_dry_run", defaults.matrix_gc_dry_run), - matrix_gc_retention_time=data.get("matrix_gc_retention_time", defaults.matrix_gc_retention_time), - auto_archive_threshold_days=data.get("auto_archive_threshold_days", defaults.auto_archive_threshold_days), - auto_archive_dry_run=data.get("auto_archive_dry_run", defaults.auto_archive_dry_run), - auto_archive_sleeping_time=data.get("auto_archive_sleeping_time", defaults.auto_archive_sleeping_time), - auto_archive_cron=data.get("auto_archive_cron", defaults.auto_archive_cron), - snapshot_retention_days=data.get("snapshot_retention_days", defaults.snapshot_retention_days), - matrixstore_format=InternalMatrixFormat(data.get("matrixstore_format", defaults.matrixstore_format)), - blobstore=Path(data["blobstore"]) if "blobstore" in data else defaults.blobstore, - blob_gc_sleeping_time=data.get("blob_gc_sleeping_time", defaults.blob_gc_sleeping_time), - blob_gc_dry_run=data.get("blob_gc_dry_run", defaults.blob_gc_dry_run), - variable_view_gc_sleeping_time=data.get( - "variable_view_gc_sleeping_time", defaults.variable_view_gc_sleeping_time - ), - variable_view_gc_dry_run=data.get("variable_view_gc_dry_run", defaults.variable_view_gc_dry_run), - variable_view_gc_retention_days=data.get( - "variable_view_gc_retention_days", defaults.variable_view_gc_retention_days - ), - watcher_scan_sleeping_time=data.get("watcher_scan_sleeping_time", defaults.watcher_scan_sleeping_time), - watcher_scan_dry_run=data.get("watcher_scan_dry_run", defaults.watcher_scan_dry_run), - tasks_gc_retention_days=data.get("tasks_gc_retention_days", defaults.tasks_gc_retention_days), - study_storage=StudyStorageConfig.from_dict(data.get("study", {}).get("storage", {})), - output=OutputStorageConfig.from_dict(data["output"]) if "output" in data else defaults.output, - ) + processed["workspaces"] = workspaces + return cls.model_validate(processed) @classmethod def validate_workspaces(cls, workspaces: dict[str, WorkspaceConfig], desktop_mode: bool) -> None: @@ -350,12 +263,13 @@ def system_workspaces(cls) -> dict[str, WorkspaceConfig]: raise NotImplementedError("System workspaces are only implemented for Linux and Windows") -@dataclass(frozen=True) -class NbCoresConfig: +class NbCoresConfig(BaseModel): """ The NBCoresConfig class is designed to manage the configuration of the number of CPU cores """ + model_config = ConfigDict(frozen=True) + min: int = 1 default: int = 22 max: int = 24 @@ -371,16 +285,16 @@ def to_json(self) -> dict[str, int]: """ return {"min": self.min, "defaultValue": self.default, "max": self.max} - def __post_init__(self) -> None: + @model_validator(mode="after") + def _validate_range(self) -> "NbCoresConfig": """validation of CPU configuration""" if 1 <= self.min <= self.default <= self.max: - return + return self msg = f"Invalid configuration: 1 <= {self.min=} <= {self.default=} <= {self.max=}" raise ValueError(msg) -@dataclass(frozen=True) -class TimeLimitConfig: +class TimeLimitConfig(BaseModel): """ The TimeLimitConfig class is designed to manage the configuration of the time limit for a job. @@ -390,6 +304,8 @@ class TimeLimitConfig: max: int: maximum allowed value for the time limit (in hours). """ + model_config = ConfigDict(frozen=True) + min: int = 1 default: int = 48 max: int = 48 @@ -405,52 +321,48 @@ def to_json(self) -> dict[str, int]: """ return {"min": self.min, "defaultValue": self.default, "max": self.max} - def __post_init__(self) -> None: - """validation of CPU configuration""" + @model_validator(mode="after") + def _validate_range(self) -> "TimeLimitConfig": + """validation of time limit configuration""" if 1 <= self.min <= self.default <= self.max: - return + return self msg = f"Invalid configuration: 1 <= {self.min=} <= {self.default=} <= {self.max=}" raise ValueError(msg) -@dataclass(frozen=True) -class LocalConfig: +class LocalConfig(BaseModel): """Sub config object dedicated to launcher module (local)""" + model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) + id: str name: str type: ClassVar[LauncherType] = LauncherType.LOCAL - binaries: dict[SolverVersion, Path] = field(default_factory=dict) + binaries: dict[SolverVersion, Path] = Field(default_factory=dict) enable_nb_cores_detection: bool = True - nb_cores: NbCoresConfig = NbCoresConfig() - time_limit: TimeLimitConfig = TimeLimitConfig() + nb_cores: NbCoresConfig = Field(default_factory=NbCoresConfig) + time_limit: TimeLimitConfig = Field(default_factory=TimeLimitConfig) xpress_dir: str | None = None local_workspace: Path = Path("./local_workspace") + @field_validator("binaries", mode="before") @classmethod - def from_dict(cls, data: JSON) -> "LocalConfig": - """ - Creates an instance of LocalConfig from a data dictionary - Args: - data: Parse config from dict. - Returns: object NbCoresConfig - """ - binaries = {SolverVersion.parse(str(k)): Path(v) for k, v in data.get("binaries", {}).items()} - enable_nb_cores_detection = data.get("enable_nb_cores_detection", True) - nb_cores = data.get("nb_cores", asdict(NbCoresConfig())) - if enable_nb_cores_detection: - nb_cores.update(cls._autodetect_nb_cores()) - xpress_dir = data.get("xpress_dir") - local_workspace = Path(data["local_workspace"]) if "local_workspace" in data else Path("./local_workspace") - return cls( - id=data["id"], - name=data["name"], - binaries=binaries, - enable_nb_cores_detection=enable_nb_cores_detection, - nb_cores=NbCoresConfig(**nb_cores), - xpress_dir=xpress_dir, - local_workspace=local_workspace, - ) + def _parse_binaries(cls, v: Any) -> Any: + """Coerce string keys/values to SolverVersion/Path.""" + if isinstance(v, dict): + return { + SolverVersion.parse(str(k)) if not isinstance(k, SolverVersion) else k: Path(val) + for k, val in v.items() + } + return v + + @model_validator(mode="wrap") + @classmethod + def apply_cpu_detection(cls, data: Any, handler: Any) -> "LocalConfig": + """Inject detected CPU core counts into data before construction.""" + if isinstance(data, dict) and data.get("enable_nb_cores_detection", True): + data = {**data, "nb_cores": cls._autodetect_nb_cores()} + return handler(data) @classmethod def _autodetect_nb_cores(cls) -> dict[str, int]: @@ -458,85 +370,64 @@ def _autodetect_nb_cores(cls) -> dict[str, int]: Automatically detects the number of cores available on the user's machine Returns: Instance of NbCoresConfig """ - min_cpu = cls.nb_cores.min + min_cpu = NbCoresConfig().min max_cpu = multiprocessing.cpu_count() default = max(min_cpu, max_cpu - 2) return {"min": min_cpu, "max": max_cpu, "default": default} -@dataclass(frozen=True) -class SlurmConfig: +class SlurmConfig(BaseModel): """ Sub config object dedicated to launcher module (slurm) """ + model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) + id: str name: str type: ClassVar[LauncherType] = LauncherType.SLURM - local_workspace: Path = Path() + local_workspace: Path = Field(default_factory=Path) username: str = "" hostname: str = "" port: int = 0 - private_key_file: Path = Path() + private_key_file: Path = Field(default_factory=Path) key_password: str = "" password: str = "" default_wait_time: int = 0 - time_limit: TimeLimitConfig = TimeLimitConfig() - nb_cores: NbCoresConfig = NbCoresConfig() + time_limit: TimeLimitConfig = Field(default_factory=TimeLimitConfig) + nb_cores: NbCoresConfig = Field(default_factory=NbCoresConfig) default_json_db_name: str = "" slurm_script_path: str = "" partition: str = "" max_cores: int = 64 - antares_versions_on_remote_server: list[SolverVersion] = field(default_factory=list) + antares_versions_on_remote_server: list[SolverVersion] = Field(default_factory=list) enable_nb_cores_detection: bool = False + @field_validator("antares_versions_on_remote_server", mode="before") @classmethod - def from_dict(cls, data: JSON) -> "SlurmConfig": - """ - Creates an instance of SlurmConfig from a data dictionary - - Args: - data: Parsed config from dict. - Returns: object SlurmConfig - """ - enable_nb_cores_detection = data.get("enable_nb_cores_detection", False) - nb_cores = data.get("nb_cores", asdict(NbCoresConfig())) - if "default_n_cpu" in data: - # Use the old way to configure the NB cores for backward compatibility - nb_cores["default"] = int(data["default_n_cpu"]) - nb_cores["min"] = min(nb_cores["min"], nb_cores["default"]) - nb_cores["max"] = max(nb_cores["max"], nb_cores["default"]) - if enable_nb_cores_detection: - nb_cores.update(cls._autodetect_nb_cores()) - # In the configuration file, the default time limit is in seconds, so we convert it to hours - max_time_limit = data.get("default_time_limit", TimeLimitConfig().max * 3600) // 3600 - time_limit = TimeLimitConfig(min=1, default=max_time_limit, max=max_time_limit) - antares_versions_as_str = data.get("antares_versions_on_remote_server", []) - antares_versions_on_remote_server = [SolverVersion.parse(str(v)) for v in antares_versions_as_str] - return cls( - id=data["id"], - name=data["name"], - local_workspace=Path(data.get("local_workspace", Path())), - username=data.get("username", ""), - hostname=data.get("hostname", ""), - port=data.get("port", 0), - private_key_file=data.get("private_key_file", Path()), - key_password=data.get("key_password", ""), - password=data.get("password", ""), - default_wait_time=data.get("default_wait_time", 0), - time_limit=time_limit, - default_json_db_name=data.get("default_json_db_name", ""), - slurm_script_path=data.get("slurm_script_path", ""), - partition=data.get("partition", ""), - antares_versions_on_remote_server=antares_versions_on_remote_server, - max_cores=data.get("max_cores", 64), - enable_nb_cores_detection=enable_nb_cores_detection, - nb_cores=NbCoresConfig(**nb_cores), - ) + def _parse_versions(cls, v: Any) -> Any: + """Coerce string items to SolverVersion.""" + if isinstance(v, list): + return [SolverVersion.parse(str(item)) if not isinstance(item, SolverVersion) else item for item in v] + return v + @model_validator(mode="before") @classmethod - def _autodetect_nb_cores(cls) -> dict[str, int]: - raise NotImplementedError("NB Cores auto-detection is not implemented for SLURM server") + def handle_backward_compatibility(cls, data: Any) -> Any: + """Handle backward compatibility for old config formats.""" + if not isinstance(data, dict): + return data + + data = dict(data) # Make a copy + # Handle old default_n_cpu field -> nb_cores + if "default_n_cpu" in data: + nb_cores = data.get("nb_cores", NbCoresConfig().model_dump()) + default_n_cpu = int(data.pop("default_n_cpu")) + nb_cores["default"] = default_n_cpu + nb_cores["min"] = min(nb_cores["min"], default_n_cpu) + nb_cores["max"] = max(nb_cores["max"], default_n_cpu) + data["nb_cores"] = nb_cores + return data class InvalidConfigurationError(Exception): @@ -550,140 +441,112 @@ def __init__(self, launcher: str): super().__init__(msg) -@dataclass(frozen=True) -class LauncherConfig: +class LauncherConfig(BaseModel): """ Sub config object dedicated to launcher module """ - default: str = "local" + model_config = ConfigDict(frozen=True, populate_by_name=True) + + default: str = Field(default="local") configs: list[LocalConfig | SlurmConfig] | None = None batch_size: int = 9999 + @model_validator(mode="before") @classmethod - def from_dict(cls, data: JSON) -> "LauncherConfig": - launchers: list[LocalConfig | SlurmConfig] = [] - defaults = cls() - default = data.get("default", cls.default) - batch_size = data.get("batch_size", defaults.batch_size) - for launcher in data["launchers"]: - match launcher["type"]: - case LauncherType.LOCAL: - launchers.append(LocalConfig.from_dict(launcher)) - case LauncherType.SLURM: - launchers.append(SlurmConfig.from_dict(launcher)) - case _: - raise InvalidConfigurationError(f"Unknown launcher type: {launcher['type']}") - - if not any(launcher.id == default for launcher in launchers): - raise InvalidConfigurationError(f"Default launcher id '{default}' not found in launcher configs") - - return cls( - default=default, - configs=launchers, - batch_size=batch_size, - ) + def _remap_launchers(cls, data: Any) -> Any: + """Remap the 'launchers' YAML key to 'configs', dispatching each entry by type.""" + if not isinstance(data, dict) or "launchers" not in data: + return data + data = dict(data) + raw_launchers = data.pop("launchers") + configs: list[Any] = [] + for launcher in raw_launchers: + launcher_type = launcher.get("type") + if launcher_type == LauncherType.LOCAL: + configs.append(LocalConfig.model_validate(launcher)) + elif launcher_type == LauncherType.SLURM: + configs.append(SlurmConfig.model_validate(launcher)) + else: + raise ValueError(f"Unknown launcher type: {launcher_type!r}") + data["configs"] = configs + return data + + @model_validator(mode="after") + def _validate_default_launcher(self) -> "LauncherConfig": + if self.configs: + if not any(launcher.id == self.default for launcher in self.configs): + ids = [launcher.id for launcher in self.configs] + raise ValueError(f"Default launcher id '{self.default}' not found in launcher configs, got {ids}") + return self def get_slurm_configs(self) -> list[SlurmConfig]: return [cfg for cfg in self.configs or [] if isinstance(cfg, SlurmConfig)] -@dataclass(frozen=True) -class LoggingConfig: +class LoggingConfig(BaseModel): """ Sub config object dedicated to logging """ + model_config = ConfigDict(frozen=True, populate_by_name=True) + logfile: Path | None = None - json: bool = False + json_format: bool = Field(default=False, alias="json") level: str = "INFO" - @classmethod - def from_dict(cls, data: JSON) -> "LoggingConfig": - defaults = cls() - return cls( - logfile=Path(data["logfile"]) if "logfile" in data else defaults.logfile, - json=data.get("json", defaults.json), - level=data.get("level", defaults.level), - ) - -@dataclass(frozen=True) -class RedisConfig: +class RedisConfig(BaseModel): """ Sub config object dedicated to redis """ + model_config = ConfigDict(frozen=True) + host: str = "localhost" port: int = 6379 password: str | None = None - @classmethod - def from_dict(cls, data: JSON) -> "RedisConfig": - defaults = cls() - return cls( - host=data.get("host", defaults.host), - port=data.get("port", defaults.port), - password=data.get("password", defaults.password), - ) - -@dataclass(frozen=True) -class EventBusConfig: +class EventBusConfig(BaseModel): """ Sub config object dedicated to eventbus module """ - # noinspection PyUnusedLocal - @classmethod - def from_dict(cls, data: JSON) -> "EventBusConfig": - return cls() + model_config = ConfigDict(frozen=True) -@dataclass(frozen=True) -class CacheConfig: +class CacheConfig(BaseModel): """ Sub config object dedicated to cache module """ + model_config = ConfigDict(frozen=True) checker_delay: float = 0.2 # in seconds - @classmethod - def from_dict(cls, data: JSON) -> "CacheConfig": - defaults = cls() - return cls( - checker_delay=data.get("checker_delay", defaults.checker_delay), - ) +class RemoteWorkerConfig(BaseModel): + model_config = ConfigDict(frozen=True) -@dataclass(frozen=True) -class RemoteWorkerConfig: name: str - queues: list[str] = field(default_factory=list) - - @classmethod - def from_dict(cls, data: JSON) -> "RemoteWorkerConfig": - defaults = cls(name="") # `name` is mandatory - return cls( - name=data["name"], - queues=data.get("queues", defaults.queues), - ) + queues: list[str] = Field(default_factory=list) -@dataclass(frozen=True) -class TaskConfig: +class TaskConfig(BaseModel): """ Sub config object dedicated to the task module """ + model_config = ConfigDict(frozen=True) + max_workers: int = 5 - remote_workers: list[RemoteWorkerConfig] = field(default_factory=list) + remote_workers: list[RemoteWorkerConfig] = Field(default_factory=list) @classmethod def from_dict(cls, data: JSON) -> "TaskConfig": defaults = cls() remote_workers = ( - [RemoteWorkerConfig.from_dict(d) for d in data["remote_workers"]] + [RemoteWorkerConfig.model_validate(d) for d in data["remote_workers"]] if "remote_workers" in data else defaults.remote_workers ) @@ -693,26 +556,19 @@ def from_dict(cls, data: JSON) -> "TaskConfig": ) -@dataclass(frozen=True) -class ServerConfig: +class ServerConfig(BaseModel): """ Sub config object dedicated to the server """ + model_config = ConfigDict(frozen=True) + worker_threadpool_size: int = 5 - services: list[str] = field(default_factory=list) + services: list[str] = Field(default_factory=list) - @classmethod - def from_dict(cls, data: JSON) -> "ServerConfig": - defaults = cls() - return cls( - worker_threadpool_size=data.get("worker_threadpool_size", defaults.worker_threadpool_size), - services=data.get("services", defaults.services), - ) -@dataclass(frozen=True) -class PrometheusConfig: +class PrometheusConfig(BaseModel): """ Sub config object dedicated to prometheus metrics @@ -720,16 +576,12 @@ class PrometheusConfig: multiprocess: if True, metrics of workers will be aggregated before exposition. Environment variable `PROMETHEUS_MULTIPROC_DIR` must be set. """ + model_config = ConfigDict(frozen=True) multiprocess: bool = False - @classmethod - def from_dict(cls, data: JSON) -> "PrometheusConfig": - return cls(multiprocess=bool(data["multiprocess"])) - -@dataclass(frozen=True) -class MetricsConfig: +class MetricsConfig(BaseModel): """ Sub config object dedicated to metrics @@ -737,15 +589,13 @@ class MetricsConfig: prometheus: if not None, metrics will be exposed in prometheus format """ + model_config = ConfigDict(frozen=True) + prometheus: PrometheusConfig | None = None - @classmethod - def from_dict(cls, data: JSON) -> "MetricsConfig": - return cls(prometheus=PrometheusConfig.from_dict(data["prometheus"]) if "prometheus" in data else None) -@dataclass(frozen=True) -class CeleryConfig: +class CeleryConfig(BaseModel): """ Sub config object dedicated to Celery technical configuration. @@ -755,6 +605,8 @@ class CeleryConfig: result_expires: Time in seconds before task results expire """ + model_config = ConfigDict(frozen=True) + # Redis database number for Celery (broker + results) REDIS_DB: ClassVar[int] = 1 @@ -783,8 +635,7 @@ def from_dict(cls, data: JSON, redis_config: RedisConfig | None = None) -> "Cele ) -@dataclass(frozen=True) -class Config: +class Config(BaseModel): """ Root server config """ @@ -816,25 +667,25 @@ def from_dict(cls, data: JSON) -> "Config": if "storage" in data else defaults.storage ) - redis_config = RedisConfig.from_dict(data["redis"]) if "redis" in data else defaults.redis + redis_config = RedisConfig.model_validate(data["redis"]) if "redis" in data else defaults.redis return cls( - server=ServerConfig.from_dict(data["server"]) if "server" in data else defaults.server, - security=SecurityConfig.from_dict(data["security"]) if "security" in data else defaults.security, + server=ServerConfig.model_validate(data["server"]) if "server" in data else defaults.server, + security=SecurityConfig.model_validate(data["security"]) if "security" in data else defaults.security, storage=storage_config, - launcher=LauncherConfig.from_dict(data["launcher"]) if "launcher" in data else defaults.launcher, - db=DbConfig.from_dict(data["db"]) if "db" in data else defaults.db, - logging=LoggingConfig.from_dict(data["logging"]) if "logging" in data else defaults.logging, + launcher=LauncherConfig.model_validate(data["launcher"]) if "launcher" in data else defaults.launcher, + db=DbConfig.model_validate(data["db"]) if "db" in data else defaults.db, + logging=LoggingConfig.model_validate(data["logging"]) if "logging" in data else defaults.logging, debug=data.get("debug", defaults.debug), resources_path=data["resources_path"] if "resources_path" in data else defaults.resources_path, redis=redis_config, - eventbus=EventBusConfig.from_dict(data["eventbus"]) if "eventbus" in data else defaults.eventbus, - cache=CacheConfig.from_dict(data["cache"]) if "cache" in data else defaults.cache, + eventbus=EventBusConfig.model_validate(data["eventbus"]) if "eventbus" in data else defaults.eventbus, + cache=CacheConfig.model_validate(data["cache"]) if "cache" in data else defaults.cache, tasks=TaskConfig.from_dict(data["tasks"]) if "tasks" in data else defaults.tasks, celery=CeleryConfig.from_dict(data.get("celery", {}), redis_config=redis_config), root_path=data.get("root_path", defaults.root_path), api_prefix=data.get("api_prefix", defaults.api_prefix), desktop_mode=desktop_mode, - metrics=MetricsConfig.from_dict(data["metrics"]) if "metrics" in data else MetricsConfig(), + metrics=MetricsConfig.model_validate(data["metrics"]) if "metrics" in data else MetricsConfig(), ) @classmethod diff --git a/antarest/core/logging/utils.py b/antarest/core/logging/utils.py index 5399f84421..6a30dff239 100644 --- a/antarest/core/logging/utils.py +++ b/antarest/core/logging/utils.py @@ -157,7 +157,7 @@ def configure_logger(config: Config, handler_cls: str = "logging.FileHandler") - "DEBUG", ]: logging_config["loggers"][""]["level"] = config.logging.level - if config.logging.json: + if config.logging.json_format: logging_config["handlers"]["default"]["formatter"] = "json" logging.config.dictConfig(logging_config) diff --git a/tests/core/remote/test_remote_executor.py b/tests/core/remote/test_remote_executor.py index d31ffe5442..afb4ea888f 100644 --- a/tests/core/remote/test_remote_executor.py +++ b/tests/core/remote/test_remote_executor.py @@ -26,7 +26,7 @@ def test_remote_executor(): config = Config() - config.tasks.remote_workers.append(RemoteWorkerConfig("worker", queues=["q1", "q2"])) + config.tasks.remote_workers.append(RemoteWorkerConfig(name="worker", queues=["q1", "q2"])) event_bus = EventBusService(LocalEventBus()) executor = RemoteWorkerExecutor(event_bus=event_bus, config=config) @@ -63,7 +63,7 @@ def start_task(): def test_remote_executor_should_reject_unknown_queue(): config = Config() - config.tasks.remote_workers.append(RemoteWorkerConfig("worker", queues=["q1", "q2"])) + config.tasks.remote_workers.append(RemoteWorkerConfig(name="worker", queues=["q1", "q2"])) executor = RemoteWorkerExecutor(event_bus=Mock(), config=config) diff --git a/tests/integration/launcher_blueprint/test_launcher_local.py b/tests/integration/launcher_blueprint/test_launcher_local.py index ef43f5f0af..c8c6f8e34c 100644 --- a/tests/integration/launcher_blueprint/test_launcher_local.py +++ b/tests/integration/launcher_blueprint/test_launcher_local.py @@ -27,7 +27,7 @@ def test_get_launchers( user_access_token: str, ) -> None: # NOTE: we have `enable_nb_cores_detection: True` in `tests/integration/assets/config.template.yml`. - local_config = LocalConfig.from_dict( + local_config = LocalConfig.model_validate( {"id": "local_id", "type": "local", "name": "local", "enable_nb_cores_detection": True} ) expected_launchers = [ diff --git a/tests/launcher/test_local_launcher.py b/tests/launcher/test_local_launcher.py index 0491bcbfd7..dcc6cb28b0 100644 --- a/tests/launcher/test_local_launcher.py +++ b/tests/launcher/test_local_launcher.py @@ -44,7 +44,7 @@ def launcher_config(tmp_path: Path) -> LocalConfig: "enable_nb_cores_detection": True, "local_workspace": tmp_path, } - return LocalConfig.from_dict(data) + return LocalConfig.model_validate(data) def test_compute(tmp_path: Path, launcher_config: LocalConfig) -> None: @@ -222,7 +222,7 @@ def test_unsupported_launcher_other_options_should_raise( def test_parse_xpress_dir(tmp_path: Path) -> None: data = {"id": "id", "name": "name", "type": "local", "xpress_dir": "fake_path_for_test"} - launcher_config = LocalConfig.from_dict(data) + launcher_config = LocalConfig.model_validate(data) local_launcher = LocalLauncher(launcher_config, callbacks=Mock(), event_bus=Mock(), cache=Mock()) launch_parameters = LauncherParametersDTO(other_options="xpress") _, env_variables = local_launcher._parse_launcher_options(launch_parameters, SolverVersion.parse("9.2")) @@ -267,7 +267,7 @@ def test_select_best_binary() -> None: SolverVersion.parse("1000"): Path("1000"), } local_launcher = LocalLauncher( - LocalConfig.from_dict({"id": "id", "name": "name", "type": "local", "binaries": binaries}), + LocalConfig.model_validate({"id": "id", "name": "name", "type": "local", "binaries": binaries}), callbacks=Mock(), event_bus=Mock(), cache=Mock(), diff --git a/tests/launcher/test_service.py b/tests/launcher/test_service.py index dd30814639..956e087930 100644 --- a/tests/launcher/test_service.py +++ b/tests/launcher/test_service.py @@ -22,6 +22,7 @@ from zipfile import ZIP_DEFLATED, ZipFile import pytest +from pydantic import ValidationError from sqlalchemy import create_engine from antarest.core.config import ( @@ -517,7 +518,7 @@ def test_service_get_jobs_from_database( id="default-config-not-found", marks=pytest.mark.xfail( reason="Default launcher id default not found in launcher configs", - raises=InvalidConfigurationError, + raises=ValidationError, strict=True, ), ), @@ -951,7 +952,7 @@ def test_save_solver_stats(self, tmp_path: Path) -> None: id="slurm launcher with no config", marks=pytest.mark.xfail( reason="Configuration is not available for the slurm launcher", - raises=InvalidConfigurationError, + raises=ValidationError, strict=True, ), ), diff --git a/tests/launcher/test_slurm_launcher.py b/tests/launcher/test_slurm_launcher.py index 3cca4f7d1c..6b0d7d2c7c 100644 --- a/tests/launcher/test_slurm_launcher.py +++ b/tests/launcher/test_slurm_launcher.py @@ -62,7 +62,7 @@ def launcher_config(tmp_path: Path) -> SlurmConfig: "enable_nb_cores_detection": False, "nb_cores": {"min": 1, "default": 34, "max": 36}, } - return SlurmConfig.from_dict(data) + return SlurmConfig.model_validate(data) def test_init_slurm_launcher_arguments(tmp_path: Path) -> None: From 91b88674537f097b46e19542645c59d510b45f78 Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Mon, 15 Jun 2026 14:17:31 +0200 Subject: [PATCH 02/12] feat(task): add pydantic model validation in StorageConfig --- antarest/core/config.py | 103 ++++++++++++++++++++--------------- tests/storage/test_config.py | 8 +-- 2 files changed, 63 insertions(+), 48 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 6a5cdeeac6..203d920263 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -215,21 +215,30 @@ class StorageConfig(BaseModel): study_storage: StudyStorageConfig = Field(default_factory=StudyStorageConfig) output: OutputStorageConfig = Field(default_factory=OutputStorageConfig) + @field_validator("workspaces", mode="before") @classmethod - def from_dict(cls, data: JSON, desktop_mode: bool = False) -> "StorageConfig": + def convert_workspaces(cls, v: Any) -> dict[str, WorkspaceConfig]: + """Convert dict entries to WorkspaceConfig objects.""" + if isinstance(v, dict): + return { + key: WorkspaceConfig.model_validate(val) if isinstance(val, dict) else val for key, val in v.items() + } + return v + + @model_validator(mode="before") + @classmethod + def _preprocess(cls, data: Any) -> Any: + """Flatten nested YAML keys and validate mutual exclusions.""" + if not isinstance(data, dict): + return data + data = dict(data) + # Flatten study.storage → study_storage + if "study" in data and isinstance(data["study"], dict): + data.setdefault("study_storage", data.pop("study").get("storage", {})) + # Mutual exclusion check if data.get("auto_archive_sleeping_time") and data.get("auto_archive_cron"): raise ValueError("auto_archive_sleeping_time and auto_archive_cron cannot be used together") - processed = dict(data) - # Flatten study.storage → study_storage - if "study" in processed and isinstance(processed["study"], dict): - processed.setdefault("study_storage", processed.pop("study").get("storage", {})) - # Validate workspaces and inject desktop system workspaces - workspaces = {k: WorkspaceConfig.model_validate(v) for k, v in processed.pop("workspaces", {}).items()} - cls.validate_workspaces(workspaces, desktop_mode) - if desktop_mode: - workspaces = {**workspaces, **cls.system_workspaces()} - processed["workspaces"] = workspaces - return cls.model_validate(processed) + return data @classmethod def validate_workspaces(cls, workspaces: dict[str, WorkspaceConfig], desktop_mode: bool) -> None: @@ -640,50 +649,56 @@ class Config(BaseModel): Root server config """ - server: ServerConfig = ServerConfig() - security: SecurityConfig = SecurityConfig() - storage: StorageConfig = StorageConfig() - launcher: LauncherConfig = LauncherConfig() - db: DbConfig = DbConfig() - logging: LoggingConfig = LoggingConfig() + model_config = ConfigDict(frozen=True) + + server: ServerConfig = Field(default_factory=ServerConfig) + security: SecurityConfig = Field(default_factory=SecurityConfig) + storage: StorageConfig = Field(default_factory=StorageConfig) + launcher: LauncherConfig = Field(default_factory=LauncherConfig) + db: DbConfig = Field(default_factory=DbConfig) + logging: LoggingConfig = Field(default_factory=LoggingConfig) debug: bool = True - resources_path: Path = Path() + resources_path: Path = Field(default_factory=Path) redis: RedisConfig | None = None - eventbus: EventBusConfig = EventBusConfig() - cache: CacheConfig = CacheConfig() - tasks: TaskConfig = TaskConfig() - metrics: MetricsConfig = MetricsConfig() - celery: CeleryConfig = CeleryConfig() + eventbus: EventBusConfig = Field(default_factory=EventBusConfig) + cache: CacheConfig = Field(default_factory=CacheConfig) + tasks: TaskConfig = Field(default_factory=TaskConfig) + metrics: MetricsConfig = Field(default_factory=MetricsConfig) + celery: CeleryConfig = Field(default_factory=CeleryConfig) root_path: str = "" api_prefix: str = "" desktop_mode: bool = False @classmethod def from_dict(cls, data: JSON) -> "Config": - defaults = cls() - desktop_mode = data.get("desktop_mode", defaults.desktop_mode) - storage_config = ( - StorageConfig.from_dict(data["storage"], desktop_mode=desktop_mode) - if "storage" in data - else defaults.storage - ) - redis_config = RedisConfig.model_validate(data["redis"]) if "redis" in data else defaults.redis + desktop_mode = data.get("desktop_mode", False) + if "storage" in data: + storage_config = StorageConfig.model_validate(data["storage"]) + # Desktop-mode workspace validation and system workspace injection + workspaces = dict(storage_config.workspaces) + StorageConfig.validate_workspaces(workspaces, desktop_mode) + if desktop_mode: + workspaces = {**workspaces, **StorageConfig.system_workspaces()} + storage_config = storage_config.model_copy(update={"workspaces": workspaces}) + else: + storage_config = StorageConfig() + redis_config = RedisConfig.model_validate(data["redis"]) if "redis" in data else None return cls( - server=ServerConfig.model_validate(data["server"]) if "server" in data else defaults.server, - security=SecurityConfig.model_validate(data["security"]) if "security" in data else defaults.security, + server=ServerConfig.model_validate(data["server"]) if "server" in data else ServerConfig(), + security=SecurityConfig.model_validate(data["security"]) if "security" in data else SecurityConfig(), storage=storage_config, - launcher=LauncherConfig.model_validate(data["launcher"]) if "launcher" in data else defaults.launcher, - db=DbConfig.model_validate(data["db"]) if "db" in data else defaults.db, - logging=LoggingConfig.model_validate(data["logging"]) if "logging" in data else defaults.logging, - debug=data.get("debug", defaults.debug), - resources_path=data["resources_path"] if "resources_path" in data else defaults.resources_path, + launcher=LauncherConfig.model_validate(data["launcher"]) if "launcher" in data else LauncherConfig(), + db=DbConfig.model_validate(data["db"]) if "db" in data else DbConfig(), + logging=LoggingConfig.model_validate(data["logging"]) if "logging" in data else LoggingConfig(), + debug=data.get("debug", True), + resources_path=data["resources_path"] if "resources_path" in data else Path(), redis=redis_config, - eventbus=EventBusConfig.model_validate(data["eventbus"]) if "eventbus" in data else defaults.eventbus, - cache=CacheConfig.model_validate(data["cache"]) if "cache" in data else defaults.cache, - tasks=TaskConfig.from_dict(data["tasks"]) if "tasks" in data else defaults.tasks, + eventbus=EventBusConfig.model_validate(data["eventbus"]) if "eventbus" in data else EventBusConfig(), + cache=CacheConfig.model_validate(data["cache"]) if "cache" in data else CacheConfig(), + tasks=TaskConfig.model_validate(data["tasks"]) if "tasks" in data else TaskConfig(), celery=CeleryConfig.from_dict(data.get("celery", {}), redis_config=redis_config), - root_path=data.get("root_path", defaults.root_path), - api_prefix=data.get("api_prefix", defaults.api_prefix), + root_path=data.get("root_path", ""), + api_prefix=data.get("api_prefix", ""), desktop_mode=desktop_mode, metrics=MetricsConfig.model_validate(data["metrics"]) if "metrics" in data else MetricsConfig(), ) diff --git a/tests/storage/test_config.py b/tests/storage/test_config.py index a0d5dde986..d3509b911e 100644 --- a/tests/storage/test_config.py +++ b/tests/storage/test_config.py @@ -60,7 +60,7 @@ def test_storage_config_from_dict(storage_config_default: dict[str, Any]) -> Non }, } - config = StorageConfig.from_dict(data) + config = StorageConfig.model_validate(data) assert config.matrixstore == Path("./custom_matrixstore") assert config.archive_dir == Path("./custom_archives") @@ -165,9 +165,9 @@ def test_storage_config_from_dict_desktop_mode_true(storage_config_default: dict }, } - config = StorageConfig.from_dict(data, desktop_mode=True) + config = Config.from_dict({"storage": data, "desktop_mode": True}) - assert "local" in config.workspaces or "C:\\" in config.workspaces + assert "local" in config.storage.workspaces or "C:\\" in config.storage.workspaces def test_storage_config_from_dict_auto_archive() -> None: @@ -177,7 +177,7 @@ def test_storage_config_from_dict_auto_archive() -> None: } with pytest.raises(ValueError): - StorageConfig.from_dict(data) + StorageConfig.model_validate(data) def test_launcher_config_solver_versions(tmp_path: Path) -> None: From 9f701024cb4378fb40f62c0b61d9faa0e1bcee05 Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Mon, 15 Jun 2026 15:56:35 +0200 Subject: [PATCH 03/12] feat(task): add pydantic model validation on new class after rebase --- antarest/core/config.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 203d920263..204f2eccb6 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -17,7 +17,7 @@ import tempfile from enum import StrEnum from pathlib import Path -from typing import Any, ClassVar, Self +from typing import Any, ClassVar import yaml from antares.study.version import SolverVersion @@ -132,9 +132,8 @@ class OutOfStudyFileOutputStorageConfig(BaseModel): model_config = ConfigDict(frozen=True) storage_dir: Path = Path("./outputs") - @classmethod @model_validator(mode="after") - def validate_range(self) -> type[Self]: + def _create_storage_dir(self) -> "OutOfStudyFileOutputStorageConfig": self.storage_dir.mkdir(parents=True, exist_ok=True) return self From 162081f46c3af0a7f79f18f3ef0060626942a11c Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Mon, 15 Jun 2026 16:47:38 +0200 Subject: [PATCH 04/12] feat(task): add pydantic model validation for OutputStorageConfig --- antarest/core/config.py | 34 +++++++++++++++++++--------------- tests/storage/test_config.py | 2 +- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 204f2eccb6..4e6424c5c7 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -111,7 +111,6 @@ class DbConfig(BaseModel): pool_use_lifo: bool = False - class V2OutputStorageConfig(BaseModel): """ Configuration for "new style" internal output storage @@ -145,20 +144,25 @@ class OutputStorageConfig(BaseModel): v2: V2OutputStorageConfig = V2OutputStorageConfig() out_of_study: OutOfStudyFileOutputStorageConfig = OutOfStudyFileOutputStorageConfig() - default_storage_type: OutputStorageType = OutputStorageType.IN_STUDY_FILE_TREE + @model_validator(mode="before") @classmethod - def from_dict(cls, data: JSON) -> "OutputStorageConfig": - defaults = cls() - config = cls( - v2=V2OutputStorageConfig.model_validate(data.get("v2", {})), - out_of_study=OutOfStudyFileOutputStorageConfig.model_validate(data.get("out_of_study", {})), - default_storage_type=OutputStorageType(data.get("default_storage_type", defaults.default_storage_type)), - ) - if config.default_storage_type == OutputStorageType.V2 and not config.v2.enable: + def _validate_model(cls, data: Any) -> Any: + if "out_of_study" in data: + data["out_of_study"] = OutOfStudyFileOutputStorageConfig.model_validate(data["out_of_study"]) + if "default_storage_type" in data: + default_storage_type = OutputStorageType(data["default_storage_type"]) + if default_storage_type not in OutputStorageType: + raise ValueError(f"Invalid default_storage_type: {default_storage_type}") + data["default_storage_type"] = default_storage_type + return data + + @model_validator(mode="after") + def _validate_default_storage_type(self) -> "OutputStorageConfig": + if self.default_storage_type == OutputStorageType.V2 and not self.v2.enable: raise ValueError("You cannot set v2 storage as your default storage and not enable it") - return config + return self class StudyStorageConfig(BaseModel): @@ -528,6 +532,7 @@ class CacheConfig(BaseModel): """ Sub config object dedicated to cache module """ + model_config = ConfigDict(frozen=True) checker_delay: float = 0.2 # in seconds @@ -544,8 +549,8 @@ class TaskConfig(BaseModel): """ Sub config object dedicated to the task module """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True) max_workers: int = 5 remote_workers: list[RemoteWorkerConfig] = Field(default_factory=list) @@ -568,14 +573,13 @@ class ServerConfig(BaseModel): """ Sub config object dedicated to the server """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True) worker_threadpool_size: int = 5 services: list[str] = Field(default_factory=list) - class PrometheusConfig(BaseModel): """ Sub config object dedicated to prometheus metrics @@ -584,6 +588,7 @@ class PrometheusConfig(BaseModel): multiprocess: if True, metrics of workers will be aggregated before exposition. Environment variable `PROMETHEUS_MULTIPROC_DIR` must be set. """ + model_config = ConfigDict(frozen=True) multiprocess: bool = False @@ -602,7 +607,6 @@ class MetricsConfig(BaseModel): prometheus: PrometheusConfig | None = None - class CeleryConfig(BaseModel): """ Sub config object dedicated to Celery technical configuration. diff --git a/tests/storage/test_config.py b/tests/storage/test_config.py index d3509b911e..5b3cabbf07 100644 --- a/tests/storage/test_config.py +++ b/tests/storage/test_config.py @@ -228,4 +228,4 @@ def test_output_storage_config_validation_errors() -> None: with pytest.raises( ValueError, match=re.escape("You cannot set v2 storage as your default storage and not enable it") ): - OutputStorageConfig.from_dict(data) + OutputStorageConfig.model_validate(data) From 32d9c2608bdfdce5d21d9c87a630a80d99b4dc87 Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Tue, 16 Jun 2026 10:08:07 +0200 Subject: [PATCH 05/12] feat(task): add pydantic model validation for TaskConfig --- antarest/core/config.py | 17 ++--------------- 1 file changed, 2 insertions(+), 15 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 4e6424c5c7..97fc23d0a9 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -220,7 +220,7 @@ class StorageConfig(BaseModel): @field_validator("workspaces", mode="before") @classmethod - def convert_workspaces(cls, v: Any) -> dict[str, WorkspaceConfig]: + def convert_workspaces(cls, v: Any) -> Any: """Convert dict entries to WorkspaceConfig objects.""" if isinstance(v, dict): return { @@ -370,7 +370,7 @@ def _parse_binaries(cls, v: Any) -> Any: @model_validator(mode="wrap") @classmethod - def apply_cpu_detection(cls, data: Any, handler: Any) -> "LocalConfig": + def apply_cpu_detection(cls, data: Any, handler: Any) -> Any: """Inject detected CPU core counts into data before construction.""" if isinstance(data, dict) and data.get("enable_nb_cores_detection", True): data = {**data, "nb_cores": cls._autodetect_nb_cores()} @@ -555,19 +555,6 @@ class TaskConfig(BaseModel): max_workers: int = 5 remote_workers: list[RemoteWorkerConfig] = Field(default_factory=list) - @classmethod - def from_dict(cls, data: JSON) -> "TaskConfig": - defaults = cls() - remote_workers = ( - [RemoteWorkerConfig.model_validate(d) for d in data["remote_workers"]] - if "remote_workers" in data - else defaults.remote_workers - ) - return cls( - max_workers=data.get("max_workers", defaults.max_workers), - remote_workers=remote_workers, - ) - class ServerConfig(BaseModel): """ From 707efc5e6abf82210156bab42bb5da3a8169ebb9 Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Tue, 16 Jun 2026 10:45:10 +0200 Subject: [PATCH 06/12] feat(task): add pydantic model validation for CeleryConfig --- antarest/core/config.py | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 97fc23d0a9..2183e66c67 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -618,20 +618,17 @@ def _build_redis_url(redis_config: RedisConfig, db: int) -> str: password_part = f":{redis_config.password}@" if redis_config.password else "" return f"redis://{password_part}{redis_config.host}:{redis_config.port}/{db}" + @model_validator(mode="before") @classmethod - def from_dict(cls, data: JSON, redis_config: RedisConfig | None = None) -> "CeleryConfig": - defaults = cls() + def _validate_model(cls, data: Any) -> Any: + redis_url = "" + if "redis_config" in data: + redis_url = cls._build_redis_url(data["redis_config"], cls.REDIS_DB) - if redis_config: - redis_url = cls._build_redis_url(redis_config, cls.REDIS_DB) - else: - redis_url = "" + data["broker_url"] = data.get("broker_url", redis_url) + data["result_backend"] = data.get("result_backend", redis_url) - return cls( - broker_url=data.get("broker_url", redis_url), - result_backend=data.get("result_backend", redis_url), - result_expires=data.get("result_expires", defaults.result_expires), - ) + return data class Config(BaseModel): @@ -686,7 +683,7 @@ def from_dict(cls, data: JSON) -> "Config": eventbus=EventBusConfig.model_validate(data["eventbus"]) if "eventbus" in data else EventBusConfig(), cache=CacheConfig.model_validate(data["cache"]) if "cache" in data else CacheConfig(), tasks=TaskConfig.model_validate(data["tasks"]) if "tasks" in data else TaskConfig(), - celery=CeleryConfig.from_dict(data.get("celery", {}), redis_config=redis_config), + celery=CeleryConfig.model_validate(data.get("celery", {}) | {redis_config: redis_config}), root_path=data.get("root_path", ""), api_prefix=data.get("api_prefix", ""), desktop_mode=desktop_mode, From 58128a39936b716ee23960a0c0e4915a75a35b41 Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Tue, 16 Jun 2026 16:32:08 +0200 Subject: [PATCH 07/12] feat(task): add pydantic model validation for Config --- antarest/core/config.py | 63 ++++++++++--------- .../test_disk_usage_log.py | 2 +- tests/storage/business/test_watcher.py | 30 +++++---- tests/storage/test_config.py | 6 +- 4 files changed, 55 insertions(+), 46 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 2183e66c67..9885ad42cd 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -23,7 +23,6 @@ from antares.study.version import SolverVersion from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator -from antarest.core.model import JSON from antarest.core.roles import RoleType from antarest.core.utils.archives import ArchiveFormat from antarest.output.storage.output_storage import OutputStorageType @@ -656,39 +655,45 @@ class Config(BaseModel): api_prefix: str = "" desktop_mode: bool = False + @model_validator(mode="before") @classmethod - def from_dict(cls, data: JSON) -> "Config": + def _model_validator(cls, data: Any) -> Any: desktop_mode = data.get("desktop_mode", False) if "storage" in data: - storage_config = StorageConfig.model_validate(data["storage"]) - # Desktop-mode workspace validation and system workspace injection - workspaces = dict(storage_config.workspaces) - StorageConfig.validate_workspaces(workspaces, desktop_mode) - if desktop_mode: - workspaces = {**workspaces, **StorageConfig.system_workspaces()} - storage_config = storage_config.model_copy(update={"workspaces": workspaces}) + storage_data = data["storage"] + storage_config = StorageConfig.model_validate(storage_data) + # Desktop-mode workspace validation and system workspace injection only when + # parsing from a raw dict (i.e. YAML config). When constructing Config directly + # with pre-built objects, skip these semantic checks. + if isinstance(storage_data, dict): + workspaces = dict(storage_config.workspaces) + StorageConfig.validate_workspaces(workspaces, desktop_mode) + if desktop_mode: + workspaces = {**workspaces, **StorageConfig.system_workspaces()} + storage_config = storage_config.model_copy(update={"workspaces": workspaces}) else: storage_config = StorageConfig() redis_config = RedisConfig.model_validate(data["redis"]) if "redis" in data else None - return cls( - server=ServerConfig.model_validate(data["server"]) if "server" in data else ServerConfig(), - security=SecurityConfig.model_validate(data["security"]) if "security" in data else SecurityConfig(), - storage=storage_config, - launcher=LauncherConfig.model_validate(data["launcher"]) if "launcher" in data else LauncherConfig(), - db=DbConfig.model_validate(data["db"]) if "db" in data else DbConfig(), - logging=LoggingConfig.model_validate(data["logging"]) if "logging" in data else LoggingConfig(), - debug=data.get("debug", True), - resources_path=data["resources_path"] if "resources_path" in data else Path(), - redis=redis_config, - eventbus=EventBusConfig.model_validate(data["eventbus"]) if "eventbus" in data else EventBusConfig(), - cache=CacheConfig.model_validate(data["cache"]) if "cache" in data else CacheConfig(), - tasks=TaskConfig.model_validate(data["tasks"]) if "tasks" in data else TaskConfig(), - celery=CeleryConfig.model_validate(data.get("celery", {}) | {redis_config: redis_config}), - root_path=data.get("root_path", ""), - api_prefix=data.get("api_prefix", ""), - desktop_mode=desktop_mode, - metrics=MetricsConfig.model_validate(data["metrics"]) if "metrics" in data else MetricsConfig(), - ) + + data["server"] = ServerConfig.model_validate(data["server"]) if "server" in data else ServerConfig() + data["security"] = SecurityConfig.model_validate(data["security"]) if "security" in data else SecurityConfig() + data["storage"] = storage_config + data["launcher"] = LauncherConfig.model_validate(data["launcher"]) if "launcher" in data else LauncherConfig() + data["db"] = DbConfig.model_validate(data["db"]) if "db" in data else DbConfig() + data["logging"] = LoggingConfig.model_validate(data["logging"]) if "logging" in data else LoggingConfig() + data["debug"] = data.get("debug", True) + data["resources_path"] = data["resources_path"] if "resources_path" in data else Path() + data["redis"] = redis_config + data["eventbus"] = EventBusConfig.model_validate(data["eventbus"]) if "eventbus" in data else EventBusConfig() + data["cache"] = CacheConfig.model_validate(data["cache"]) if "cache" in data else CacheConfig() + data["tasks"] = TaskConfig.model_validate(data["tasks"]) if "tasks" in data else TaskConfig() + data["celery"] = CeleryConfig.model_validate(data.get("celery", {}) | {redis_config: redis_config}) + data["root_path"] = data.get("root_path", "") + data["api_prefix"] = data.get("api_prefix", "") + data["desktop_mode"] = desktop_mode + data["metrics"] = MetricsConfig.model_validate(data["metrics"]) if "metrics" in data else MetricsConfig() + + return data @classmethod def from_yaml_file(cls, file: Path, res: Path | None = None) -> "Config": @@ -706,7 +711,7 @@ def from_yaml_file(cls, file: Path, res: Path | None = None) -> "Config": data = yaml.safe_load(f) if res is not None: data["resources_path"] = res - return cls.from_dict(data) + return cls.model_validate(data) def get_workspace_path(self, *, workspace: str = DEFAULT_WORKSPACE_NAME) -> Path: """ diff --git a/tests/integration/maintenance_blueprint/test_disk_usage_log.py b/tests/integration/maintenance_blueprint/test_disk_usage_log.py index 0343a464dc..14d59790d4 100644 --- a/tests/integration/maintenance_blueprint/test_disk_usage_log.py +++ b/tests/integration/maintenance_blueprint/test_disk_usage_log.py @@ -29,7 +29,7 @@ def test_disk_usage_log(self): assert result.status == BackGroundTaskStatus.SUCCESS def test_returns_skipped_when_lock_held(self): - config = Config.from_dict({"storage": {"tmp_dir": Path(tempfile.gettempdir())}}) + config = Config.model_validate({"storage": {"tmp_dir": Path(tempfile.gettempdir())}}) with db(): with create_file_lock(lock_id=LockId.DISK_USAGE, lock_folder=Path(config.storage.tmp_dir)): result = disk_usage_logging(config) diff --git a/tests/storage/business/test_watcher.py b/tests/storage/business/test_watcher.py index d8b12c916e..cd53218c37 100644 --- a/tests/storage/business/test_watcher.py +++ b/tests/storage/business/test_watcher.py @@ -44,22 +44,26 @@ def build_config(root: Path, desktop_mode: bool = False) -> Config: + + no_desktop_workespaces = { + DEFAULT_WORKSPACE_NAME: WorkspaceConfig(path=root / DEFAULT_WORKSPACE_NAME, groups=["toto"]), + "diese": WorkspaceConfig( + path=root / "diese", + groups=["tata"], + filter_out=["to_skip.*"], + ), + "test": WorkspaceConfig( + path=root / "test", + groups=["toto"], + filter_out=["to_skip.*"], + ), + } + desktop_workspace = {DEFAULT_WORKSPACE_NAME: no_desktop_workespaces[DEFAULT_WORKSPACE_NAME]} + return Config( desktop_mode=desktop_mode, storage=StorageConfig( - workspaces={ - DEFAULT_WORKSPACE_NAME: WorkspaceConfig(path=root / DEFAULT_WORKSPACE_NAME, groups=["toto"]), - "diese": WorkspaceConfig( - path=root / "diese", - groups=["tata"], - filter_out=["to_skip.*"], - ), - "test": WorkspaceConfig( - path=root / "test", - groups=["toto"], - filter_out=["to_skip.*"], - ), - } + workspaces=desktop_workspace if desktop_mode else no_desktop_workespaces, ), ) diff --git a/tests/storage/test_config.py b/tests/storage/test_config.py index 5b3cabbf07..c5ae1c2cda 100644 --- a/tests/storage/test_config.py +++ b/tests/storage/test_config.py @@ -150,9 +150,9 @@ def test_storage_config_from_dict_validation_errors( if should_raise: with pytest.raises(ValueError): - Config.from_dict(config_data) + Config.model_validate(config_data) else: - Config.from_dict(config_data) + Config.model_validate(config_data) def test_storage_config_from_dict_desktop_mode_true(storage_config_default: dict[str, Any]) -> None: @@ -165,7 +165,7 @@ def test_storage_config_from_dict_desktop_mode_true(storage_config_default: dict }, } - config = Config.from_dict({"storage": data, "desktop_mode": True}) + config = Config.model_validate({"storage": data, "desktop_mode": True}) assert "local" in config.storage.workspaces or "C:\\" in config.storage.workspaces From 8760f2556516b186fec61567e76c421717fbe70c Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Tue, 16 Jun 2026 17:24:30 +0200 Subject: [PATCH 08/12] feat(task): refactor --- antarest/core/config.py | 45 ++++++++++++++++++++++------------------- 1 file changed, 24 insertions(+), 21 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 9885ad42cd..daf3922ecf 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -97,7 +97,7 @@ class DbConfig(BaseModel): Sub config object dedicated to db """ - model_config = ConfigDict(frozen=True, populate_by_name=True) + model_config = ConfigDict(frozen=True) db_url: str = Field(default="", alias="url") db_admin_url: str | None = Field(default=None, alias="admin_url") @@ -152,8 +152,6 @@ def _validate_model(cls, data: Any) -> Any: data["out_of_study"] = OutOfStudyFileOutputStorageConfig.model_validate(data["out_of_study"]) if "default_storage_type" in data: default_storage_type = OutputStorageType(data["default_storage_type"]) - if default_storage_type not in OutputStorageType: - raise ValueError(f"Invalid default_storage_type: {default_storage_type}") data["default_storage_type"] = default_storage_type return data @@ -344,7 +342,7 @@ def _validate_range(self) -> "TimeLimitConfig": class LocalConfig(BaseModel): """Sub config object dedicated to launcher module (local)""" - model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) + model_config = ConfigDict(frozen=True) id: str name: str @@ -392,7 +390,7 @@ class SlurmConfig(BaseModel): Sub config object dedicated to launcher module (slurm) """ - model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) + model_config = ConfigDict(frozen=True) id: str name: str @@ -457,7 +455,7 @@ class LauncherConfig(BaseModel): Sub config object dedicated to launcher module """ - model_config = ConfigDict(frozen=True, populate_by_name=True) + model_config = ConfigDict(frozen=True) default: str = Field(default="local") configs: list[LocalConfig | SlurmConfig] | None = None @@ -500,7 +498,7 @@ class LoggingConfig(BaseModel): Sub config object dedicated to logging """ - model_config = ConfigDict(frozen=True, populate_by_name=True) + model_config = ConfigDict(frozen=True) logfile: Path | None = None json_format: bool = Field(default=False, alias="json") @@ -659,20 +657,7 @@ class Config(BaseModel): @classmethod def _model_validator(cls, data: Any) -> Any: desktop_mode = data.get("desktop_mode", False) - if "storage" in data: - storage_data = data["storage"] - storage_config = StorageConfig.model_validate(storage_data) - # Desktop-mode workspace validation and system workspace injection only when - # parsing from a raw dict (i.e. YAML config). When constructing Config directly - # with pre-built objects, skip these semantic checks. - if isinstance(storage_data, dict): - workspaces = dict(storage_config.workspaces) - StorageConfig.validate_workspaces(workspaces, desktop_mode) - if desktop_mode: - workspaces = {**workspaces, **StorageConfig.system_workspaces()} - storage_config = storage_config.model_copy(update={"workspaces": workspaces}) - else: - storage_config = StorageConfig() + storage_config = cls._build_storage_config(data, desktop_mode) redis_config = RedisConfig.model_validate(data["redis"]) if "redis" in data else None data["server"] = ServerConfig.model_validate(data["server"]) if "server" in data else ServerConfig() @@ -695,6 +680,24 @@ def _model_validator(cls, data: Any) -> Any: return data + @classmethod + def _build_storage_config(cls, data: Any, desktop_mode: bool) -> StorageConfig: + if "storage" in data: + storage_data = data["storage"] + storage_config = StorageConfig.model_validate(storage_data) + # Desktop-mode workspace validation and system workspace injection only when + # parsing from a raw dict (i.e. YAML config). When constructing Config directly + # with pre-built objects, skip these semantic checks. + if isinstance(storage_data, dict): + workspaces = dict(storage_config.workspaces) + StorageConfig.validate_workspaces(workspaces, desktop_mode) + if desktop_mode: + workspaces = {**workspaces, **StorageConfig.system_workspaces()} + storage_config = storage_config.model_copy(update={"workspaces": workspaces}) + else: + storage_config = StorageConfig() + return storage_config + @classmethod def from_yaml_file(cls, file: Path, res: Path | None = None) -> "Config": """ From 850afa49d599f878cdb2eb894d56f7bc85d8f0b6 Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Wed, 17 Jun 2026 15:23:31 +0200 Subject: [PATCH 09/12] feat(task): add dedicated test --- antarest/core/config.py | 30 +-- tests/core/test_config.py | 266 +++++++++++++++++++++++++ tests/storage/business/test_watcher.py | 6 +- 3 files changed, 278 insertions(+), 24 deletions(-) create mode 100644 tests/core/test_config.py diff --git a/antarest/core/config.py b/antarest/core/config.py index daf3922ecf..865af6b028 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -97,7 +97,8 @@ class DbConfig(BaseModel): Sub config object dedicated to db """ - model_config = ConfigDict(frozen=True) + # populate_by_name so DbConfig can be creating using DbConfig(url=...) or DbConfig(db_url=...) + model_config = ConfigDict(frozen=True, populate_by_name=True) db_url: str = Field(default="", alias="url") db_admin_url: str | None = Field(default=None, alias="admin_url") @@ -142,7 +143,7 @@ class OutputStorageConfig(BaseModel): model_config = ConfigDict(frozen=True) v2: V2OutputStorageConfig = V2OutputStorageConfig() - out_of_study: OutOfStudyFileOutputStorageConfig = OutOfStudyFileOutputStorageConfig() + out_of_study: OutOfStudyFileOutputStorageConfig = Field(default_factory=OutOfStudyFileOutputStorageConfig) default_storage_type: OutputStorageType = OutputStorageType.IN_STUDY_FILE_TREE @model_validator(mode="before") @@ -365,13 +366,13 @@ def _parse_binaries(cls, v: Any) -> Any: } return v - @model_validator(mode="wrap") + @model_validator(mode="before") @classmethod - def apply_cpu_detection(cls, data: Any, handler: Any) -> Any: + def apply_cpu_detection(cls, data: Any) -> Any: """Inject detected CPU core counts into data before construction.""" if isinstance(data, dict) and data.get("enable_nb_cores_detection", True): data = {**data, "nb_cores": cls._autodetect_nb_cores()} - return handler(data) + return data @classmethod def _autodetect_nb_cores(cls) -> dict[str, int]: @@ -436,6 +437,9 @@ def handle_backward_compatibility(cls, data: Any) -> Any: nb_cores["min"] = min(nb_cores["min"], default_n_cpu) nb_cores["max"] = max(nb_cores["max"], default_n_cpu) data["nb_cores"] = nb_cores + + max_time = data.pop("default_time_limit", TimeLimitConfig().max * 3600) / 3600 + data["time_limit"] = TimeLimitConfig(min=1, default=max_time, max=max_time) return data @@ -659,24 +663,8 @@ def _model_validator(cls, data: Any) -> Any: desktop_mode = data.get("desktop_mode", False) storage_config = cls._build_storage_config(data, desktop_mode) redis_config = RedisConfig.model_validate(data["redis"]) if "redis" in data else None - - data["server"] = ServerConfig.model_validate(data["server"]) if "server" in data else ServerConfig() - data["security"] = SecurityConfig.model_validate(data["security"]) if "security" in data else SecurityConfig() data["storage"] = storage_config - data["launcher"] = LauncherConfig.model_validate(data["launcher"]) if "launcher" in data else LauncherConfig() - data["db"] = DbConfig.model_validate(data["db"]) if "db" in data else DbConfig() - data["logging"] = LoggingConfig.model_validate(data["logging"]) if "logging" in data else LoggingConfig() - data["debug"] = data.get("debug", True) - data["resources_path"] = data["resources_path"] if "resources_path" in data else Path() data["redis"] = redis_config - data["eventbus"] = EventBusConfig.model_validate(data["eventbus"]) if "eventbus" in data else EventBusConfig() - data["cache"] = CacheConfig.model_validate(data["cache"]) if "cache" in data else CacheConfig() - data["tasks"] = TaskConfig.model_validate(data["tasks"]) if "tasks" in data else TaskConfig() - data["celery"] = CeleryConfig.model_validate(data.get("celery", {}) | {redis_config: redis_config}) - data["root_path"] = data.get("root_path", "") - data["api_prefix"] = data.get("api_prefix", "") - data["desktop_mode"] = desktop_mode - data["metrics"] = MetricsConfig.model_validate(data["metrics"]) if "metrics" in data else MetricsConfig() return data diff --git a/tests/core/test_config.py b/tests/core/test_config.py new file mode 100644 index 0000000000..c412fa1f81 --- /dev/null +++ b/tests/core/test_config.py @@ -0,0 +1,266 @@ +# Copyright (c) 2026, RTE (https://www.rte-france.com) +# +# See AUTHORS.txt +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. +# +# SPDX-License-Identifier: MPL-2.0 +# +# This file is part of the Antares project. +import os.path +import tempfile +from pathlib import Path +from typing import cast + +import yaml +from antares.study.version import SolverVersion +from core.config import CacheConfig, CeleryConfig, DbConfig, LoggingConfig, MetricsConfig, RedisConfig, TaskConfig + +from antarest.core.config import ( + Config, + LauncherConfig, + LocalConfig, + SecurityConfig, + ServerConfig, + SlurmConfig, + StorageConfig, +) + + +class TestConfig: + def test_config_is_fully_loaded(self) -> None: + """ + Test that the launcher configuration correctly parses solver versions written in different formats. + """ + data = { + "server": {"worker_threadpool_size": 12, "services": ["watcher"]}, + "desktop_mode": False, + "security": {"disabled": "true", "jwt": {"key": "my-secret-key"}}, + "storage": { + "tmp_dir": "/tmp", + "matrixstore": "/tmp/matrixstore", + "archive_dir": "/tmp/archive", + "workspaces": { + "default": { + "path": "/tmp/default_workspace", + "filter_in": ["filter_in_1", "filter_in_2"], + "filter_out": ["filter_out_1", "filter_out_2"], + "groups": ["group_1", "group_2"], + }, + "studies": {"path": "/tmp/studies"}, + }, + }, + "launcher": { + "default": "test_local", + "launchers": [ + { + "id": "test_local", + "name": "local", + "type": "local", + "binaries": {"880": "", "9.2": "", "10.0": ""}, + "enable_nb_cores_detection": False, + "nb_cores": {"min": 10, "default": 100, "max": 1000}, + "time_limit": {"min": 11, "default": 101, "max": 1001}, + "xpress_dir": "/tmp/xpress_dir", + "local_workspace": "/tmp/local_workspace", + }, + { + "id": "slurm_id", + "name": "slurm_name", + "type": "slurm", + "antares_versions_on_remote_server": ["8", "880", "9.2", "10.0"], + "local_workspace": "/tmp/slurm/local_workspace", + "username": "slurm_username", + "hostname": "slurm_hostname", + "port": 8888, + "private_key_file": "/tmp/slurm_key", + "key_password": "slurm_key_password", + "password": "slurm_password", + "default_wait_time": 60, + "nb_cores": {"min": 20, "default": 200, "max": 2000}, + "default_json_db_name": "slurm_default_db_name", + "slurm_script_path": "slurm_script_path", + "partition": "slurm_partition", + "max_cores": 250, + "enable_nb_cores_detection": True, + }, + ], + }, + "db": { + "url": "db_url", + "admin_url": "db_admin_url", + "db_connect_timeout": 15, + "pool_recycle": 3, + "pool_pre_ping": True, + "pool_use_null": True, + "pool_max_overflow": 13, + "pool_size": 20, + "pool_use_lifo": True, + }, + "logging": {"logfile": "/tmp/logfile", "json": True, "level": "DEBUG"}, + "debug": False, + "resources_path": "/tmp/resources_path", + "redis": {"host": "redis_host", "port": 1234, "password": "redis_password"}, + "cache": {"checker_delay": 0.5}, + "tasks": { + "max_workers": 22, + "remote_workers": [ + {"name": "worker_1", "queues": ["queue_1_1", "queue_1_2"]}, + {"name": "worker_2", "queues": ["queue_2_1", "queue_2_2"]}, + ], + }, + "metrics": {"prometheus": {"multiprocess": True}}, + "celery": {"broker_url": "celery_url", "result_backend": "celery_backend", "result_expires": 60}, + "root_path": "/tmp/root_path", + "api_prefix": "/api_prefix", + } + + tmp_path = tempfile.mkdtemp() + config_path = Path(os.path.join(tmp_path, "config.yaml")) + + with config_path.open(mode="w", encoding="utf-8") as fd: + yaml.dump(data, fd) + config = Config.from_yaml_file(config_path) + + assert config.desktop_mode is False + assert config.debug is False + assert config.resources_path == Path("/tmp/resources_path") + + self.check_server_config(config.server) + self.check_security_config(config.security) + self.check_storage_config(config.storage) + self.check_launchers_config(config.launcher) + self.check_db_config(config.db) + self.check_logging_config(config.logging) + self.check_redis_config(config.redis) + self.check_cache_config(config.cache) + self.check_task_config(config.tasks) + self.check_metrics_config(config.metrics) + self.check_celery_config(config.celery) + assert config.root_path == "/tmp/root_path" + assert config.api_prefix == "/api_prefix" + + def check_server_config(self, server_config: ServerConfig): + assert server_config.worker_threadpool_size == 12 + assert server_config.services == ["watcher"] + + def check_security_config(self, security_config: SecurityConfig) -> None: + assert security_config.disabled is True + assert security_config.jwt_key == "my-secret-key" + + def check_storage_config(self, storage_config: StorageConfig) -> None: + assert storage_config.tmp_dir == Path("/tmp") + assert storage_config.matrixstore == Path("/tmp/matrixstore") + assert storage_config.archive_dir == Path("/tmp/archive") + + workspaces = storage_config.workspaces + assert workspaces["default"].path == Path("/tmp/default_workspace") + assert workspaces["default"].filter_in == ["filter_in_1", "filter_in_2"] + assert workspaces["default"].filter_out == ["filter_out_1", "filter_out_2"] + assert workspaces["default"].groups == ["group_1", "group_2"] + + assert workspaces["studies"].path == Path("/tmp/studies") + + def check_launchers_config(self, launcher_config: LauncherConfig) -> None: + assert launcher_config.default == "test_local" + + launchers = launcher_config.configs + assert launchers is not None + assert len(launchers) == 2 + + local_launcher = cast(LocalConfig, launchers[0]) + slurm_launcher = cast(SlurmConfig, launchers[1]) + self.check_local_launcher(local_launcher) + self.check_slurm_launcher(slurm_launcher) + + def check_local_launcher(self, local_launcher: LocalConfig) -> None: + assert local_launcher.id == "test_local" + assert local_launcher.name == "local" + assert local_launcher.type == "local" + assert local_launcher.binaries == { + SolverVersion(major=8, minor=8, patch=0): Path("."), + SolverVersion(major=9, minor=2, patch=0): Path("."), + SolverVersion(major=10, minor=0, patch=0): Path("."), + } + assert local_launcher.enable_nb_cores_detection is False + assert local_launcher.nb_cores.min == 10 + assert local_launcher.nb_cores.default == 100 + assert local_launcher.nb_cores.max == 1000 + + assert local_launcher.time_limit.min == 11 + assert local_launcher.time_limit.default == 101 + assert local_launcher.time_limit.max == 1001 + + assert local_launcher.xpress_dir == "/tmp/xpress_dir" + assert local_launcher.local_workspace == Path("/tmp/local_workspace") + + def check_slurm_launcher(self, slurm_launcher): + assert slurm_launcher.id == "slurm_id" + assert slurm_launcher.name == "slurm_name" + assert slurm_launcher.type == "slurm" + assert slurm_launcher.antares_versions_on_remote_server == [ + SolverVersion(major=8, minor=0, patch=0), + SolverVersion(major=8, minor=8, patch=0), + SolverVersion(major=9, minor=2, patch=0), + SolverVersion(major=10, minor=0, patch=0), + ] + assert slurm_launcher.local_workspace == Path("/tmp/slurm/local_workspace") + assert slurm_launcher.username == "slurm_username" + assert slurm_launcher.hostname == "slurm_hostname" + assert slurm_launcher.port == 8888 + assert slurm_launcher.private_key_file == Path("/tmp/slurm_key") + assert slurm_launcher.key_password == "slurm_key_password" + assert slurm_launcher.password == "slurm_password" + assert slurm_launcher.default_wait_time == 60 + assert slurm_launcher.nb_cores.min == 20 + assert slurm_launcher.nb_cores.default == 200 + assert slurm_launcher.nb_cores.max == 2000 + assert slurm_launcher.default_json_db_name == "slurm_default_db_name" + assert slurm_launcher.slurm_script_path == "slurm_script_path" + assert slurm_launcher.partition == "slurm_partition" + assert slurm_launcher.max_cores == 250 + assert slurm_launcher.enable_nb_cores_detection is True + + def check_db_config(self, db_config: DbConfig) -> None: + assert db_config.db_url == "db_url" + assert db_config.db_admin_url == "db_admin_url" + assert db_config.db_connect_timeout == 15 + assert db_config.pool_recycle == 3 + assert db_config.pool_pre_ping == True + assert db_config.pool_use_null == True + assert db_config.pool_max_overflow == 13 + assert db_config.pool_size == 20 + assert db_config.pool_use_lifo == True + + def check_logging_config(self, logging_config: LoggingConfig) -> None: + assert logging_config.logfile == Path("/tmp/logfile") + assert logging_config.json_format == True + assert logging_config.level == "DEBUG" + + def check_redis_config(self, redis_config: RedisConfig | None) -> None: + assert redis_config is not None + assert redis_config.host == "redis_host" + assert redis_config.port == 1234 + assert redis_config.password == "redis_password" + + def check_cache_config(self, cache_config: CacheConfig) -> None: + assert cache_config.checker_delay == 0.5 + + def check_task_config(self, task_config: TaskConfig) -> None: + assert task_config.max_workers == 22 + workers = task_config.remote_workers + assert workers[0].name == "worker_1" + assert workers[0].queues == ["queue_1_1", "queue_1_2"] + + assert workers[1].name == "worker_2" + assert workers[1].queues == ["queue_2_1", "queue_2_2"] + + def check_metrics_config(self, metrics_config: MetricsConfig) -> None: + assert metrics_config.prometheus.multiprocess == True + + def check_celery_config(self, celery_config: CeleryConfig) -> None: + assert celery_config.broker_url == "celery_url" + assert celery_config.result_backend == "celery_backend" + assert celery_config.result_expires == 60 diff --git a/tests/storage/business/test_watcher.py b/tests/storage/business/test_watcher.py index cd53218c37..4000ed8e6c 100644 --- a/tests/storage/business/test_watcher.py +++ b/tests/storage/business/test_watcher.py @@ -45,7 +45,7 @@ def build_config(root: Path, desktop_mode: bool = False) -> Config: - no_desktop_workespaces = { + no_desktop_workspaces = { DEFAULT_WORKSPACE_NAME: WorkspaceConfig(path=root / DEFAULT_WORKSPACE_NAME, groups=["toto"]), "diese": WorkspaceConfig( path=root / "diese", @@ -58,12 +58,12 @@ def build_config(root: Path, desktop_mode: bool = False) -> Config: filter_out=["to_skip.*"], ), } - desktop_workspace = {DEFAULT_WORKSPACE_NAME: no_desktop_workespaces[DEFAULT_WORKSPACE_NAME]} + desktop_workspace = {DEFAULT_WORKSPACE_NAME: no_desktop_workspaces[DEFAULT_WORKSPACE_NAME]} return Config( desktop_mode=desktop_mode, storage=StorageConfig( - workspaces=desktop_workspace if desktop_mode else no_desktop_workespaces, + workspaces=desktop_workspace if desktop_mode else no_desktop_workspaces, ), ) From 200e3d3dde0f25e589d9d5a7ec6d73611293b83d Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Wed, 17 Jun 2026 15:36:17 +0200 Subject: [PATCH 10/12] feat(task): refactor --- antarest/core/config.py | 103 ++++++++++++---------------------------- 1 file changed, 31 insertions(+), 72 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 865af6b028..0f891e8062 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -41,26 +41,28 @@ class InternalMatrixFormat(StrEnum): FEATHER = "feather" -class ExternalAuthConfig(BaseModel): +class ConfigBaseModel(BaseModel): + """Base model for all configuration classes.""" + + model_config = ConfigDict(frozen=True, populate_by_name=True, extra="ignore", validate_default=True) + + +class ExternalAuthConfig(ConfigBaseModel): """ Sub config object dedicated to external auth service """ - model_config = ConfigDict(frozen=True) - url: str | None = None default_group_role: RoleType = RoleType.READER add_ext_groups: bool = False group_mapping: dict[str, str] = Field(default_factory=dict) -class SecurityConfig(BaseModel): +class SecurityConfig(ConfigBaseModel): """ Sub config object dedicated to security """ - model_config = ConfigDict(frozen=True) - jwt_key: str = "" admin_pwd: str = "" disabled: bool = False @@ -79,20 +81,18 @@ def _flatten_nested_keys(cls, data: Any) -> Any: return data -class WorkspaceConfig(BaseModel): +class WorkspaceConfig(ConfigBaseModel): """ Sub config object dedicated to workspace """ - model_config = ConfigDict(frozen=True) - filter_in: list[str] = Field(default_factory=lambda: [".*"]) filter_out: list[str] = Field(default_factory=list) groups: list[str] = Field(default_factory=list) path: Path = Field(default_factory=Path) -class DbConfig(BaseModel): +class DbConfig(ConfigBaseModel): """ Sub config object dedicated to db """ @@ -111,24 +111,21 @@ class DbConfig(BaseModel): pool_use_lifo: bool = False -class V2OutputStorageConfig(BaseModel): +class V2OutputStorageConfig(ConfigBaseModel): """ Configuration for "new style" internal output storage """ - model_config = ConfigDict(frozen=True) - enable: bool = False archive_dir: Path = Path("./output-archives") variables_dir: Path = Path("./output-variables") -class OutOfStudyFileOutputStorageConfig(BaseModel): +class OutOfStudyFileOutputStorageConfig(ConfigBaseModel): """ Configuration for output storage in a single dir for all studies """ - model_config = ConfigDict(frozen=True) storage_dir: Path = Path("./outputs") @model_validator(mode="after") @@ -137,11 +134,9 @@ def _create_storage_dir(self) -> "OutOfStudyFileOutputStorageConfig": return self -class OutputStorageConfig(BaseModel): +class OutputStorageConfig(ConfigBaseModel): """Configuration for output storage""" - model_config = ConfigDict(frozen=True) - v2: V2OutputStorageConfig = V2OutputStorageConfig() out_of_study: OutOfStudyFileOutputStorageConfig = Field(default_factory=OutOfStudyFileOutputStorageConfig) default_storage_type: OutputStorageType = OutputStorageType.IN_STUDY_FILE_TREE @@ -163,23 +158,19 @@ def _validate_default_storage_type(self) -> "OutputStorageConfig": return self -class StudyStorageConfig(BaseModel): +class StudyStorageConfig(ConfigBaseModel): """ Sub config object dedicated to study storage configuration (from study.storage in YAML) """ - model_config = ConfigDict(frozen=True) - database_mode_enabled: bool = False -class StorageConfig(BaseModel): +class StorageConfig(ConfigBaseModel): """ Sub config object dedicated to study module """ - model_config = ConfigDict(frozen=True) - matrixstore: Path = Path("./matrixstore") archive_dir: Path = Path("./archives") archive_format: ArchiveFormat = ArchiveFormat.ZIP @@ -273,13 +264,11 @@ def system_workspaces(cls) -> dict[str, WorkspaceConfig]: raise NotImplementedError("System workspaces are only implemented for Linux and Windows") -class NbCoresConfig(BaseModel): +class NbCoresConfig(ConfigBaseModel): """ The NBCoresConfig class is designed to manage the configuration of the number of CPU cores """ - model_config = ConfigDict(frozen=True) - min: int = 1 default: int = 22 max: int = 24 @@ -304,7 +293,7 @@ def _validate_range(self) -> "NbCoresConfig": raise ValueError(msg) -class TimeLimitConfig(BaseModel): +class TimeLimitConfig(ConfigBaseModel): """ The TimeLimitConfig class is designed to manage the configuration of the time limit for a job. @@ -314,8 +303,6 @@ class TimeLimitConfig(BaseModel): max: int: maximum allowed value for the time limit (in hours). """ - model_config = ConfigDict(frozen=True) - min: int = 1 default: int = 48 max: int = 48 @@ -340,11 +327,9 @@ def _validate_range(self) -> "TimeLimitConfig": raise ValueError(msg) -class LocalConfig(BaseModel): +class LocalConfig(ConfigBaseModel): """Sub config object dedicated to launcher module (local)""" - model_config = ConfigDict(frozen=True) - id: str name: str type: ClassVar[LauncherType] = LauncherType.LOCAL @@ -386,13 +371,11 @@ def _autodetect_nb_cores(cls) -> dict[str, int]: return {"min": min_cpu, "max": max_cpu, "default": default} -class SlurmConfig(BaseModel): +class SlurmConfig(ConfigBaseModel): """ Sub config object dedicated to launcher module (slurm) """ - model_config = ConfigDict(frozen=True) - id: str name: str type: ClassVar[LauncherType] = LauncherType.SLURM @@ -454,13 +437,11 @@ def __init__(self, launcher: str): super().__init__(msg) -class LauncherConfig(BaseModel): +class LauncherConfig(ConfigBaseModel): """ Sub config object dedicated to launcher module """ - model_config = ConfigDict(frozen=True) - default: str = Field(default="local") configs: list[LocalConfig | SlurmConfig] | None = None batch_size: int = 9999 @@ -497,78 +478,64 @@ def get_slurm_configs(self) -> list[SlurmConfig]: return [cfg for cfg in self.configs or [] if isinstance(cfg, SlurmConfig)] -class LoggingConfig(BaseModel): +class LoggingConfig(ConfigBaseModel): """ Sub config object dedicated to logging """ - model_config = ConfigDict(frozen=True) - logfile: Path | None = None json_format: bool = Field(default=False, alias="json") level: str = "INFO" -class RedisConfig(BaseModel): +class RedisConfig(ConfigBaseModel): """ Sub config object dedicated to redis """ - model_config = ConfigDict(frozen=True) - host: str = "localhost" port: int = 6379 password: str | None = None -class EventBusConfig(BaseModel): +class EventBusConfig(ConfigBaseModel): """ Sub config object dedicated to eventbus module """ - model_config = ConfigDict(frozen=True) - -class CacheConfig(BaseModel): +class CacheConfig(ConfigBaseModel): """ Sub config object dedicated to cache module """ - model_config = ConfigDict(frozen=True) - checker_delay: float = 0.2 # in seconds -class RemoteWorkerConfig(BaseModel): - model_config = ConfigDict(frozen=True) - +class RemoteWorkerConfig(ConfigBaseModel): name: str queues: list[str] = Field(default_factory=list) -class TaskConfig(BaseModel): +class TaskConfig(ConfigBaseModel): """ Sub config object dedicated to the task module """ - model_config = ConfigDict(frozen=True) - max_workers: int = 5 remote_workers: list[RemoteWorkerConfig] = Field(default_factory=list) -class ServerConfig(BaseModel): +class ServerConfig(ConfigBaseModel): """ Sub config object dedicated to the server """ - model_config = ConfigDict(frozen=True) - worker_threadpool_size: int = 5 services: list[str] = Field(default_factory=list) -class PrometheusConfig(BaseModel): +class PrometheusConfig(ConfigBaseModel): """ Sub config object dedicated to prometheus metrics @@ -577,12 +544,10 @@ class PrometheusConfig(BaseModel): Environment variable `PROMETHEUS_MULTIPROC_DIR` must be set. """ - model_config = ConfigDict(frozen=True) - multiprocess: bool = False -class MetricsConfig(BaseModel): +class MetricsConfig(ConfigBaseModel): """ Sub config object dedicated to metrics @@ -590,12 +555,10 @@ class MetricsConfig(BaseModel): prometheus: if not None, metrics will be exposed in prometheus format """ - model_config = ConfigDict(frozen=True) - prometheus: PrometheusConfig | None = None -class CeleryConfig(BaseModel): +class CeleryConfig(ConfigBaseModel): """ Sub config object dedicated to Celery technical configuration. @@ -605,8 +568,6 @@ class CeleryConfig(BaseModel): result_expires: Time in seconds before task results expire """ - model_config = ConfigDict(frozen=True) - # Redis database number for Celery (broker + results) REDIS_DB: ClassVar[int] = 1 @@ -632,13 +593,11 @@ def _validate_model(cls, data: Any) -> Any: return data -class Config(BaseModel): +class Config(ConfigBaseModel): """ Root server config """ - model_config = ConfigDict(frozen=True) - server: ServerConfig = Field(default_factory=ServerConfig) security: SecurityConfig = Field(default_factory=SecurityConfig) storage: StorageConfig = Field(default_factory=StorageConfig) From be87c2970a6f91433f759a8688fd94db3991e071 Mon Sep 17 00:00:00 2001 From: pigeonoli Date: Wed, 17 Jun 2026 16:24:26 +0200 Subject: [PATCH 11/12] feat(task): load redis config in celery config --- antarest/core/config.py | 7 ++++-- tests/core/test_config.py | 46 +++++++++++++++++++++++++++++---------- 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/antarest/core/config.py b/antarest/core/config.py index 0f891e8062..47562cab4a 100644 --- a/antarest/core/config.py +++ b/antarest/core/config.py @@ -44,7 +44,7 @@ class InternalMatrixFormat(StrEnum): class ConfigBaseModel(BaseModel): """Base model for all configuration classes.""" - model_config = ConfigDict(frozen=True, populate_by_name=True, extra="ignore", validate_default=True) + model_config = ConfigDict(frozen=True, populate_by_name=True, validate_default=True) class ExternalAuthConfig(ConfigBaseModel): @@ -584,7 +584,7 @@ def _build_redis_url(redis_config: RedisConfig, db: int) -> str: @classmethod def _validate_model(cls, data: Any) -> Any: redis_url = "" - if "redis_config" in data: + if "redis_config" in data and data["redis_config"]: redis_url = cls._build_redis_url(data["redis_config"], cls.REDIS_DB) data["broker_url"] = data.get("broker_url", redis_url) @@ -624,6 +624,9 @@ def _model_validator(cls, data: Any) -> Any: redis_config = RedisConfig.model_validate(data["redis"]) if "redis" in data else None data["storage"] = storage_config data["redis"] = redis_config + data["celery"] = CeleryConfig.model_validate( + data["celery"] if "celery" in data else {} | {"redis_config": redis_config} + ) return data diff --git a/tests/core/test_config.py b/tests/core/test_config.py index c412fa1f81..a2210b6c34 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -12,20 +12,26 @@ import os.path import tempfile from pathlib import Path -from typing import cast +from typing import Any, cast import yaml from antares.study.version import SolverVersion -from core.config import CacheConfig, CeleryConfig, DbConfig, LoggingConfig, MetricsConfig, RedisConfig, TaskConfig from antarest.core.config import ( + CacheConfig, + CeleryConfig, Config, + DbConfig, LauncherConfig, LocalConfig, + LoggingConfig, + MetricsConfig, + RedisConfig, SecurityConfig, ServerConfig, SlurmConfig, StorageConfig, + TaskConfig, ) @@ -117,11 +123,7 @@ def test_config_is_fully_loaded(self) -> None: "api_prefix": "/api_prefix", } - tmp_path = tempfile.mkdtemp() - config_path = Path(os.path.join(tmp_path, "config.yaml")) - - with config_path.open(mode="w", encoding="utf-8") as fd: - yaml.dump(data, fd) + config_path = self.save_data_to_temporary_file(data) config = Config.from_yaml_file(config_path) assert config.desktop_mode is False @@ -142,6 +144,26 @@ def test_config_is_fully_loaded(self) -> None: assert config.root_path == "/tmp/root_path" assert config.api_prefix == "/api_prefix" + def test_redis_config_loaded_in_celery_config(self) -> None: + data = { + "redis": {"host": "redis_host", "port": 1234, "password": "redis_password"}, + } + + config_path = self.save_data_to_temporary_file(data) + + config = Config.from_yaml_file(config_path) + assert config.redis is not None + assert config.celery is not None + assert config.celery.broker_url == "redis://:redis_password@redis_host:1234/1" + + def save_data_to_temporary_file(self, data: dict[str, Any]) -> Path: + tmp_path = tempfile.mkdtemp() + config_path = Path(os.path.join(tmp_path, "config.yaml")) + + with config_path.open(mode="w", encoding="utf-8") as fd: + yaml.dump(data, fd) + return config_path + def check_server_config(self, server_config: ServerConfig): assert server_config.worker_threadpool_size == 12 assert server_config.services == ["watcher"] @@ -228,15 +250,15 @@ def check_db_config(self, db_config: DbConfig) -> None: assert db_config.db_admin_url == "db_admin_url" assert db_config.db_connect_timeout == 15 assert db_config.pool_recycle == 3 - assert db_config.pool_pre_ping == True - assert db_config.pool_use_null == True + assert db_config.pool_pre_ping + assert db_config.pool_use_null assert db_config.pool_max_overflow == 13 assert db_config.pool_size == 20 - assert db_config.pool_use_lifo == True + assert db_config.pool_use_lifo def check_logging_config(self, logging_config: LoggingConfig) -> None: assert logging_config.logfile == Path("/tmp/logfile") - assert logging_config.json_format == True + assert logging_config.json_format assert logging_config.level == "DEBUG" def check_redis_config(self, redis_config: RedisConfig | None) -> None: @@ -258,7 +280,7 @@ def check_task_config(self, task_config: TaskConfig) -> None: assert workers[1].queues == ["queue_2_1", "queue_2_2"] def check_metrics_config(self, metrics_config: MetricsConfig) -> None: - assert metrics_config.prometheus.multiprocess == True + assert metrics_config.prometheus.multiprocess def check_celery_config(self, celery_config: CeleryConfig) -> None: assert celery_config.broker_url == "celery_url" From ec46b8580f4074eb85f158ae911b1733cc4840fe Mon Sep 17 00:00:00 2001 From: Sylvain Leclerc Date: Mon, 29 Jun 2026 16:59:27 +0200 Subject: [PATCH 12/12] update unit tests Signed-off-by: Sylvain Leclerc --- tests/study/storage/test_database_storage.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/study/storage/test_database_storage.py b/tests/study/storage/test_database_storage.py index a681cfb5bc..f7e704398e 100644 --- a/tests/study/storage/test_database_storage.py +++ b/tests/study/storage/test_database_storage.py @@ -34,7 +34,7 @@ def test_upgrade_fails(tmp_path: Path) -> None: """ Ensures that if the upgrader fails when exporting data to the FS or during the upgrade process, the study remains in its original state. """ - config = Config.from_dict({"storage": {"tmp_dir": tmp_path}}) + config = Config.model_validate({"storage": {"tmp_dir": tmp_path}}) db_dao_factory = Mock() db_dao_factory.get_study_dao.side_effect = ValueError("Raises for the test") @@ -72,7 +72,7 @@ def test_upgrade_does_not_use_cache(tmp_path: Path, study_factory) -> None: # Set Up ########################## - config = Config.from_dict({"storage": {"tmp_dir": tmp_path}}) + config = Config.model_validate({"storage": {"tmp_dir": tmp_path}}) cache = LocalCache() matrix_service = InMemorySimpleMatrixService() generator_matrix_constants = GeneratorMatrixConstants(matrix_service)