From 8da7087bbea18353ef702916ff2e1a50f5604b1e Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 15 Oct 2026 10:14:06 +0200 Subject: [PATCH 1/5] Update pyproject.toml --- pyproject.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 299f33f..bd0bf55 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,10 @@ classifiers = [ ] dependencies = [ "huggingface_hub>=0.11.1", + "numpy>=1.22.4", + "pydantic>=2,<3", "ruamel.yaml>=0.16", + "typing_extensions", ] [project.urls] From 89fbb19762aabd4de0fc634edb3aa63718df5753 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 15 Oct 2026 10:16:25 +0200 Subject: [PATCH 2/5] Move core config files --- dlclibrary/deprecation.py | 14 + dlclibrary/io/__init__.py | 10 + dlclibrary/io/config/__init__.py | 38 +++ dlclibrary/io/config/base_config.py | 439 +++++++++++++++++++++++++ dlclibrary/io/config/project_config.py | 305 +++++++++++++++++ dlclibrary/io/config/utils.py | 196 +++++++++++ dlclibrary/io/config/validation.py | 105 ++++++ dlclibrary/io/config/versioning.py | 264 +++++++++++++++ 8 files changed, 1371 insertions(+) create mode 100644 dlclibrary/deprecation.py create mode 100644 dlclibrary/io/__init__.py create mode 100644 dlclibrary/io/config/__init__.py create mode 100644 dlclibrary/io/config/base_config.py create mode 100644 dlclibrary/io/config/project_config.py create mode 100644 dlclibrary/io/config/utils.py create mode 100644 dlclibrary/io/config/validation.py create mode 100644 dlclibrary/io/config/versioning.py diff --git a/dlclibrary/deprecation.py b/dlclibrary/deprecation.py new file mode 100644 index 0000000..c66c1cf --- /dev/null +++ b/dlclibrary/deprecation.py @@ -0,0 +1,14 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# + + +class DLCDeprecationWarning(DeprecationWarning): + """Project-specific deprecation warning. Helps with filtering.""" diff --git a/dlclibrary/io/__init__.py b/dlclibrary/io/__init__.py new file mode 100644 index 0000000..2dd1b06 --- /dev/null +++ b/dlclibrary/io/__init__.py @@ -0,0 +1,10 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# diff --git a/dlclibrary/io/config/__init__.py b/dlclibrary/io/config/__init__.py new file mode 100644 index 0000000..0121229 --- /dev/null +++ b/dlclibrary/io/config/__init__.py @@ -0,0 +1,38 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +from dlclibrary.io.config.base_config import DLCBaseConfig, DLCVersionedConfig +from dlclibrary.io.config.project_config import ProjectConfig +from dlclibrary.io.config.utils import ( + get_yaml_dumper, + get_yaml_loader, + normalize_for_serialization, + pretty_print, + read_config_as_dict, + resolve_alias, + resolve_aliases_in_dict, + write_config, +) + +__all__ = [ + "DLCBaseConfig", + "DLCVersionedConfig", + "ProjectConfig", + "pretty_print", + "read_config_as_dict", + "resolve_alias", + "resolve_aliases_in_dict", + "write_config", + "get_yaml_loader", + "get_yaml_dumper", + "versioning", + "validation", + "normalize_for_serialization", +] diff --git a/dlclibrary/io/config/base_config.py b/dlclibrary/io/config/base_config.py new file mode 100644 index 0000000..fe9761e --- /dev/null +++ b/dlclibrary/io/config/base_config.py @@ -0,0 +1,439 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +from __future__ import annotations + +import functools +import logging +import sys +from collections.abc import Callable, Iterator +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator +from pydantic_core import ArgsKwargs +from ruamel.yaml.comments import CommentedMap +from typing_extensions import Self + +from dlclibrary.io.config import versioning as versioning +from dlclibrary.io.config.utils import ( + normalize_for_serialization, + pretty_print, + read_config_as_dict, + resolve_alias, + resolve_aliases_in_dict, + write_config, +) + +logger = logging.getLogger(__name__) + + +class DLCBaseConfig(BaseModel): + """Pydantic base for DeepLabCut configuration models. + + This class is used to create configuration models for DeepLabCut. + It provides a base class for all configuration models that need YAML/dict I/O + and optional deprecated field names via ``json_schema_extra["aliases"]``. + (Use for all nested configs, e.g. pytorch ``DataConfig``, ``InferenceConfig``, etc.) + + For project-level schema migration and dirty-field tracking, subclass + `DLCVersionedConfig` instead. + + Features: + + - Strict schema (`extra="forbid"`, `validate_assignment=True`). + - Load and save: `from_yaml`, `from_dict`, `from_any`, `to_yaml`, `to_dict`. + - Pretty-print via `print`. + - Hooks: `_post_yaml_load_updates`. + - Nested dot-notation via `select`. + - Dict-like access over declared fields (legacy compatibility). + - In-place bulk updates via `update`. + - Field aliases from `json_schema_extra`. + """ + + model_config = ConfigDict(extra="forbid", validate_assignment=True) + + # ------------------------------------------------------------------ + # Validation (before pydantic field validation) + # ------------------------------------------------------------------ + + @model_validator(mode="before") + @classmethod + def resolve_aliases_before_validate(cls, data: Any) -> Any: + """Resolves aliases to their canonical names. (Normalizes ArgsKwargs + input to a dict for downstream validation.) + + Args: + data: Raw validator input (`dict`, `ArgsKwargs`, or other). + + Returns: + A dict with canonical field names when input is ArgsKwargs or dict; + otherwise `data` unchanged. + """ + if isinstance(data, ArgsKwargs): + data: dict = cls._args_kwargs_to_dict(data) + if isinstance(data, dict): + return resolve_aliases_in_dict(data, cls._alias_map(), target=cls.__name__) + return data + + # ------------------------------------------------------------------ + # Construction + # ------------------------------------------------------------------ + + @classmethod + def from_dict(cls, cfg_dict: dict) -> Self: + return cls.model_validate(cfg_dict) + + @classmethod + def from_any( + cls, + config: Self | dict | str | Path, + ) -> Self: + if isinstance(config, cls): + return config + elif isinstance(config, str | Path): + return cls.from_yaml(config) + elif isinstance(config, dict): + return cls.from_dict(config) + else: + raise TypeError( + "Failure to load configuration: Expected a config instance, " + f"dictionary, string, or Path. Got {type(config)}" + ) + + @classmethod + def from_yaml(cls, yaml_path: str | Path) -> Self: + yaml_dict = read_config_as_dict(yaml_path) + cfg = cls.from_dict(yaml_dict) + cfg._post_yaml_load_updates(yaml_path=Path(yaml_path)) + return cfg + + # ------------------------------------------------------------------ + # Serialization + # ------------------------------------------------------------------ + + def to_commented_map(self) -> CommentedMap: + """Recursively convert the config to a CommentedMap with YAML comments.""" + dumped = self.to_dict(normalize=True) + data = CommentedMap() + for name, info in type(self).model_fields.items(): + extra = info.json_schema_extra + if isinstance(extra, dict) and (comment := extra.get("comment")): + data.yaml_set_comment_before_after_key(name, before=comment) + value = getattr(self, name) + if isinstance(value, DLCBaseConfig): + data[name] = value.to_commented_map() + else: + data[name] = dumped[name] + return data + + def to_yaml( + self, + yaml_path: str | Path, + *, + overwrite: bool = True, + ) -> None: + write_config(yaml_path, self.to_commented_map(), overwrite=overwrite) + + def to_dict(self, *, normalize: bool = False) -> dict: + if not normalize: + return self.model_dump() + return normalize_for_serialization(self.model_dump()) + + def print( + self, + indent: int = 0, + print_fn: Callable[[str], None] | None = None, + ) -> None: + pretty_print(config=self.to_dict(), indent=indent, print_fn=print_fn) + + # ------------------------------------------------------------------ + # Hooks (override in subclasses) + # ------------------------------------------------------------------ + + def _post_yaml_load_updates(self, *, yaml_path: Path) -> None: + pass + + # ------------------------------------------------------------------ + # Field aliases (deprecated names in json_schema_extra) + # ------------------------------------------------------------------ + + @classmethod + @functools.cache + def _alias_map(cls) -> dict[str, str]: + """Build a map of deprecated aliases to canonical field names. + + Returns: + Dict mapping each alias in `json_schema_extra["aliases"]` to its + canonical field name. + + Raises: + ValueError: If the same alias is declared on more than one field. + """ + mapping: dict[str, str] = {} + for name, info in cls.model_fields.items(): + extra = info.json_schema_extra + if not isinstance(extra, dict): + continue + for alias in extra.get("aliases", []): + if alias in mapping: + raise ValueError(f"Duplicate alias '{alias}' for fields '{mapping[alias]}' and '{name}'") + mapping[alias] = name + return mapping + + def _resolve_alias( + self, + name: str, + *, + warn: bool = True, + stacklevel: int = 4, + ) -> str: + return resolve_alias(name, type(self)._alias_map(), warn=warn, stacklevel=stacklevel) + + # ------------------------------------------------------------------ + # Dict-like access (canonical field names only in keys()/iter) + # ------------------------------------------------------------------ + + def __setattr__(self, name: str, value: Any) -> None: + name = self._resolve_alias(name) + super().__setattr__(name, value) + + def __getattr__(self, name: str) -> Any: + # Only runs after normal lookup fails; try resolved alias or raise AttributeError via BaseModel.__getattr__. + if name in type(self)._alias_map(): + return getattr(self, self._resolve_alias(name)) + return super().__getattr__(name) + + def __getitem__(self, key: str) -> Any: + key = self._resolve_alias(key) + try: + return getattr(self, key) + except AttributeError: + raise KeyError(key) from None + + def __setitem__(self, key: str, value: Any) -> None: + canonical = self._resolve_alias(key, warn=True) + if canonical not in self._field_names(): + raise KeyError(f"'{type(self).__name__}' has no field '{key}'") + setattr(self, canonical, value) + + def __contains__(self, key: object) -> bool: + if not isinstance(key, str): + return False + if key in self._field_names(): + return True + return key in type(self)._alias_map() + + def __iter__(self) -> Iterator[str]: + return iter(self._field_names()) + + def __len__(self) -> int: + return len(self._field_names()) + + def get(self, key: str, default: Any = None) -> Any: + try: + return self[key] + except KeyError: + return default + + def update( + self, + updates: dict[str, Any] | None = None, + /, + **kwargs: Any, + ) -> Self: + if updates is not None and kwargs: + raise TypeError(f"{type(self).__name__}.update() accepts either a dict or keyword args, not both.") + overrides = updates if updates is not None else kwargs + # Resolve aliases together (checks for duplicate canonical field names) + resolved = resolve_aliases_in_dict(overrides, type(self)._alias_map(), target=type(self).__name__) + for name, value in resolved.items(): + setattr(self, name, value) + return self + + def keys(self) -> list[str]: + return self._field_names() + + def values(self) -> list[Any]: + return [getattr(self, name) for name in self._field_names()] + + def items(self) -> list[tuple[str, Any]]: + return [(name, getattr(self, name)) for name in self._field_names()] + + def select(self, path: str, default: Any = None) -> Any: + """Select a value from the config using dot notation for nested keys.""" + try: + return self._navigate_nested(fields=path.split(".")) + except (AttributeError, KeyError, TypeError): + return default + + def set_nested(self, path: str, value: Any) -> Self: + if "." not in path: + setattr(self, path, value) + return self + parts = path.split(".") + parent_fields, final_field = parts[:-1], parts[-1] + try: + parent = self._navigate_nested(parent_fields) + parent[final_field] = value + except (AttributeError, KeyError, TypeError) as e: + raise AttributeError(f"{type(self).__name__} has no '{path}'") from e + return self + + def _navigate_nested(self, fields: list[str]) -> Any: + """Navigate a nested structure of fields. Raises AttributeError / KeyError if any field is not found.""" + obj: Any = self + for field in fields: + obj = obj[field] if isinstance(obj, dict) else getattr(obj, field) + return obj + + def _field_names(self) -> list[str]: + cls = type(self) + if not isinstance(self, BaseModel): + raise TypeError(f"{cls.__name__} must inherit from pydantic.BaseModel") + return list(cls.model_fields.keys()) + + @classmethod + def _args_kwargs_to_dict(cls, data: ArgsKwargs) -> dict: + """Map positional and keyword constructor args to a field-name dict.""" + names = list(cls.model_fields.keys()) + return dict( + zip(names, data.args or [], strict=False), + **(data.kwargs or {}), + ) + + +class DLCVersionedConfig(DLCBaseConfig): + """Top-level configs with schema migration and change tracking. + + Subclass of `DLCBaseConfig` for project and pose YAML configs such as + `ProjectConfig` and `PoseConfig`. + + Note: + Pydantic runs `migrate_before_validate` before the base + `resolve_aliases_before_validate` (child-first order): schema migration + on legacy keys, then alias resolution for the current model. + + Additional behavior: + + - `migrate_before_validate` upgrades raw dicts to `CURRENT_CONFIG_VERSION`. + - Tracks fields modified after load; `to_yaml` can log changes and mark clean. + - Patches `__setattr__` once per class to record dirty fields while delegating + alias warnings to the base `__setattr__`. + """ + + config_version: int = Field( + default=versioning.CURRENT_CONFIG_VERSION, + json_schema_extra={"comment": "Config schema version. Do not edit manually."}, + ) + + _initialized: bool = PrivateAttr(default=False) + _dirty_fields: set[str] = PrivateAttr(default_factory=set) + _change_notes: dict[str, Any] = PrivateAttr(default_factory=dict) + + # ------------------------------------------------------------------ + # Version migration (before pydantic field validation) + # ------------------------------------------------------------------ + + @classmethod + def from_dict(cls, cfg_dict: dict) -> Self: + cfg_dict = versioning.migrate_config( + cfg_dict, + config_type=cls.__name__, + target_version=versioning.CURRENT_CONFIG_VERSION, + ) + return super().from_dict(cfg_dict) + + # ------------------------------------------------------------------ + # Serialization + # ------------------------------------------------------------------ + + def to_yaml( + self, + yaml_path: str | Path, + *, + overwrite: bool = True, + log_changes: bool = True, + mark_clean: bool = True, + ) -> None: + super().to_yaml(yaml_path, overwrite=overwrite) + if log_changes: + self.log_changes() + if mark_clean: + self.mark_clean() + + # ------------------------------------------------------------------ + # Change tracking + # ------------------------------------------------------------------ + + def model_post_init(self, __context: Any) -> None: + super().model_post_init(__context) + self._initialized = True + + def __setattr__(self, name: str, value: Any) -> None: + name = self._resolve_alias(name) + + # Private attributes (not a model field) skip tracking logic + if name not in type(self).model_fields or not self._initialized: + super().__setattr__(name, value) + return + + # Get the coerced values before and after setting; log changes + old_value = getattr(self, name, None) + super().__setattr__(name, value) + new_value = getattr(self, name) + if old_value != new_value: + self._dirty_fields.add(name) + + @property + def is_dirty(self) -> bool: + return bool(self._dirty_fields) + + @property + def dirty_fields(self) -> frozenset[str]: + return frozenset(self._dirty_fields) + + @property + def change_notes(self) -> list[str]: + return list(self._change_notes.values()) + + def record_change_note( + self, + field_name: str, + message: str, + *, + include_caller: bool = False, + _stack_depth: int = 1, + ) -> None: + field_name = self._resolve_alias(field_name) + + if field_name not in type(self).model_fields: + raise KeyError(f"'{type(self).__name__}' has no field '{field_name}'") + + if include_caller: + frame = sys._getframe(_stack_depth) + filename = frame.f_code.co_filename.rsplit("/", 1)[-1] + message = f"{message} [{filename}:{frame.f_lineno}]" + + self._change_notes[field_name] = message + + def log_changes(self) -> None: + if not self.is_dirty: + return + logger.info(f"Updates to {type(self).__name__}:") + for field_name in sorted(self._dirty_fields): + if field_name in self._change_notes: + logger.info(f" {self._change_notes[field_name]}") + else: + logger.info(f" {field_name} was modified") + + def mark_clean(self) -> None: + self._dirty_fields.clear() + self._change_notes.clear() diff --git a/dlclibrary/io/config/project_config.py b/dlclibrary/io/config/project_config.py new file mode 100644 index 0000000..d50127b --- /dev/null +++ b/dlclibrary/io/config/project_config.py @@ -0,0 +1,305 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Project configuration classes for DeepLabCut pose estimation models.""" + +import logging +from pathlib import Path +from typing import Any, Literal, get_args + +from pydantic import Field, model_validator +from typing_extensions import Self + +from dlclibrary.io.config.base_config import DLCVersionedConfig +from dlclibrary.io.config.validation import ( + BodypartPair, + Fraction, + NonNegativeInt, + StrictPositiveInt, + UniqueStrList, + less_than, + validate_crop_bounds, +) + +logger = logging.getLogger(__name__) + + +class ProjectConfig(DLCVersionedConfig): + """Complete project configuration. + + Mirrors the structure of the project config.yaml (and metadata in pose config). + Field names match the old dictionary keys for round-trip compatibility. + + Attributes: + Task: Project task identifier (do not edit). + scorer: Scorer name (do not edit). + date: Project date (do not edit). + multianimalproject: Whether the project is multi-animal. + identity: Whether identity tracking is enabled (project config.yaml key). + project_path: Path to the DeepLabCut project. + pose_config_path: Path to the pose configuration file (metadata only). + engine: Default DeepLabCut engine (e.g. pytorch). + video_sets: Video set configuration. + bodyparts: List of body parts. + individuals: List of individual animal identities (multi-animal). + uniquebodyparts: List of unique body parts (multi-animal project key). + multianimalbodyparts: List of multi-animal body parts (multi-animal key). + start: Fraction of video to start extracting frames. + stop: Fraction of video to stop extracting frames. + numframes2pick: Number of frames to pick for labeling. + skeleton: Skeleton connectivity for plotting. + skeleton_color: Skeleton color for plotting. + pcutoff: Confidence cutoff for plotting. + dotsize: Dot size for visualization. + alphavalue: Alpha value for visualization. + colormap: Colormap for visualization. + TrainingFraction: Training fractions for dataset splits. + iteration: Training iteration. + default_net_type: Default network architecture. + default_augmenter: Default data augmenter. + default_track_method: Default tracking method. + snapshotindex: Snapshot index for evaluation. + detector_snapshotindex: Detector snapshot index. + batch_size: Training batch size. + detector_batch_size: Detector batch size. + cropping: Whether cropping is enabled for analysis. + x1: Cropping x1 coordinate. + x2: Cropping x2 coordinate. + y1: Cropping y1 coordinate. + y2: Cropping y2 coordinate. + corner2move2: Refinement corner configuration. + move2corner: Refinement move-to-corner flag. + SuperAnimalConversionTables: Conversion tables for SuperAnimal weights. + """ + + # Project definitions (do not edit) + Task: str = Field(default="", json_schema_extra={"comment": "Project definitions (do not edit)"}) + scorer: str = "" + date: str = "" + multianimalproject: bool = False + identity: bool | None = Field(default=None, json_schema_extra={"aliases": ["with_identity"]}) + + # Project path + project_path: Path = Field( + default_factory=Path, + json_schema_extra={"comment": "\nProject path (change when moving around)"}, + ) + pose_config_path: Path | None = None + + # Engine + engine: Literal["pytorch", "tensorflow"] = Field( + default="pytorch", + json_schema_extra={ + "comment": "\nDefault DeepLabCut engine to use for shuffle creation (either pytorch or tensorflow)" + }, + ) + + # Annotation dataset configuration (and individual video cropping parameters) + video_sets: dict[str, Any] = Field( + default_factory=dict, + json_schema_extra={"comment": "\nAnnotation data set configuration (and individual video cropping parameters)"}, + ) + # VV TODO @deruyter92 2026-01-30: following the old original config.yaml template for now. VV + # VV We should change this to a list[str] in the future. VV + bodyparts: UniqueStrList | Literal["MULTI!"] = Field(default_factory=list) + + # TODO @deruyter92 2026-02-06: The current pipeline requires at least one individual defined in the + # default configuration. This will be removed in the future. + individuals: UniqueStrList = Field(default_factory=lambda: ["individual_1"]) + uniquebodyparts: UniqueStrList = Field(default_factory=list, json_schema_extra={"aliases": ["unique_bodyparts"]}) + multianimalbodyparts: UniqueStrList = Field(default_factory=list) # multi-animal project key + + # Fraction of video to start/stop when extracting frames for labeling/refinement + start: Fraction = Field( + default=0.0, + json_schema_extra={ + "comment": "\nFraction of video to start/stop when extracting frames for labeling/refinement" + }, + ) + stop: Fraction = 1.0 + numframes2pick: NonNegativeInt = 20 + + # Plotting configuration + skeleton: list[BodypartPair] = Field( + default_factory=list, + json_schema_extra={"comment": "\nPlotting configuration"}, + ) + skeleton_color: str = "black" + pcutoff: Fraction = 0.6 + dotsize: NonNegativeInt = 12 + alphavalue: Fraction = 0.7 + colormap: str = "rainbow" + + # Training, evaluation and analysis configuration + TrainingFraction: list[Fraction] = Field( + default_factory=lambda: [0.95], + json_schema_extra={"comment": "\nTraining,Evaluation and Analysis configuration"}, + ) + iteration: NonNegativeInt | None = None + default_net_type: str = "resnet_50" + default_augmenter: str | None = None + default_track_method: str | None = None + snapshotindex: Literal["all"] | int = "all" + detector_snapshotindex: int = -1 + batch_size: StrictPositiveInt = 8 + detector_batch_size: StrictPositiveInt = 1 + + # Cropping parameters (for analysis and outlier frame detection) + cropping: bool = Field( + default=False, + json_schema_extra={"comment": "\nCropping Parameters (for analysis and outlier frame detection)"}, + ) + x1: NonNegativeInt | None = Field( + default=None, + json_schema_extra={"comment": "if cropping is true for analysis, then set the values here:"}, + ) + x2: NonNegativeInt | None = None + y1: NonNegativeInt | None = None + y2: NonNegativeInt | None = None + + # Refinement configuration (parameters from annotation dataset configuration also relevant in this stage) + corner2move2: list[NonNegativeInt] | None = Field( + default=None, + json_schema_extra={ + "comment": ( + "\nRefinement configuration (parameters from annotation dataset " + "configuration also relevant in this stage)" + ) + }, + ) + move2corner: bool | None = None + + # Conversion tables to fine-tune SuperAnimal weights + SuperAnimalConversionTables: dict[str, Any] | None = Field( + default=None, + json_schema_extra={"comment": "\nConversion tables to fine-tune SuperAnimal weights"}, + ) + + # TODO @deruyter92 2026-02-06: These parameters are no longer used in the new pipeline. + # We should remove them in config schema v1. They are needed now to support reading old configs. + resnet: int | None = Field( + default=None, + json_schema_extra={ + "comment": "\nThese are very old parameters that are no longer used They are simply ignored." + }, + ) + croppedtraining: bool | None = None + + # TODO @deruyter92 2026-09-01: This should be included as field in the + # config. It requires a version bump of the schema version. + # see https://github.com/DeepLabCut/DeepLabCut/pull/3470 + @property + def bboxes_pcutoff(self) -> Fraction: + """Default confidence cutoff for plotting bounding boxes from detectors. + + Note: This property is introduced in v3.1.0 as prospective config + field, but is intentionally not yet defined as such in the schema to + avoid breaking compatibility with older configs. + """ + return 0.6 + + @property + def bodyparts_list(self) -> list[str]: + # Animal-count agnostic; Always return a list (never "MULTI!", None, etc.) + if self.multianimalproject: + return list(self.multianimalbodyparts) + if self.bodyparts == "MULTI!": + raise ValueError("bodyparts must be a list of strings when multianimalproject is False, got 'MULTI!'") + return list(self.bodyparts) + + @property + def config_yaml_path(self) -> Path: + return self.project_path / "config.yaml" + + def validate_project_path(self, *, yaml_path: str | Path | None = None, write: bool = False) -> Self: + """Repair project_path from yaml location; optionally persist.""" + path = Path(yaml_path) if yaml_path is not None else self.config_yaml_path + if not path.is_file(): + raise FileNotFoundError(f"config.yaml not found: {path}") + self._post_yaml_load_updates(yaml_path=path) + if write and "project_path" in self.dirty_fields: + self.to_yaml(path, log_changes=True, mark_clean=True) + return self + + @classmethod + def from_any(cls, config: Self | dict | str | Path, *, repair_path: bool = False) -> Self: + cfg = super().from_any(config) + if repair_path: + cfg.validate_project_path(yaml_path=config if isinstance(config, (str, Path)) else None, write=True) + return cfg + + def _post_yaml_load_updates(self, *, yaml_path: Path) -> None: + """ + Override method for post-yaml load updates. Called automatically by from_yaml(). + These are logged but not written to disk -- call to_yaml() explicitly if needed. + """ + project_path = yaml_path.parent + if project_path.absolute() != self.project_path.absolute(): + old = self.project_path + self.project_path = project_path + self.record_change_note( + "project_path", + f"project_path updated: {old} -> {project_path} (resolved from YAML location when reading config.yaml)", + ) + + @model_validator(mode="before") + @classmethod + def normalize_legacy_empty_values(cls, data: Any) -> Any: + if not isinstance(data, dict): + return data + data = dict(data) + + # Some old configs used empty strings for unset fields + for fieldname in ("skeleton", "TrainingFraction", "video_sets", "bodyparts"): + if data.get(fieldname) == "": + data.pop(fieldname) + + # Support for legacy config.yaml templates that use bare (null) keys as placeholders + # for unset fields. Only fields whose type doesn't already accept None are affected. + for name, value in list(data.items()): + if value is not None or name not in cls.model_fields: + continue + none_is_invalid = type(None) not in get_args(cls.model_fields[name].annotation) + if none_is_invalid: + logger.warning( + f"Found invalid empty/null/None value for `{name}` in the project " + "config. This is only supported for legacy compatibility. " + f"Treating `{name}` as unset and using the field default instead." + ) + data.pop(name) + + # NOTE @deruyter92 2026-06-15: This should be removed in v1. + if data.get("multianimalproject") and not data.get("bodyparts"): + data["bodyparts"] = "MULTI!" + + return data + + @model_validator(mode="after") + def validate_start_before_stop(self) -> Self: + less_than(self.start, self.stop, name="start", threshold_name="stop") + return self + + @model_validator(mode="after") + def validate_bodyparts_single_animal(self) -> Self: + if not self.multianimalproject and self.bodyparts == "MULTI!": + raise ValueError("bodyparts must be a list of strings when multianimalproject is False, got 'MULTI!'") + + # TODO @deruyter92 2026-06-15: This sentinel should be removed in v1. + elif self.multianimalproject and self.bodyparts != "MULTI!": + raise ValueError(f"bodyparts must be 'MULTI!' when multianimalproject is True, got {self.bodyparts}") + + return self + + @model_validator(mode="after") + def validate_cropping_bounds(self) -> Self: + if self.cropping and any(value is None for value in (self.x1, self.x2, self.y1, self.y2)): + raise ValueError("When cropping is enabled, x1, x2, y1, and y2 must all be set") + validate_crop_bounds(x1=self.x1, x2=self.x2, y1=self.y1, y2=self.y2) + return self diff --git a/dlclibrary/io/config/utils.py b/dlclibrary/io/config/utils.py new file mode 100644 index 0000000..b7466c4 --- /dev/null +++ b/dlclibrary/io/config/utils.py @@ -0,0 +1,196 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Centralized helpers for reading, writing, and creating configuration files (YAML).""" + +from __future__ import annotations + +import logging +import warnings +from collections import Counter +from collections.abc import Callable, Mapping +from enum import Enum +from pathlib import Path, PurePath +from typing import Any + +import numpy as np +from ruamel.yaml import YAML + +logger = logging.getLogger(__name__) + + +def get_yaml_loader() -> YAML: + """Get a ruamel.yaml YAML handler with safe mode.""" + yaml = YAML(typ="safe", pure=True) + return yaml + + +def get_yaml_dumper() -> YAML: + """Get a ruamel.yaml YAML handler with representers for Enum and Path objects.""" + yaml = YAML(typ="rt", pure=True) + + # Use a very large width so long strings (e.g., file paths or keys with spaces) + # are kept on a single line instead of being wrapped, which can otherwise cause + # them to be emitted as complex keys. See also: + # https://stackoverflow.com/questions/31197268/pyyaml-yaml-dump-produces-complex-key-for-string-key-122-chars/31199123#31199123 + # See PR https://github.com/DeepLabCut/DeepLabCut/pull/3140 for more details. + yaml.width = 1_000_000 + + # Auto-serialize Path objects as strings + yaml.representer.add_multi_representer(PurePath, lambda r, p: r.represent_str(str(p))) + yaml.representer.add_multi_representer(Enum, lambda r, e: r.represent_str(e.value)) + return yaml + + +def read_config_as_dict(config_path: str | Path) -> dict: + """ + Args: + config_path: the path to the configuration file to load + + Returns: + The configuration file with pure Python classes + + Raises: + FileNotFoundError: if the config file does not exist + """ + if not Path(config_path).exists(): + raise FileNotFoundError(f"Config {config_path} is not found. Please make sure that the file exists.") + with open(config_path) as f: + cfg = get_yaml_loader().load(f) + if cfg is None: + raise ValueError(f"Config {config_path} is empty or null.") + if not isinstance(cfg, dict): + raise ValueError(f"Config {config_path} must be a YAML mapping at the top level, got {type(cfg).__name__}.") + return cfg + + +def write_config(config_path: str | Path, config: dict, overwrite: bool = True) -> None: + """Writes a pose configuration file to disk. + + Args: + config_path: the path where the config should be saved + config: the config to save + overwrite: whether to overwrite the file if it already exists + + Raises: + FileExistsError if overwrite=True and the file already exists + """ + if not overwrite and Path(config_path).exists(): + raise FileExistsError(f"Cannot write to {config_path} - set overwrite=True to force") + + with open(config_path, "w") as file: + get_yaml_dumper().dump(config, file) + + +def resolve_alias( + name: str, + alias_map: dict[str, str], + *, + warn: bool = True, + stacklevel: int = 3, +) -> str: + """Resolve a config key to its canonical field name. + Args: + name: Raw key name (alias or canonical). + alias_map: ``{alias: canonical_name}`` for deprecated keys. + warn: If True, emit :class:`DLCDeprecationWarning` when ``name`` is an alias. + stacklevel: Passed to :func:`warnings.warn` for deprecation messages. + Returns: + Canonical field name, or ``name`` unchanged if it is not an alias. + """ + canonical = alias_map.get(name, name) + if warn and name in alias_map: + from dlclibrary.deprecation import DLCDeprecationWarning + + warnings.warn( + f"'{name}' is deprecated, use '{canonical}' instead.", + DLCDeprecationWarning, + stacklevel=stacklevel, + ) + return canonical + + +def resolve_aliases_in_dict( + cfg_dict: dict, + alias_map: dict[str, str], + *, + target: str = "config", + warn: bool = True, + stacklevel: int = 3, +) -> dict: + """Rename deprecated config keys to their canonical names. + + Args: + cfg_dict: Raw configuration mapping (e.g. from YAML). + alias_map: ``{alias: canonical_name}`` for deprecated keys. + target: Config class name shown in errors. + stacklevel: Passed to :func:`warnings.warn` for deprecation messages. + + Returns: + A new dict with alias keys replaced by canonical names. Unchanged if + ``alias_map`` is empty. + + Raises: + TypeError: If multiple keys resolve to the same canonical field name + (e.g. an alias and its canonical name, or two aliases for one field). + """ + if not alias_map: + return cfg_dict + + def _raise_for_duplicates(raw_to_canonical: dict[str, str]): + counts = Counter(raw_to_canonical.values()) + conflicts = [f"{raw} -> {canonical}" for raw, canonical in raw_to_canonical.items() if counts[canonical] > 1] + if conflicts: + raise TypeError(f"{target} received duplicate canonical field names: {', '.join(conflicts)}.") + + raw_to_canonical = {raw: resolve_alias(raw, alias_map, warn=warn, stacklevel=stacklevel + 1) for raw in cfg_dict} + _raise_for_duplicates(raw_to_canonical) + return {raw_to_canonical[raw]: v for raw, v in cfg_dict.items()} + + +def normalize_for_serialization(obj: Any) -> Any: + """Recursively normalize Paths to strings and Enums to values.""" + if isinstance(obj, Path): + return str(obj) + if isinstance(obj, Enum): + return obj.value + if isinstance(obj, Mapping): + return type(obj)({k: normalize_for_serialization(v) for k, v in obj.items()}) + if isinstance(obj, tuple): + return tuple(normalize_for_serialization(v) for v in obj) + if isinstance(obj, (list, set)): + return [normalize_for_serialization(v) for v in obj] + if isinstance(obj, np.ndarray): + return obj.tolist() + return obj + + +def pretty_print( + config: dict, + indent: int = 0, + print_fn: Callable[[str], None] | None = None, +) -> None: + """Prints a model configuration in a pretty and readable way. + + Args: + config: the config to print + indent: the base indent on all keys + print_fn: custom function to call (simply calls ``print`` if None) + """ + if print_fn is None: + print_fn = print + + for k, v in config.items(): + if isinstance(v, dict): + print_fn(f"{indent * ' '}{k}:") + pretty_print(v, indent + 2, print_fn=print_fn) + else: + print_fn(f"{indent * ' '}{k}: {v}") + diff --git a/dlclibrary/io/config/validation.py b/dlclibrary/io/config/validation.py new file mode 100644 index 0000000..61df0ed --- /dev/null +++ b/dlclibrary/io/config/validation.py @@ -0,0 +1,105 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +from collections.abc import Sequence +from typing import Annotated, Any, TypeAlias, TypeVar + +import numpy as np +from numpy.typing import NDArray +from pydantic import AfterValidator, BeforeValidator, Field, GetPydanticSchema, InstanceOf +from pydantic_core import PydanticUseDefault + + +def _describe(value: float, name: str | None = None) -> str: + return f"{name} ({value})" if name else f"{value}" + + +def greater_than( + value: float, + threshold: float, + name: str | None = None, + threshold_name: str | None = None, +) -> None: + if value <= threshold: + raise ValueError(f"{_describe(value, name)} must be greater than {_describe(threshold, threshold_name)}") + + +def less_than( + value: float, + threshold: float, + name: str, + threshold_name: str | None = None, +) -> None: + if value >= threshold: + raise ValueError(f"{_describe(value, name)} must be less than {_describe(threshold, threshold_name)}") + + +def unique_values(values: Sequence[Any]) -> Sequence[Any]: + if len(values) != len(set(values)): + raise ValueError("Values must be unique") + return values + + +def validate_crop_bounds( + *, + x1: int | None, + x2: int | None, + y1: int | None, + y2: int | None, +) -> None: + bounds = (x1, x2, y1, y2) + if any(value is None for value in bounds): + if any(value is not None for value in bounds): + raise ValueError("Crop bounds x1, x2, y1, and y2 must either all be set or all be omitted") + return + + less_than(x1, x2, name="x1", threshold_name="x2") + less_than(y1, y2, name="y1", threshold_name="y2") + + +def _coerce_ndarray(v): + if isinstance(v, np.ndarray): + return v + return np.asarray(v, dtype=int) + + +def _bodypart_pair(values: Sequence[Any]) -> list[str]: + if len(values) != 2: + raise ValueError(f"Each bodypart pair must contain exactly two bodyparts, got {len(values)}") + return list(unique_values(values)) + + +def _use_default_if_none(value: Any) -> Any: + """Fall back to the field's declared default when the input value is None.""" + if value is None: + raise PydanticUseDefault + return value + + +T = TypeVar("T") + +# Normalize an explicit ``null`` to the field's default, +# e.g. ``box_score_thresh: DefaultIfNone[Fraction] = 0.01`` +DefaultIfNone: TypeAlias = Annotated[T, BeforeValidator(_use_default_if_none)] + +Fraction = Annotated[float, Field(ge=0.0, le=1.0)] +UniqueStrList = Annotated[list[str], AfterValidator(unique_values)] +NonNegativeFloat = Annotated[float, Field(ge=0.0)] +NonNegativeInt = Annotated[int, Field(ge=0)] +StrictPositiveInt = Annotated[int, Field(ge=1)] +NDArrayInt = Annotated[ + NDArray, + BeforeValidator(_coerce_ndarray), + GetPydanticSchema( + lambda _s, h: h(InstanceOf[np.ndarray]), + lambda _s, h: h(InstanceOf[np.ndarray]), + ), +] +BodypartPair = Annotated[list[str], AfterValidator(_bodypart_pair)] diff --git a/dlclibrary/io/config/versioning.py b/dlclibrary/io/config/versioning.py new file mode 100644 index 0000000..3cb5a52 --- /dev/null +++ b/dlclibrary/io/config/versioning.py @@ -0,0 +1,264 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Configuration migration system for handling version upgrades and downgrades. + +This module provides a versioned migration system that allows configurations +to be upgraded from older versions to newer ones, or downgraded to older formats. +Upgrade migrations are chained together, so any version can be upgraded to the +latest by applying all intermediate migrations in sequence. Downgrade migrations +can be registered for specific version pairs when backward compatibility is needed. +""" + +import copy +import logging +from collections.abc import Callable +from functools import wraps + +logger = logging.getLogger(__name__) + +# Current configuration schema version +# Increment this when making breaking changes to the config structure +CURRENT_CONFIG_VERSION = 0 + + +# Version registry: maps (config_type, from_version, to_version) -> migration function. +# config_type is the class name of the DLCVersionedConfig subclass the migration applies to +# (e.g. "ProjectConfig", "PoseConfig"). Every migration must declare its target type. +_MIGRATIONS: dict[tuple[str, int, int], Callable[[dict], dict]] = {} + + +def _diff_dicts( + before: dict, + after: dict, + path: str = "", + ignore_keys: set[str] | None = None, +) -> list[str]: + """Recursively diff two dicts and return human-readable change descriptions. + + Args: + before: The dict before the migration step. + after: The dict after the migration step. + path: Dot-separated key path for nested context (used in recursion). + ignore_keys: Top-level keys to skip (only applied at the root level). + + Returns: + A list of strings, each describing a single added/removed/updated field. + """ + ignore = ignore_keys or set() + before_keys = set(before.keys()) - ignore + after_keys = set(after.keys()) - ignore + + changes: list[str] = [] + + for key in sorted(before_keys - after_keys): + full = f"{path}.{key}" if path else str(key) + changes.append(f"Removed field '{full}' (was: {before[key]!r})") + + for key in sorted(after_keys - before_keys): + full = f"{path}.{key}" if path else str(key) + changes.append(f"Added field '{full}' = {after[key]!r}") + + for key in sorted(before_keys & after_keys): + old_val, new_val = before[key], after[key] + if old_val == new_val: + continue + full = f"{path}.{key}" if path else str(key) + if isinstance(old_val, dict) and isinstance(new_val, dict): + changes.extend(_diff_dicts(old_val, new_val, path=full)) + else: + changes.append(f"Updated field '{full}': {old_val!r} -> {new_val!r}") + + return changes + + +def _log_field_changes(before: dict, after: dict, from_v: int, to_v: int) -> None: + """Log which fields were added, removed, or changed during a migration step.""" + step = f"v{from_v} -> v{to_v}" + changes = _diff_dicts(before, after, ignore_keys={"config_version"}) + + if changes: + for change in changes: + logger.debug(" [%s] %s", step, change) + else: + logger.debug(" [%s] No field changes", step) + + +def register_migration( + from_version: int, + to_version: int, + config_type: str, +): + """Decorator to register a migration function for a specific config type. + + Every migration must be scoped to a concrete ``DLCVersionedConfig`` subclass. + This keeps ``ProjectConfig`` and ``PoseConfig`` migration chains fully independent. + + Args: + from_version: The source version number (>= 0). + to_version: The target version number (>= 0, from_version ± 1 by convention). + config_type: Class name of the config this migration applies to (e.g. + ``"ProjectConfig"`` or ``"PoseConfig"``). + + Raises: + ValueError: If version numbers are invalid or a migration for the same + (config_type, from_version, to_version) triple is already registered. + + Example:: + + @register_migration(0, 1, config_type="ProjectConfig") + def migrate_project_v0_to_v1(config: dict) -> dict: + config["unique_bodyparts"] = config.pop("uniquebodyparts", []) + return config + """ + if from_version < 0 or to_version < 0: + raise ValueError(f"Version numbers must be non-negative, got ({from_version}, {to_version})") + if from_version == to_version: + raise ValueError(f"from_version and to_version must differ, got ({from_version}, {to_version})") + + def decorator(func: Callable[[dict], dict]) -> Callable[[dict], dict]: + key = (config_type, from_version, to_version) + if key in _MIGRATIONS: + raise ValueError( + f"Duplicate migration registered for {key}. " + f"Existing: {_MIGRATIONS[key].__wrapped__.__qualname__}, " + f"new: {func.__qualname__}" + ) + + @wraps(func) + def wrapper(config: dict) -> dict: + verbose = logger.isEnabledFor(logging.DEBUG) + if verbose: + before = copy.deepcopy(config) + result = func(config.copy()) # Don't mutate caller's dict + result["config_version"] = to_version + if verbose: + _log_field_changes(before, result, from_version, to_version) + return result + + _MIGRATIONS[key] = wrapper + return wrapper + + return decorator + + +def get_config_version(config: dict) -> int: + """Extract the configuration version from a config dict. + + Args: + config: Configuration dictionary + + Returns: + Version number (0 for legacy/unversioned configs) + """ + return config.get("config_version", 0) + + +def migrate_config( + config: dict, + config_type: str, + target_version: int = CURRENT_CONFIG_VERSION, +) -> dict: + """Migrate a configuration to the target version. + + Applies all necessary migrations in sequence to upgrade the config + from its current version to the target version. Only migrations registered + for ``config_type`` are applied. + + Args: + config: Configuration dictionary to migrate. + config_type: Class name of the config being migrated (e.g. ``"ProjectConfig"``). + Only migrations registered for this type are applied. + target_version: Target version to migrate to (default: current). + + Returns: + Migrated configuration dictionary. + + Raises: + ValueError: If migration chain is incomplete or target version is invalid. + """ + current_version = get_config_version(config) + + if current_version == target_version: + return config + + if target_version > CURRENT_CONFIG_VERSION: + raise ValueError(f"Target version {target_version} exceeds current version {CURRENT_CONFIG_VERSION}") + + direction = "upgrade" if target_version > current_version else "downgrade" + logger.info( + "Migrating %s from version %d to %d (%s)", + config_type, + current_version, + target_version, + direction, + ) + + # Chain migrations one step at a time (e.g. 0→1→2 or 3→2→1). + migrated = config + step = 1 if target_version > current_version else -1 + for v in range(current_version, target_version, step): + next_v = v + step + key = (config_type, v, next_v) + if key not in _MIGRATIONS: + raise ValueError( + f"No migration registered for '{config_type}' v{v} -> v{next_v}. " + f"Available migrations: {list(_MIGRATIONS.keys())}" + ) + logger.debug( + "Applying migration %s v%d -> v%d (%s)", + config_type, + v, + next_v, + _MIGRATIONS[key].__wrapped__.__qualname__, + ) + try: + migrated = _MIGRATIONS[key](migrated) + except Exception as exc: + raise type(exc)( + f"Migration for '{config_type}' v{v} -> v{next_v} failed " + f"({_MIGRATIONS[key].__wrapped__.__qualname__}): {exc}" + ) from exc + + logger.info("Migration complete: %s is now at version %d", config_type, target_version) + return migrated + + +# ============================================================================ +# Migration Definitions +# ============================================================================ + + +@register_migration(0, 1, config_type="ProjectConfig") +def migrate_project_v0_to_v1(config: dict) -> dict: + """Migrate ProjectConfig from unversioned/legacy (v0) to v1.""" + # Migration logic goes here, when migrating to v1. + return config + + +@register_migration(1, 0, config_type="ProjectConfig") +def migrate_project_v1_to_v0(config: dict) -> dict: + """Migrate ProjectConfig from v1 back to v0 (legacy format).""" + # Migration logic goes here, when migrating to v1. + return config + + +@register_migration(0, 1, config_type="PoseConfig") +def migrate_pose_v0_to_v1(config: dict) -> dict: + """Migrate PoseConfig from v0 to v1.""" + # Migration logic goes here, when migrating to v1. + return config + + +@register_migration(1, 0, config_type="PoseConfig") +def migrate_pose_v1_to_v0(config: dict) -> dict: + """Migrate PoseConfig from v1 back to v0.""" + # Migration logic goes here, when migrating to v1. + return config From 2c17ca163b37f85ae93153fe7c740913fa81e3ca Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 15 Oct 2026 10:17:20 +0200 Subject: [PATCH 3/5] Move core config tests --- tests/io/config/conftest.py | 48 + .../openfield-Pranav-2018-10-30/config.yaml | 76 ++ tests/io/config/test_base_config.py | 496 ++++++++++ tests/io/config/test_change_tracking.py | 336 +++++++ tests/io/config/test_config_breakage.py | 292 ++++++ tests/io/config/test_core_config.py | 223 +++++ tests/io/config/test_owed.py | 30 + tests/io/config/test_project_config.py | 337 +++++++ tests/io/config/test_validation.py | 220 +++++ tests/io/config/test_versioning.py | 921 ++++++++++++++++++ 10 files changed, 2979 insertions(+) create mode 100644 tests/io/config/conftest.py create mode 100644 tests/io/config/fixtures/openfield-Pranav-2018-10-30/config.yaml create mode 100644 tests/io/config/test_base_config.py create mode 100644 tests/io/config/test_change_tracking.py create mode 100644 tests/io/config/test_config_breakage.py create mode 100644 tests/io/config/test_core_config.py create mode 100644 tests/io/config/test_owed.py create mode 100644 tests/io/config/test_project_config.py create mode 100644 tests/io/config/test_validation.py create mode 100644 tests/io/config/test_versioning.py diff --git a/tests/io/config/conftest.py b/tests/io/config/conftest.py new file mode 100644 index 0000000..a5f54e0 --- /dev/null +++ b/tests/io/config/conftest.py @@ -0,0 +1,48 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Shared fixtures for config tests (no import-time migration registration).""" + +import pytest + +from dlclibrary.io.config import versioning +from dlclibrary.io.config.versioning import register_migration + +_TOY_VERSION_OLD = 98 +_TOY_VERSION_NEW = 99 +_LEGACY_FIELD = "toy_legacy_field" +_NEW_FIELD = "toy_new_field" + + +@pytest.fixture(autouse=True) +def isolated_migration_registry(monkeypatch): + """Snapshot the migration registry per test; production migrations only at baseline.""" + monkeypatch.setattr(versioning, "_MIGRATIONS", versioning._MIGRATIONS.copy()) + + +@pytest.fixture +def register_toy_migrations(monkeypatch): + """Register v98 <-> v99 toy migrations for the given config_type(s).""" + monkeypatch.setattr(versioning, "CURRENT_CONFIG_VERSION", _TOY_VERSION_NEW) + + def register_pair(config_type: str) -> None: + @register_migration(_TOY_VERSION_OLD, _TOY_VERSION_NEW, config_type=config_type) + def up(config: dict) -> dict: + if _LEGACY_FIELD in config: + config[_NEW_FIELD] = config.pop(_LEGACY_FIELD) + return config + + @register_migration(_TOY_VERSION_NEW, _TOY_VERSION_OLD, config_type=config_type) + def down(config: dict) -> dict: + if _NEW_FIELD in config: + config[_LEGACY_FIELD] = config.pop(_NEW_FIELD) + return config + + return register_pair diff --git a/tests/io/config/fixtures/openfield-Pranav-2018-10-30/config.yaml b/tests/io/config/fixtures/openfield-Pranav-2018-10-30/config.yaml new file mode 100644 index 0000000..ed8c31f --- /dev/null +++ b/tests/io/config/fixtures/openfield-Pranav-2018-10-30/config.yaml @@ -0,0 +1,76 @@ +# Project definitions (do not edit) +Task: openfield +scorer: Pranav +date: Oct30 +multianimalproject: +identity: + + +# Project path (change when moving around) +project_path: WILL BE AUTOMATICALLY UPDATED BY DEMO CODE + +# Default DeepLabCut engine to use for shuffle creation (either pytorch or tensorflow) +engine: pytorch + + +# Annotation data set configuration (and individual video cropping parameters) +video_sets: + WILL BE AUTOMATICALLY UPDATED BY DEMO CODE: + crop: 0, 640, 0, 480 +bodyparts: +- snout +- leftear +- rightear +- tailbase + + +# Fraction of video to start/stop when extracting frames for labeling/refinement +start: 0 +stop: 1 +numframes2pick: 20 + + +# Plotting configuration +skeleton: [] +skeleton_color: black +pcutoff: 0.4 +dotsize: 8 +alphavalue: 0.7 +colormap: jet + + +# Training,Evaluation and Analysis configuration +TrainingFraction: +- 0.95 +iteration: 0 +default_net_type: resnet_50 +default_augmenter: imgaug +snapshotindex: -1 +detector_snapshotindex: -1 +batch_size: 4 +detector_batch_size: 1 + + +# Cropping Parameters (for analysis and outlier frame detection) +cropping: false +#if cropping is true for analysis, then set the values here: +x1: 0 +x2: 640 +y1: 277 +y2: 624 + + +# Refinement configuration (parameters from annotation dataset configuration also relevant in this stage) +corner2move2: +- 50 +- 50 +move2corner: true + + +# Conversion tables to fine-tune SuperAnimal weights +SuperAnimalConversionTables: + superanimal_topviewmouse: + snout: nose + leftear: left_ear + rightear: right_ear + tailbase: tail_base diff --git a/tests/io/config/test_base_config.py b/tests/io/config/test_base_config.py new file mode 100644 index 0000000..53a20dc --- /dev/null +++ b/tests/io/config/test_base_config.py @@ -0,0 +1,496 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Tests for DLCBaseConfig.""" + +import pytest +from pydantic import Field, ValidationError + +from dlclibrary.io.config import DLCBaseConfig, DLCVersionedConfig +from dlclibrary.deprecation import DLCDeprecationWarning + +_TOY_VERSION_OLD = 98 +_TOY_VERSION_NEW = 99 +_LEGACY_FIELD = "toy_legacy_field" + + +class ToyConfig(DLCBaseConfig): + """Minimal config used to exercise DLCBaseConfig.""" + + Task: str = "DefaultTask" + project_path: str = Field( + default="DefaultProjectPath", + json_schema_extra={"aliases": ["projectPath"]}, + ) + + +class NestedInner(DLCBaseConfig): + lr: float = 0.001 + momentum: float = 0.9 + + +class NestedOuter(DLCBaseConfig): + name: str = "outer" + inner: NestedInner | None = Field(default_factory=NestedInner) + + +class DictBranchConfig(DLCBaseConfig): + payload: dict = Field(default_factory=lambda: {"section": {"key": "old", "count": 1}}) + + +# ------------------------------------------------------------------ +# Dict-like access protocol +# ------------------------------------------------------------------ + + +class TestGetitem: + def test_getitem_returns_field_value(self): + cfg = ToyConfig(Task="t", project_path="/p") + assert cfg["Task"] == "t" + assert cfg["project_path"] == "/p" + + def test_getitem_missing_key_raises_key_error(self): + cfg = ToyConfig() + with pytest.raises(KeyError): + cfg["nonexistent"] + + +class TestSetitem: + def test_setitem_updates_field(self): + cfg = ToyConfig() + cfg["Task"] = "new_task" + assert cfg.Task == "new_task" + assert cfg["Task"] == "new_task" + + def test_setitem_unknown_field_raises_key_error(self): + cfg = ToyConfig() + with pytest.raises(KeyError, match="no field"): + cfg["nonexistent"] = 42 + + +class TestContains: + def test_contains_existing_field(self): + cfg = ToyConfig() + assert "Task" in cfg + assert "project_path" in cfg + + def test_contains_missing_field(self): + cfg = ToyConfig() + assert "nonexistent" not in cfg + + def test_contains_non_string_returns_false(self): + cfg = ToyConfig() + assert 42 not in cfg + + +class TestGet: + def test_get_existing_key(self): + cfg = ToyConfig(Task="hello") + assert cfg.get("Task") == "hello" + + def test_get_missing_key_returns_default(self): + cfg = ToyConfig() + assert cfg.get("nonexistent") is None + assert cfg.get("nonexistent", 42) == 42 + + +class TestUpdate: + def test_update_applies_fields_dict(self): + cfg = ToyConfig() + cfg.update({"Task": "t", "project_path": "/p"}) + assert cfg.Task == "t" + assert cfg.project_path == "/p" + + def test_update_applies_fields_kwargs(self): + cfg = ToyConfig() + cfg.update(Task="t", project_path="/p") + assert cfg.Task == "t" + assert cfg.project_path == "/p" + + def test_update_empty_is_noop(self): + cfg = ToyConfig(Task="unchanged") + assert cfg.update() is cfg + assert cfg.update({}) is cfg + assert cfg.Task == "unchanged" + + def test_update_rejects_dict_and_kwargs_together(self): + cfg = ToyConfig() + with pytest.raises(TypeError, match="either a dict or keyword"): + cfg.update({"Task": "x"}, Task="y") + + def test_update_unknown_field_raises_validation_error(self): + cfg = ToyConfig() + with pytest.raises(ValidationError): + cfg.update({"nonexistent": 42}) + + def test_update_invalid_value_raises(self): + cfg = TypedBase() + with pytest.raises(ValidationError): + cfg.update({"count": "not-a-number"}) + assert cfg.count == 0 + + def test_update_resolves_alias_with_warning(self): + cfg = ToyConfig() + with pytest.warns(DLCDeprecationWarning, match="projectPath") as record: + cfg.update({"projectPath": "/alias"}) + assert len(record) == 1 + assert cfg.project_path == "/alias" + + def test_update_rejects_alias_and_canonical_together(self): + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + with pytest.raises(TypeError, match=r"projectPath.*project_path"): + ToyConfig().update({"projectPath": "/a", "project_path": "/b"}) + + +class TestKeysValuesItems: + def test_keys(self): + cfg = ToyConfig(Task="t", project_path="/p") + assert cfg.keys() == ["Task", "project_path"] + + def test_values(self): + cfg = ToyConfig(Task="t", project_path="/p") + assert cfg.values() == ["t", "/p"] + + def test_items(self): + cfg = ToyConfig(Task="t", project_path="/p") + assert cfg.items() == [("Task", "t"), ("project_path", "/p")] + + +class TestIterAndLen: + def test_iter_yields_keys(self): + cfg = ToyConfig() + assert list(cfg) == ["Task", "project_path"] + + def test_len(self): + cfg = ToyConfig() + assert len(cfg) == 2 + + def test_dict_constructor(self): + cfg = ToyConfig(Task="t", project_path="/p") + d = dict(cfg.items()) + assert d == {"Task": "t", "project_path": "/p"} + + +# ------------------------------------------------------------------ +# select() utility +# ------------------------------------------------------------------ + + +class TestSelect: + def test_select_single_level(self): + cfg = ToyConfig(Task="t") + assert cfg.select("Task") == "t" + + def test_select_nested_config(self): + cfg = NestedOuter(name="outer", inner=NestedInner(lr=0.01)) + assert cfg.select("inner.lr") == 0.01 + assert cfg.select("inner.momentum") == 0.9 + + def test_select_missing_returns_default(self): + cfg = ToyConfig() + assert cfg.select("nonexistent") is None + assert cfg.select("nonexistent", "fallback") == "fallback" + + def test_select_deep_missing_returns_default(self): + cfg = NestedOuter() + assert cfg.select("inner.nonexistent") is None + assert cfg.select("nope.deep.path", 0) == 0 + + def test_select_none_intermediate(self): + cfg = NestedOuter(inner=None) + assert cfg.select("inner.lr") is None + + +# ------------------------------------------------------------------ +# set_nested() utility +# ------------------------------------------------------------------ + + +class TestSetNested: + def test_set_nested_top_level_field(self): + cfg = ToyConfig() + cfg.set_nested("Task", "updated") + assert cfg.Task == "updated" + assert cfg.select("Task") == "updated" + + def test_set_nested_nested_typed_field(self): + cfg = NestedOuter(inner=NestedInner(lr=0.001)) + cfg.set_nested("inner.lr", 0.05) + assert cfg.inner.lr == 0.05 + assert cfg.select("inner.lr") == 0.05 + + def test_set_nested_dict_field(self): + cfg = DictBranchConfig() + cfg.set_nested("payload.section.key", "new") + cfg.set_nested("payload.section.count", 42) + assert cfg.select("payload.section.key") == "new" + assert cfg.select("payload.section.count") == 42 + + def test_set_nested_returns_self_for_nested_path(self): + cfg = NestedOuter() + assert cfg.set_nested("inner.lr", 0.02) is cfg + + def test_set_nested_missing_path_raises(self): + cfg = NestedOuter() + with pytest.raises(AttributeError, match="has no 'inner.missing'"): + cfg.set_nested("inner.missing", 1.0) + + def test_set_nested_none_intermediate_raises(self): + cfg = NestedOuter(inner=None) + with pytest.raises(AttributeError, match="has no 'inner.lr'"): + cfg.set_nested("inner.lr", 0.1) + + def test_set_nested_invalid_value_raises(self): + cfg = NestedOuter() + with pytest.raises(ValidationError): + cfg.set_nested("inner.lr", "not-a-float") + assert cfg.inner.lr == 0.001 + + def test_set_nested_unknown_top_level_field_raises(self): + cfg = ToyConfig() + with pytest.raises(ValidationError): + cfg.set_nested("nonexistent", 42) + + +# ------------------------------------------------------------------ +# model_validate / extra=forbid +# ------------------------------------------------------------------ + + +class TestModelValidate: + def test_model_validate_returns_instance(self): + cfg = ToyConfig.model_validate({"Task": "task", "project_path": ""}) + assert isinstance(cfg, ToyConfig) + assert cfg.Task == "task" + + def test_model_validate_rejects_extra_keys(self): + with pytest.raises(ValidationError): + ToyConfig.model_validate({"Task": "ok", "project_path": "", "extra": 1}) + + +# ------------------------------------------------------------------ +# aliases +# ------------------------------------------------------------------ + + +class TestAliases: + def test_model_validate_resolves_alias_without_from_dict(self): + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + cfg = ToyConfig.model_validate({"Task": "t", "projectPath": "/alias"}) + assert cfg.project_path == "/alias" + + def test_model_validate_rejects_alias_and_canonical_together(self): + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + with pytest.raises(TypeError, match=r"projectPath.*project_path"): + ToyConfig.model_validate({"Task": "t", "projectPath": "/alias", "project_path": "/canonical"}) + + def test_getitem_alias_warns_and_reads_canonical(self): + cfg = ToyConfig(project_path="/p") + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + assert cfg["projectPath"] == "/p" + + def test_setitem_alias_warns_once_and_writes_canonical(self): + cfg = ToyConfig() + with pytest.warns(DLCDeprecationWarning, match="projectPath") as record: + cfg["projectPath"] = "/new" + assert len(record) == 1 + assert cfg.project_path == "/new" + + def test_setattr_alias_warns_and_validates(self): + cfg = ToyConfig() + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + cfg.projectPath = "/attr" + assert cfg.project_path == "/attr" + + def test_getattr_alias_warns(self): + cfg = ToyConfig(project_path="/p") + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + assert cfg.projectPath == "/p" + + def test_contains_accepts_alias(self): + cfg = ToyConfig() + assert "projectPath" in cfg + + def test_keys_does_not_include_alias(self): + """keys() / iter() only yield canonical field names, never alias names.""" + cfg = ToyConfig() + assert "projectPath" not in cfg.keys() + assert "projectPath" not in list(cfg) + + def test_to_dict_uses_canonical_names_only(self): + """to_dict() always emits canonical field names, even when loaded via alias.""" + with pytest.warns(DLCDeprecationWarning): + cfg = ToyConfig.model_validate({"Task": "t", "projectPath": "/alias"}) + d = cfg.to_dict() + assert "project_path" in d + assert "projectPath" not in d + + def test_from_yaml_with_alias_key_warns_and_loads(self, tmp_path): + """A YAML file that contains an alias key emits a warning on load.""" + path = tmp_path / "config.yaml" + path.write_text("Task: t\nprojectPath: /from_yaml\n") + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + cfg = ToyConfig.from_yaml(path) + assert cfg.project_path == "/from_yaml" + + def test_to_yaml_then_from_yaml_no_alias_warning(self, tmp_path): + """After saving via to_yaml, reloading must not emit any alias warning + (the YAML file should contain canonical field names).""" + import warnings + + path = tmp_path / "config.yaml" + cfg = ToyConfig(Task="t", project_path="/p") + cfg.to_yaml(path) + + with warnings.catch_warnings(): + warnings.simplefilter("error", DLCDeprecationWarning) + loaded = ToyConfig.from_yaml(path) + assert loaded.project_path == "/p" + + +# ------------------------------------------------------------------ +# validate_assignment on DLCBaseConfig (not just versioned) +# ------------------------------------------------------------------ + + +class TypedBase(DLCBaseConfig): + count: int = 0 + + +class TestBaseConfigValidateAssignment: + def test_invalid_assignment_raises(self): + cfg = TypedBase() + with pytest.raises(ValidationError): + cfg.count = "not-a-number" + assert cfg.count == 0 + + def test_valid_assignment_takes_effect(self): + cfg = TypedBase() + cfg.count = 7 + assert cfg.count == 7 + + def test_coercion_applied(self): + cfg = TypedBase() + cfg.count = True # coerced to 1 + assert cfg.count == 1 + + +# ------------------------------------------------------------------ +# from_dict / from_any / from_yaml / to_dict / to_yaml +# ------------------------------------------------------------------ + + +def test_from_dict_returns_instance(): + cfg = ToyConfig.from_dict({"Task": "t", "project_path": "/p"}) + assert isinstance(cfg, ToyConfig) + assert cfg.Task == "t" + assert cfg.project_path == "/p" + + +def test_from_dict_incomplete_uses_defaults(): + cfg = ToyConfig.from_dict({"Task": "custom_task"}) + assert cfg.Task == "custom_task" + assert cfg.project_path == "DefaultProjectPath" + + cfg_empty = ToyConfig.from_dict({}) + assert cfg_empty.Task == "DefaultTask" + assert cfg_empty.project_path == "DefaultProjectPath" + + +def test_from_any_with_instance_returns_same(): + cfg = ToyConfig.from_dict({"Task": "x", "project_path": ""}) + out = ToyConfig.from_any(cfg) + assert out is cfg + + +def test_from_any_with_dict_returns_instance(): + cfg = ToyConfig.from_any({"Task": "y", "project_path": ""}) + assert isinstance(cfg, ToyConfig) + assert cfg.Task == "y" + + +def test_from_any_with_invalid_type_raises(): + with pytest.raises(TypeError, match="Expected.*Got "): + ToyConfig.from_any(42) + + +def test_from_yaml_loads_file(tmp_path): + path = tmp_path / "config.yaml" + path.write_text("Task: mytask\nproject_path: /foo\n") + cfg = ToyConfig.from_yaml(path) + assert cfg.Task == "mytask" + assert cfg.project_path == "/foo" + + +def test_to_dict_returns_dict(): + cfg = ToyConfig.from_dict({"Task": "t", "project_path": ""}) + d = cfg.to_dict() + assert isinstance(d, dict) + assert d["Task"] == "t" + + +def test_from_dict_to_dict_roundtrip(): + d = {"Task": "rt", "project_path": "/x"} + cfg = ToyConfig.from_dict(d) + out = cfg.to_dict() + restored = ToyConfig.from_dict(out) + assert restored.Task == cfg.Task + assert restored.project_path == cfg.project_path + + +def test_to_yaml_roundtrip(tmp_path): + path = tmp_path / "out.yaml" + orig = ToyConfig.from_dict({"Task": "rt", "project_path": "/x"}) + orig.to_yaml(path) + loaded = ToyConfig.from_yaml(path) + assert loaded.Task == orig.Task + assert loaded.project_path == orig.project_path + + +def test_print_no_error(capsys): + cfg = ToyConfig.from_dict({"Task": "p", "project_path": ""}) + cfg.print() + out, _ = capsys.readouterr() + assert "Task" in out + + +# ------------------------------------------------------------------ +# No schema migration on DLCBaseConfig +# ------------------------------------------------------------------ + + +class _ToyBaseOnly(DLCBaseConfig): + toy_new_field: str = "" + + +class _ToyVersioned(DLCVersionedConfig): + config_version: int = _TOY_VERSION_NEW + toy_new_field: str = "" + + +class TestNoMigrationOnBaseConfig: + """DLCBaseConfig resolves aliases but does not run version migrations.""" + + def test_base_config_rejects_legacy_key_without_migration(self): + legacy_cfg = { + "config_version": _TOY_VERSION_OLD, + _LEGACY_FIELD: "value", + } + with pytest.raises(ValidationError): + _ToyBaseOnly.from_dict(legacy_cfg) + + def test_versioned_config_applies_migration_for_same_input(self, register_toy_migrations): + register_toy_migrations("_ToyVersioned") + legacy_cfg = { + "config_version": _TOY_VERSION_OLD, + _LEGACY_FIELD: "value", + } + cfg = _ToyVersioned.from_dict(legacy_cfg) + assert cfg.toy_new_field == "value" diff --git a/tests/io/config/test_change_tracking.py b/tests/io/config/test_change_tracking.py new file mode 100644 index 0000000..b0eadec --- /dev/null +++ b/tests/io/config/test_change_tracking.py @@ -0,0 +1,336 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Tests for change tracking on DLCVersionedConfig.""" + +from __future__ import annotations + +import logging + +import pytest +from pydantic import Field, ValidationError + +from dlclibrary.io.config import DLCVersionedConfig +from dlclibrary.deprecation import DLCDeprecationWarning + + +class TrackedConfig(DLCVersionedConfig): + name: str = "default" + count: int = 0 + flag: bool = False + + +# ------------------------------------------------------------------ +# Dirty-state tracking +# ------------------------------------------------------------------ + + +class TestDirtyState: + def test_clean_after_construction(self): + cfg = TrackedConfig() + assert not cfg.is_dirty + assert cfg.dirty_fields == frozenset() + + def test_dirty_after_field_change(self): + cfg = TrackedConfig() + cfg.name = "changed" + assert cfg.is_dirty + assert "name" in cfg.dirty_fields + + def test_not_dirty_when_set_to_same_value(self): + cfg = TrackedConfig(name="same") + cfg.name = "same" + assert not cfg.is_dirty + + def test_multiple_fields_tracked(self): + cfg = TrackedConfig() + cfg.name = "new" + cfg.count = 5 + assert cfg.dirty_fields == frozenset({"name", "count"}) + + def test_update_marks_fields_dirty(self): + cfg = TrackedConfig() + cfg.update({"name": "new", "count": 5}) + assert cfg.is_dirty + assert cfg.dirty_fields == frozenset({"name", "count"}) + + def test_dirty_fields_is_frozen(self): + cfg = TrackedConfig() + cfg.name = "x" + fs = cfg.dirty_fields + assert isinstance(fs, frozenset) + + +# ------------------------------------------------------------------ +# mark_clean +# ------------------------------------------------------------------ + + +class TestMarkClean: + def test_mark_clean_resets_dirty_fields(self): + cfg = TrackedConfig() + cfg.name = "x" + cfg.mark_clean() + assert not cfg.is_dirty + assert cfg.dirty_fields == frozenset() + + def test_mark_clean_resets_change_notes(self): + cfg = TrackedConfig() + cfg.name = "x" + cfg.record_change_note("name", "renamed") + cfg.mark_clean() + assert cfg.change_notes == [] + + def test_dirty_again_after_mark_clean(self): + cfg = TrackedConfig() + cfg.name = "x" + cfg.mark_clean() + cfg.count = 10 + assert cfg.is_dirty + assert cfg.dirty_fields == frozenset({"count"}) + + +# ------------------------------------------------------------------ +# Change notes +# ------------------------------------------------------------------ + + +class TestChangeNotes: + def test_no_notes_by_default(self): + cfg = TrackedConfig() + assert cfg.change_notes == [] + + def test_record_and_retrieve_note(self): + cfg = TrackedConfig() + cfg.name = "new" + cfg.record_change_note("name", "name was updated to 'new'") + assert cfg.change_notes == ["name was updated to 'new'"] + + def test_note_overwrites_previous_for_same_field(self): + cfg = TrackedConfig() + cfg.name = "a" + cfg.record_change_note("name", "first") + cfg.record_change_note("name", "second") + assert cfg.change_notes == ["second"] + + def test_multiple_notes_for_different_fields(self): + cfg = TrackedConfig() + cfg.name = "x" + cfg.count = 1 + cfg.record_change_note("name", "note-name") + cfg.record_change_note("count", "note-count") + assert set(cfg.change_notes) == {"note-name", "note-count"} + + def test_include_caller_appends_tag(self): + cfg = TrackedConfig() + cfg.name = "x" + cfg.record_change_note("name", "updated", include_caller=True) + notes = cfg.change_notes + assert len(notes) == 1 + assert notes[0].startswith("updated [") + assert "test_change_tracking.py:" in notes[0] + + +# ------------------------------------------------------------------ +# log_changes +# ------------------------------------------------------------------ + + +class TestLogChanges: + def test_log_changes_no_output_when_clean(self, caplog): + cfg = TrackedConfig() + with caplog.at_level(logging.INFO): + cfg.log_changes() + assert caplog.text == "" + + def test_log_changes_includes_note(self, caplog): + cfg = TrackedConfig() + cfg.name = "x" + cfg.record_change_note("name", "name was updated") + with caplog.at_level(logging.INFO): + cfg.log_changes() + assert "name was updated" in caplog.text + + def test_log_changes_shows_field_without_note(self, caplog): + cfg = TrackedConfig() + cfg.count = 99 + with caplog.at_level(logging.INFO): + cfg.log_changes() + assert "count was modified" in caplog.text + + def test_log_changes_mixes_notes_and_bare_fields(self, caplog): + cfg = TrackedConfig() + cfg.name = "y" + cfg.count = 5 + cfg.record_change_note("name", "renamed to y") + with caplog.at_level(logging.INFO): + cfg.log_changes() + assert "renamed to y" in caplog.text + assert "count was modified" in caplog.text + + def test_log_changes_header_contains_class_name(self, caplog): + cfg = TrackedConfig() + cfg.flag = True + with caplog.at_level(logging.INFO): + cfg.log_changes() + assert "TrackedConfig" in caplog.text + + +# ------------------------------------------------------------------ +# Integration with DLCVersionedConfig.from_yaml +# ------------------------------------------------------------------ + + +class TestFromYamlIntegration: + def test_clean_after_from_yaml(self, tmp_path): + path = tmp_path / "config.yaml" + path.write_text("name: loaded\ncount: 42\nflag: true\n") + cfg = TrackedConfig.from_yaml(path) + assert cfg.name == "loaded" + assert not cfg.is_dirty + assert cfg.change_notes == [] + + def test_roundtrip_stays_clean(self, tmp_path): + path = tmp_path / "config.yaml" + orig = TrackedConfig(name="rt", count=7) + orig.to_yaml(path) + loaded = TrackedConfig.from_yaml(path) + assert not loaded.is_dirty + + +# ------------------------------------------------------------------ +# Interaction with pydantic validate_assignment +# ------------------------------------------------------------------ + + +class TestValidateAssignment: + """Verify that change tracking wraps *around* pydantic's validator.""" + + def test_valid_assignment_is_tracked(self): + cfg = TrackedConfig() + cfg.count = 7 + assert cfg.count == 7 + assert "count" in cfg.dirty_fields + + def test_invalid_assignment_rejected_and_not_tracked(self): + cfg = TrackedConfig() + with pytest.raises(ValidationError): + cfg.count = "not-an-int" + assert cfg.count == 0 + assert not cfg.is_dirty + + def test_coerced_value_stored(self): + """Pydantic coerces compatible types (e.g. bool -> int); the coerced + value should be stored and tracked.""" + cfg = TrackedConfig() + cfg.count = True # coerced to 1 + assert cfg.count == 1 + assert "count" in cfg.dirty_fields + + +# ------------------------------------------------------------------ +# Instance isolation — two instances of the same class are independent +# ------------------------------------------------------------------ + + +class TestInstanceIsolation: + """The class-level __setattr__ patch must not cause instances to share state.""" + + def test_two_instances_independent_dirty_sets(self): + cfg1 = TrackedConfig() + cfg2 = TrackedConfig() + cfg1.name = "one" + assert "name" in cfg1.dirty_fields + assert not cfg2.is_dirty + + def test_mark_clean_on_one_does_not_affect_other(self): + cfg1 = TrackedConfig() + cfg2 = TrackedConfig() + cfg1.name = "x" + cfg2.count = 5 + cfg1.mark_clean() + assert not cfg1.is_dirty + assert cfg2.is_dirty + assert "count" in cfg2.dirty_fields + + def test_different_subclasses_independent(self): + class ConfigA(DLCVersionedConfig): + x: int = 0 + + class ConfigB(DLCVersionedConfig): + y: int = 0 + + a = ConfigA() + b = ConfigB() + a.x = 1 + assert "x" in a.dirty_fields + assert not b.is_dirty + + +# ------------------------------------------------------------------ +# from_dict and to_yaml lifecycle +# ------------------------------------------------------------------ + + +class TestLifecycle: + def test_from_dict_is_clean(self): + cfg = TrackedConfig.from_dict({"name": "loaded", "count": 5}) + assert not cfg.is_dirty + assert cfg.change_notes == [] + + def test_to_yaml_marks_writer_clean(self, tmp_path): + path = tmp_path / "out.yaml" + cfg = TrackedConfig(name="dirty") + cfg.name = "changed" + assert cfg.is_dirty + cfg.to_yaml(path) + assert not cfg.is_dirty + + def test_to_yaml_with_mark_clean_false_does_not_reset(self, tmp_path): + path = tmp_path / "out.yaml" + cfg = TrackedConfig() + cfg.name = "changed" + cfg.to_yaml(path, mark_clean=False) + assert cfg.is_dirty + + def test_record_change_note_for_unmodified_field_is_stored(self): + """record_change_note does not require the field to be dirty; it just stores the note.""" + cfg = TrackedConfig() + cfg.record_change_note("name", "set during init") + assert cfg.change_notes == ["set during init"] + assert not cfg.is_dirty # still clean — no assignment was made + + +# ------------------------------------------------------------------ +# Alias access with change tracking +# ------------------------------------------------------------------ + + +class TrackedAliasConfig(DLCVersionedConfig): + project_path: str = Field( + default="", + json_schema_extra={"aliases": ["projectPath"]}, + ) + + +class TestAliasChangeTracking: + def test_setitem_alias_marks_canonical_field_dirty(self): + cfg = TrackedAliasConfig() + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + cfg["projectPath"] = "/new" + assert cfg.project_path == "/new" + assert "project_path" in cfg.dirty_fields + + def test_setattr_alias_marks_canonical_field_dirty(self): + cfg = TrackedAliasConfig() + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + cfg.projectPath = "/attr" + assert cfg.project_path == "/attr" + assert "project_path" in cfg.dirty_fields diff --git a/tests/io/config/test_config_breakage.py b/tests/io/config/test_config_breakage.py new file mode 100644 index 0000000..db874ac --- /dev/null +++ b/tests/io/config/test_config_breakage.py @@ -0,0 +1,292 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Pathological casess tests for the centralized config model.""" + +from __future__ import annotations + +import enum +import logging +from pathlib import Path + +import pytest +from pydantic import Field, ValidationError + +from dlclibrary.io.config import DLCBaseConfig, DLCVersionedConfig, ProjectConfig, versioning +from dlclibrary.deprecation import DLCDeprecationWarning + +# ----------------------------------------------------------------------------- +# In-place nested mutation +# ----------------------------------------------------------------------------- + + +@pytest.mark.xfail( + reason="This may make validation difficult and cause subtle issues " + "if we start changing how functions such as append() operate." + " Avoiding lists in configs could be a better long-term solution." +) +def test_in_place_list_mutation_should_be_validated_like_assignment(): + """Appending to a config list changes config state without calling __setattr__. + + A strict config system should reject invalid values even when users mutate an + existing list instead of assigning a replacement list. + + Note that this pattern would incur validation: + cfg.TrainingFraction = [*cfg.TrainingFraction, 0.8] + cfg.video_sets = {**cfg.video_sets, "video.mp4": {"crop": "0, 100, 0, 100"}} + """ + cfg = ProjectConfig(TrainingFraction=[0.95]) + cfg.mark_clean() + + with pytest.raises(ValidationError): + cfg.TrainingFraction.append(1.5) + + +@pytest.mark.xfail(reason="Not implemented yet") +@pytest.mark.parametrize( + ("field_name", "mutate"), + [ + ("TrainingFraction", lambda cfg: cfg.TrainingFraction.append(0.8)), + ("video_sets", lambda cfg: cfg.video_sets.__setitem__("video.mp4", {"crop": "0, 100, 0, 100"})), + ], +) +def test_in_place_nested_mutation_should_mark_config_dirty(field_name, mutate): + """Mutating nested containers changes the saved config but may bypass dirty tracking. + + Dirty tracking is useful only if semantic config changes are recorded, whether + the user assigns a whole field or mutates a nested list/dict in place. + This ties with the problem that some nested fields are still dict instead of config models, + maybe once we have full config models we can come up with a parent/child system where changes in a child + are detected when querying the parent for its dirty state (or a simpler equivalent) + """ + cfg = ProjectConfig(TrainingFraction=[0.95], video_sets={}) + cfg.mark_clean() + + mutate(cfg) + + assert cfg.is_dirty + assert field_name in cfg.dirty_fields + + +# ----------------------------------------------------------------------------- +# Private state isolation +# ----------------------------------------------------------------------------- + + +class _TrackedForIsolation(DLCVersionedConfig): + name: str = "default" + count: int = 0 + + +def test_dirty_state_is_not_shared_between_instances(): + """Private dirty-field state must be per-instance, not shared via a mutable default.""" + first = _TrackedForIsolation() + second = _TrackedForIsolation() + + first.name = "changed" + + assert "name" in first.dirty_fields + assert not second.is_dirty + assert "name" not in second.dirty_fields + + +def test_change_notes_are_not_shared_between_instances(): + """Private change-note state must be per-instance, not shared via a mutable default.""" + first = _TrackedForIsolation() + second = _TrackedForIsolation() + + first.record_change_note("name", "name changed on first instance") + + assert first.change_notes == ["name changed on first instance"] + assert second.change_notes == [] + + +# ----------------------------------------------------------------------------- +# Changes note consistency +# ----------------------------------------------------------------------------- + + +def test_change_note_should_reject_unknown_field_names(): + """Notes for misspelled fields are silently lost during logging. + + Rejecting unknown field names catches typos at the call site instead of + storing a note that can never match a dirty field. + """ + cfg = ProjectConfig() + + with pytest.raises(KeyError): + cfg.record_change_note("not_a_real_field", "this note should not be accepted") + + +def test_change_note_recorded_with_alias_should_follow_canonical_dirty_field(caplog): + """Aliases should not split notes from the canonical dirty field name. + + If a note is recorded with a deprecated alias, log_changes should still use + that note when the canonical field is modified. + """ + cfg = ProjectConfig() + cfg.mark_clean() + + with pytest.warns(DLCDeprecationWarning, match="with_identity"): + cfg.record_change_note("with_identity", "identity changed through compatibility alias") + cfg.identity = True + + with caplog.at_level(logging.INFO): + cfg.log_changes() + + assert "identity changed through compatibility alias" in caplog.text + + +# ----------------------------------------------------------------------------- +# Nested YAML comments and nested serialization +# ----------------------------------------------------------------------------- + + +class _CommentedInner(DLCBaseConfig): + threshold: float = Field( + default=0.5, + json_schema_extra={"comment": "Nested threshold comment"}, + ) + + +class _CommentedOuter(DLCBaseConfig): + inner: _CommentedInner = Field( + default_factory=_CommentedInner, + json_schema_extra={"comment": "Inner config section"}, + ) + + +def test_to_yaml_should_emit_comments_for_nested_config_fields(tmp_path): + """Nested config models can have field comments too. + + This protects against only applying YAML comments to the top-level model and + silently dropping useful nested schema documentation. + """ + path = tmp_path / "commented.yaml" + + _CommentedOuter().to_yaml(path) + + text = path.read_text() + assert "Inner config section" in text + assert "Nested threshold comment" in text + + +class _SerializationMode(enum.Enum): + FAST = "fast" + ACCURATE = "accurate" + + +class _SerializableInner(DLCBaseConfig): + output_path: Path = Path("outputs/predictions") + mode: _SerializationMode = _SerializationMode.FAST + + +class _SerializableOuter(DLCBaseConfig): + inner: _SerializableInner = Field(default_factory=_SerializableInner) + + +def test_nested_config_normalization_should_handle_paths_and_enums(): + """Nested model dumps should normalize non-primitive values before YAML output. + + Path and Enum values are common in configs and should become plain YAML-safe + values even when they appear inside nested config models. + """ + cfg = _SerializableOuter() + + assert cfg.to_dict(normalize=True) == { + "inner": { + "output_path": str(Path("outputs/predictions")), + "mode": "fast", + } + } + + +def test_nested_config_to_yaml_should_write_plain_path_and_enum_values(tmp_path): + """YAML output should not rely on Python-specific object representation. + + This catches failures where nested BaseModel values, pathlib paths, or enums + are passed to the YAML dumper without normalization. + """ + path = tmp_path / "serializable.yaml" + + _SerializableOuter().to_yaml(path) + + text = path.read_text() + assert str(Path("outputs/predictions")) in text + assert "fast" in text + + +# ----------------------------------------------------------------------------- +# Alias warnings and canonical writes +# ----------------------------------------------------------------------------- + + +def test_item_assignment_with_alias_should_warn_once_and_track_canonical_field(): + """Dict-style alias assignment should not warn twice or track the alias name. + + __setitem__ resolves aliases before delegating to assignment validation, so + the visible side effects should be one warning and one canonical dirty field. + """ + cfg = ProjectConfig() + cfg.mark_clean() + + with pytest.warns(DLCDeprecationWarning, match="with_identity") as caught: + cfg["with_identity"] = True + + assert len(caught) == 1 + assert cfg.identity is True + assert "identity" in cfg.dirty_fields + assert "with_identity" not in cfg.dirty_fields + + +# ----------------------------------------------------------------------------- +# Cross-field validation vs bulk update +# ----------------------------------------------------------------------------- + + +@pytest.mark.xfail( + reason=( + "update() applies overrides one setattr at a time; cross-field " + "model validators run on each assignment, so intermediate states " + "like multianimalproject=True with bodyparts=[] are rejected." + ), +) +def test_update_applies_cross_field_overrides_atomically(): + """Bulk update should validate the merged state, not each setattr in isolation.""" + cfg = ProjectConfig(bodyparts=[], multianimalproject=False) + # Invalid state, midway updating: when bodyparts is set to "MULTI!" before multianimalproject is set to True. + cfg.update({"bodyparts": "MULTI!", "multianimalproject": True}) + assert cfg.multianimalproject is True + assert cfg.bodyparts == "MULTI!" + + +# ----------------------------------------------------------------------------- +# Migration skipped on typed construction +# ----------------------------------------------------------------------------- + + +class _TypedConstructionCfg(DLCVersionedConfig): + config_version: int = 1 + epochs: int = 50 + + +@pytest.mark.parametrize("kwargs", [{}, {"epochs": 100}]) +def test_migration_skipped_on_typed_construction(monkeypatch, kwargs): + """Typed construction must not run the legacy YAML migration chain. + + ``MyCfg(epochs=100)`` is current-schema construction, not loading an old file. + Migration should only run via ``from_dict`` / ``from_yaml``. + """ + monkeypatch.setattr(versioning, "CURRENT_CONFIG_VERSION", 1) + + cfg = _TypedConstructionCfg(**kwargs) + + assert cfg.epochs == kwargs.get("epochs", 50) + assert cfg.config_version == 1 diff --git a/tests/io/config/test_core_config.py b/tests/io/config/test_core_config.py new file mode 100644 index 0000000..bc6d317 --- /dev/null +++ b/tests/io/config/test_core_config.py @@ -0,0 +1,223 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Tests for dlclibrary.io.config.""" + +import warnings +from pathlib import Path + +import pytest +from ruamel.yaml.error import YAMLError + +from dlclibrary.io.config import ( + pretty_print, + read_config_as_dict, + resolve_alias, + resolve_aliases_in_dict, + write_config, +) +from dlclibrary.deprecation import DLCDeprecationWarning + +# ----------------------------------------------------------------------------- +# read_config_as_dict +# ----------------------------------------------------------------------------- + + +def test_read_config_as_dict_loads_yaml(tmp_path): + config_path = tmp_path / "config.yaml" + config_path.write_text("a: 1\nb: [2, 3]\nc:\n d: 4\n") + cfg = read_config_as_dict(str(config_path)) + assert cfg == {"a": 1, "b": [2, 3], "c": {"d": 4}} + + +def test_read_config_as_dict_raises_when_file_missing(): + with pytest.raises(FileNotFoundError): + read_config_as_dict(Path("/nonexistent/config.yaml")) + + +def test_read_config_as_dict_raises_on_invalid_yaml(tmp_path): + """Broken YAML syntax (e.g. unclosed bracket) raises an exception.""" + config_path = tmp_path / "broken.yaml" + config_path.write_text("key: [unclosed\n") + with pytest.raises(YAMLError): + read_config_as_dict(config_path) + + +@pytest.mark.parametrize("content", ["", "null", "~"]) +def test_read_config_as_dict_raises_when_empty_or_null(tmp_path, content): + config_path = tmp_path / "config.yaml" + config_path.write_text(content) + with pytest.raises(ValueError, match="empty or null"): + read_config_as_dict(config_path) + + +def test_read_config_as_dict_raises_when_root_is_not_mapping(tmp_path): + config_path = tmp_path / "config.yaml" + config_path.write_text("- item\n") + with pytest.raises(ValueError, match="must be a YAML mapping"): + read_config_as_dict(config_path) + + +def test_read_config_as_dict_breaks_for_yaml_tags(tmp_path): + """read_config breaks for YAML tags like !!python/tuple""" + config_path = tmp_path / "config.yaml" + config_path.write_text("project_path: /old/path\nengine: pytorch\nbodyparts: !!python/tuple [a, b, c]\n") + with pytest.raises(YAMLError): + read_config_as_dict(config_path) + + +def test_read_config_as_dict_accepts_config_with_misnamed_fields(tmp_path): + """Config with typos/misnamed keys still loads; keys are not validated.""" + # NOTE @deruyter92 2026-02-03: This test captures the current behavior where + # read_config_as_dict does not validate the keys. This behavior is different + # from read_config, which should validate the keys (in the future). + # The tests can be updated according, depending on the preferred behavior. + config_path = tmp_path / "typos.yaml" + config_path.write_text( + "project_pathh: /wrong\n" # typo + "bodypartz: [a, b]\n" # typo + "Task: mytask\n" + ) + cfg = read_config_as_dict(config_path) + assert cfg["project_pathh"] == "/wrong" + assert cfg["bodypartz"] == ["a", "b"] + assert cfg["Task"] == "mytask" + assert "project_path" not in cfg + assert "bodyparts" not in cfg + + +# ----------------------------------------------------------------------------- +# write_config +# ----------------------------------------------------------------------------- + + +def test_write_config_creates_file(tmp_path): + config_path = tmp_path / "out.yaml" + write_config(config_path, {"a": 1, "b": 2}) + assert config_path.exists() + cfg = read_config_as_dict(config_path) + assert cfg["a"] == 1 and cfg["b"] == 2 + + +def test_write_config_overwrites_existing(tmp_path): + config_path = tmp_path / "out.yaml" + config_path.write_text("old: true\n") + write_config(config_path, {"new": True}, overwrite=True) + cfg = read_config_as_dict(config_path) + assert "new" in cfg and "old" not in cfg + + +def test_write_config_raises_when_overwrite_false_and_file_exists(tmp_path): + config_path = tmp_path / "out.yaml" + config_path.write_text("x: 1\n") + with pytest.raises(FileExistsError): + write_config(config_path, {"x": 2}, overwrite=False) + + +def test_write_config_allows_overwrite_false_when_file_missing(tmp_path): + config_path = tmp_path / "new.yaml" + write_config(config_path, {"x": 1}, overwrite=False) + assert read_config_as_dict(config_path) == {"x": 1} + + +# ----------------------------------------------------------------------------- +# pretty_print +# ----------------------------------------------------------------------------- + + +def test_pretty_print_flat_config(capsys): + pretty_print({"a": 1, "b": 2}) + out, _ = capsys.readouterr() + assert "a: 1" in out and "b: 2" in out + + +def test_pretty_print_nested_config(capsys): + pretty_print({"top": {"nested": 42}}) + out, _ = capsys.readouterr() + assert "top:" in out and "nested: 42" in out + + +def test_pretty_print_with_indent(capsys): + pretty_print({"k": "v"}, indent=4) + out, _ = capsys.readouterr() + assert out.startswith(" k:") + + +def test_pretty_print_with_custom_print_fn(): + lines = [] + + def capture(s): + lines.append(s) + + pretty_print({"a": 1}, print_fn=capture) + assert any("a: 1" in line for line in lines) + + +# ----------------------------------------------------------------------------- +# resolve_alias / resolve_aliases_in_dict +# ----------------------------------------------------------------------------- + + +ALIAS_MAP = {"projectPath": "project_path"} + + +def test_resolve_alias_returns_canonical_for_alias(): + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + assert resolve_alias("projectPath", ALIAS_MAP) == "project_path" + + +def test_resolve_alias_returns_name_unchanged_for_unknown_key(): + assert resolve_alias("Task", ALIAS_MAP, warn=False) == "Task" + + +def test_resolve_alias_no_warning_when_warn_false(): + + with warnings.catch_warnings(): + warnings.simplefilter("error", DLCDeprecationWarning) + assert resolve_alias("projectPath", ALIAS_MAP, warn=False) == "project_path" + + +def test_resolve_aliases_in_dict_resolves_single_alias(): + with pytest.warns(DLCDeprecationWarning, match="projectPath"): + resolved = resolve_aliases_in_dict( + {"Task": "t", "projectPath": "/p"}, + ALIAS_MAP, + target="ToyConfig", + ) + assert resolved == {"Task": "t", "project_path": "/p"} + + +def test_resolve_aliases_in_dict_empty_alias_map_returns_input_unchanged(): + cfg_dict = {"Task": "t", "project_path": "/p"} + assert resolve_aliases_in_dict(cfg_dict, {}) is cfg_dict + + +def test_resolve_aliases_in_dict_rejects_alias_and_canonical_together(): + with pytest.raises(TypeError, match=r"projectPath.*project_path"): + resolve_aliases_in_dict( + {"projectPath": "/alias", "project_path": "/canonical"}, + ALIAS_MAP, + target="ToyConfig", + warn=False, + ) + + +def test_resolve_aliases_in_dict_rejects_two_aliases_for_same_field(): + alias_map = { + "projectPath": "project_path", + "legacyProjectPath": "project_path", + } + with pytest.raises(TypeError, match=r"projectPath.*legacyProjectPath"): + resolve_aliases_in_dict( + {"projectPath": "/a", "legacyProjectPath": "/b"}, + alias_map, + target="ToyConfig", + warn=False, + ) diff --git a/tests/io/config/test_owed.py b/tests/io/config/test_owed.py new file mode 100644 index 0000000..970d241 --- /dev/null +++ b/tests/io/config/test_owed.py @@ -0,0 +1,30 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Owed changes relative to DeepLabCut's copy of the config code.""" + +from pathlib import Path + +import pytest + +from dlclibrary.io.config import ProjectConfig + + +@pytest.mark.xfail( + strict=True, + raises=AssertionError, + reason="ProjectConfig._post_yaml_load_updates replaces project_path on every from_yaml, as in DeepLabCut", +) +def test_a_plain_read_leaves_project_path_as_stored(tmp_path): + stored = tmp_path / "elsewhere" + config_path = tmp_path / "config.yaml" + config_path.write_text(f"project_path: {stored.as_posix()}\n") + + assert ProjectConfig.from_yaml(config_path).project_path == Path(stored.as_posix()) diff --git a/tests/io/config/test_project_config.py b/tests/io/config/test_project_config.py new file mode 100644 index 0000000..c80c3c2 --- /dev/null +++ b/tests/io/config/test_project_config.py @@ -0,0 +1,337 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Tests for ProjectConfig — the real DLCVersionedConfig consumer.""" + +import logging +from pathlib import Path + +import pytest + +from dlclibrary.io.config import ProjectConfig, read_config_as_dict +from dlclibrary.io.config.versioning import CURRENT_CONFIG_VERSION + +# ----------------------------------------------------------------------------- +# Basic construction / defaults +# ----------------------------------------------------------------------------- + + +class TestProjectConfigDefaults: + def test_defaults(self): + cfg = ProjectConfig() + assert cfg.config_version == CURRENT_CONFIG_VERSION + assert cfg.engine == "pytorch" + assert cfg.Task == "" + assert isinstance(cfg.project_path, Path) + assert cfg.pose_config_path is None + assert cfg.config_yaml_path == cfg.project_path / "config.yaml" + assert cfg.config_yaml_path == cfg.project_path / "config.yaml" + assert cfg.config_yaml_path == cfg.project_path / "config.yaml" + + def test_from_dict_minimal(self): + cfg = ProjectConfig.from_dict( + { + "Task": "mytask", + "scorer": "scorer", + "date": "Jan01", + "project_path": "/tmp/project", + } + ) + assert cfg.Task == "mytask" + assert cfg.project_path == Path("/tmp/project") + + def test_from_dict_rejects_unknown_keys(self): + from pydantic import ValidationError + + with pytest.raises(ValidationError): + ProjectConfig.from_dict({"Task": "t", "not_a_real_field": 1}) + + def test_is_versioned_config(self): + cfg = ProjectConfig() + assert hasattr(cfg, "is_dirty") + assert not cfg.is_dirty + + +# ----------------------------------------------------------------------------- +# bodyparts_list +# ----------------------------------------------------------------------------- + + +class TestProjectConfigBodypartsList: + def test_single_animal_bodyparts_list_returns_bodyparts(self): + cfg = ProjectConfig(multianimalproject=False, bodyparts=["nose", "tail"]) + assert cfg.bodyparts_list == ["nose", "tail"] + + def test_multi_animal_bodyparts_list_returns_multianimalbodyparts(self): + cfg = ProjectConfig( + multianimalproject=True, + bodyparts="MULTI!", + multianimalbodyparts=["nose", "tail"], + ) + assert cfg.bodyparts_list == ["nose", "tail"] + + def test_single_animal_multi_sentinel_raises(self): + from pydantic import ValidationError + + with pytest.raises(ValidationError, match="MULTI!"): + ProjectConfig(multianimalproject=False, bodyparts="MULTI!") + + +# ----------------------------------------------------------------------------- +# validate_project_path / from_any(repair_path=True) +# ----------------------------------------------------------------------------- + + +def _write_wrong_project_path_yaml(config_path: Path) -> None: + config_path.write_text("project_path: /completely/wrong/path\nengine: pytorch\n") + + +class TestValidateProjectPath: + def test_raises_when_yaml_not_found(self, tmp_path): + cfg = ProjectConfig(project_path=tmp_path) + with pytest.raises(FileNotFoundError, match="config.yaml not found"): + cfg.validate_project_path() + + @pytest.mark.parametrize("write", [False, True]) + def test_validate_project_path_write_flag(self, tmp_path, write): + config_path = tmp_path / "config.yaml" + _write_wrong_project_path_yaml(config_path) + cfg = ProjectConfig(project_path=Path("/completely/wrong/path")) + cfg.validate_project_path(yaml_path=config_path, write=write) + assert cfg.project_path == tmp_path + saved = read_config_as_dict(config_path) + if write: + assert Path(saved["project_path"]) == tmp_path + assert not cfg.is_dirty + else: + assert saved["project_path"] == "/completely/wrong/path" + assert "project_path" in cfg.dirty_fields + + @pytest.mark.parametrize( + ("config_input", "repair_path"), + [ + ("yaml_path", True), + ("dict", True), + ("dict", False), + ], + ids=["yaml_path", "dict_repair", "dict_no_repair"], + ) + def test_from_any_repair_path(self, tmp_path, config_input, repair_path): + config_path = tmp_path / "config.yaml" + _write_wrong_project_path_yaml(config_path) + if config_input == "yaml_path": + cfg = ProjectConfig.from_any(config_path, repair_path=repair_path) + else: + cfg = ProjectConfig.from_any( + {"project_path": tmp_path, "engine": "pytorch"}, + repair_path=repair_path, + ) + assert cfg.project_path == tmp_path + assert not cfg.is_dirty + saved = read_config_as_dict(config_path) + if config_input == "yaml_path" and repair_path: + assert Path(saved["project_path"]) == tmp_path + else: + assert saved["project_path"] == "/completely/wrong/path" + + +# ----------------------------------------------------------------------------- +# _post_yaml_load_updates: project_path repair +# ----------------------------------------------------------------------------- + + +class TestPostYamlLoadUpdates: + def test_project_path_updated_when_differs_from_yaml_location(self, tmp_path): + """When the stored project_path differs from the YAML's directory, + _post_yaml_load_updates corrects it and marks the field dirty.""" + config_path = tmp_path / "config.yaml" + config_path.write_text("project_path: /completely/wrong/path\nengine: pytorch\n") + cfg = ProjectConfig.from_yaml(config_path) + assert cfg.project_path == tmp_path + assert "project_path" in cfg.dirty_fields + assert any("project_path updated" in n for n in cfg.change_notes) + + def test_project_path_not_updated_when_already_correct(self, tmp_path): + """When the stored project_path already matches the YAML directory, + nothing is changed and the config stays clean.""" + config_path = tmp_path / "config.yaml" + config_path.write_text(f"project_path: {tmp_path}\nengine: pytorch\n") + cfg = ProjectConfig.from_yaml(config_path) + assert cfg.project_path == tmp_path + assert not cfg.is_dirty + + +# ----------------------------------------------------------------------------- +# Legacy empty / null values (normalize_legacy_empty_values) +# ----------------------------------------------------------------------------- + + +class TestNormalizeLegacyEmptyValues: + def test_none_for_non_optional_field_uses_default(self): + """Bare/null YAML values become None in dicts; non-optional fields use defaults.""" + cfg = ProjectConfig.from_dict({"multianimalproject": None, "engine": "pytorch"}) + assert cfg.multianimalproject is False + + def test_none_for_optional_field_kept_as_none(self): + """Optional fields (T | None) accept None; it is not rewritten to a non-None default.""" + cfg = ProjectConfig.from_dict({"identity": None, "engine": "pytorch"}) + assert cfg.identity is None + + def test_empty_string_for_legacy_list_fields_uses_default(self): + cfg = ProjectConfig.from_dict( + { + "skeleton": "", + "TrainingFraction": "", + "video_sets": "", + "bodyparts": "", + "engine": "pytorch", + } + ) + defaults = ProjectConfig() + assert cfg.skeleton == defaults.skeleton + assert cfg.TrainingFraction == defaults.TrainingFraction + assert cfg.video_sets == defaults.video_sets + assert cfg.bodyparts == defaults.bodyparts + + def test_from_dict_and_from_yaml_agree_on_null_keys(self, tmp_path): + """Dict path (legacy scripts) must match YAML path for bare/null keys.""" + config_path = tmp_path / "config.yaml" + config_path.write_text( + "\n".join( + [ + f"project_path: {tmp_path}", + "engine: pytorch", + "multianimalproject:", + "identity:", + "bodyparts:", + " - snout", + "", + ] + ) + ) + from_yaml = ProjectConfig.from_yaml(config_path) + from_dict = ProjectConfig.from_dict(read_config_as_dict(config_path)) + assert from_dict.multianimalproject == from_yaml.multianimalproject is False + assert from_dict.identity == from_yaml.identity is None + assert from_dict.bodyparts == from_yaml.bodyparts == ["snout"] + + def test_openfield_example_loads_via_dict_and_yaml(self): + """Example project configs with empty fields must load via both entrypoints.""" + config_path = Path(__file__).resolve().parent / "fixtures" / "openfield-Pranav-2018-10-30" / "config.yaml" + assert config_path.is_file() + raw = read_config_as_dict(config_path) + assert raw["multianimalproject"] is None + assert raw["identity"] is None + + from_dict = ProjectConfig.from_dict(raw) + from_yaml = ProjectConfig.from_yaml(config_path) + assert from_dict.multianimalproject == from_yaml.multianimalproject is False + assert from_dict.identity == from_yaml.identity is None + + def test_null_non_optional_emits_warning(self, caplog): + with caplog.at_level(logging.WARNING, logger="dlclibrary.io.config.project_config"): + ProjectConfig.from_dict({"multianimalproject": None, "engine": "pytorch"}) + assert any("multianimalproject" in r.message for r in caplog.records) + + +# ----------------------------------------------------------------------------- +# YAML round-trip +# ----------------------------------------------------------------------------- + + +class TestProjectConfigYamlRoundtrip: + def test_roundtrip_preserves_scalar_fields(self, tmp_path): + config_path = tmp_path / "config.yaml" + orig = ProjectConfig( + Task="pose", + scorer="scorer1", + date="2026-01-01", + project_path=tmp_path, + engine="pytorch", + pcutoff=0.6, + ) + orig.to_yaml(config_path) + loaded = ProjectConfig.from_yaml(config_path) + assert loaded.Task == "pose" + assert loaded.scorer == "scorer1" + assert loaded.pcutoff == 0.6 + assert loaded.engine == "pytorch" + + def test_roundtrip_config_is_clean(self, tmp_path): + """After saving and reloading, the loaded config must be clean + (project_path already matches YAML location).""" + config_path = tmp_path / "config.yaml" + orig = ProjectConfig(project_path=tmp_path) + orig.to_yaml(config_path) + loaded = ProjectConfig.from_yaml(config_path) + assert not loaded.is_dirty + + def test_roundtrip_no_unknown_keys_in_yaml(self, tmp_path): + """The saved YAML must not contain keys that ProjectConfig doesn't declare.""" + config_path = tmp_path / "config.yaml" + ProjectConfig(project_path=tmp_path).to_yaml(config_path) + saved = read_config_as_dict(config_path) + declared = set(ProjectConfig.model_fields.keys()) + unknown = set(saved.keys()) - declared + assert unknown == set(), f"Unknown keys written to YAML: {unknown}" + + def test_to_yaml_writes_config_version(self, tmp_path): + config_path = tmp_path / "config.yaml" + ProjectConfig(project_path=tmp_path).to_yaml(config_path) + saved = read_config_as_dict(config_path) + assert "config_version" in saved + assert saved["config_version"] == CURRENT_CONFIG_VERSION + + +# ----------------------------------------------------------------------------- +# Migration / versioning integration +# ----------------------------------------------------------------------------- + + +class TestProjectConfigVersioning: + def test_no_migration_log_for_current_version_yaml(self, tmp_path, caplog): + """Loading a config already at CURRENT_CONFIG_VERSION emits no migration INFO log.""" + config_path = tmp_path / "config.yaml" + config_path.write_text(f"config_version: {CURRENT_CONFIG_VERSION}\nproject_path: {tmp_path}\nengine: pytorch\n") + with caplog.at_level(logging.INFO, logger="dlclibrary.io.config.versioning"): + ProjectConfig.from_yaml(config_path) + + migration_records = [r for r in caplog.records if "igrating" in r.message] + assert len(migration_records) == 0 + + def test_unversioned_yaml_treated_as_v0(self, tmp_path): + """A YAML file without config_version is treated as version 0.""" + config_path = tmp_path / "config.yaml" + config_path.write_text(f"project_path: {tmp_path}\nengine: pytorch\n") + cfg = ProjectConfig.from_yaml(config_path) + assert cfg.config_version == CURRENT_CONFIG_VERSION + + +# ----------------------------------------------------------------------------- +# Fields awaiting the next schema version bump +# +# Each test here describes a field that is exposed as a read-only property today +# because promoting it to a real schema field requires bumping +# CURRENT_CONFIG_VERSION and writing the migration. They are marked strict xfail +# so that the migration landing turns them into failures, as a reminder to drop +# the property and the marker together. +# ----------------------------------------------------------------------------- + + +class TestFieldsPendingSchemaMigration: + @pytest.mark.xfail( + strict=True, + reason=( + "bboxes_pcutoff is a property until the schema version is bumped, " + "see https://github.com/DeepLabCut/DeepLabCut/pull/3470" + ), + ) + def test_bboxes_pcutoff_is_a_schema_field(self): + assert "bboxes_pcutoff" in ProjectConfig.model_fields diff --git a/tests/io/config/test_validation.py b/tests/io/config/test_validation.py new file mode 100644 index 0000000..75d337b --- /dev/null +++ b/tests/io/config/test_validation.py @@ -0,0 +1,220 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Tests for annotated types in dlclibrary.io.config.validation.""" + +import numpy as np +import pytest +from pydantic import ValidationError + +from dlclibrary.io.config import DLCBaseConfig +from dlclibrary.io.config.validation import ( + BodypartPair, + DefaultIfNone, + Fraction, + NDArrayInt, + UniqueStrList, + greater_than, + less_than, + unique_values, + validate_crop_bounds, +) + + +class ThresholdConfig(DLCBaseConfig): + """Config exercising DefaultIfNone over a constrained type.""" + + threshold: DefaultIfNone[Fraction] = 0.01 + name: DefaultIfNone[str] = "detector" + + +class ListConfig(DLCBaseConfig): + """Config exercising the list-based annotated types.""" + + bodyparts: UniqueStrList = [] + skeleton: list[BodypartPair] = [] + indices: NDArrayInt = None + + +# ------------------------------------------------------------------ +# DefaultIfNone +# ------------------------------------------------------------------ + + +class TestDefaultIfNone: + def test_none_falls_back_to_default(self): + assert ThresholdConfig(threshold=None).threshold == 0.01 + + def test_explicit_value_is_kept(self): + assert ThresholdConfig(threshold=0.6).threshold == 0.6 + + def test_falsy_value_is_kept(self): + # A plain ``value or default`` would silently turn 0.0 into the default. + assert ThresholdConfig(threshold=0.0).threshold == 0.0 + + def test_applies_to_any_wrapped_type(self): + assert ThresholdConfig(name=None).name == "detector" + assert ThresholdConfig(name="ssdlite").name == "ssdlite" + + def test_wrapped_annotation_is_not_swallowed(self): + # The generic alias must compose with Fraction, not replace it. + with pytest.raises(ValidationError): + ThresholdConfig(threshold=1.5) + + def test_none_is_normalized_on_assignment(self): + # DLCBaseConfig sets validate_assignment=True, so the validator runs here too. + cfg = ThresholdConfig(threshold=0.6) + cfg.threshold = None + assert cfg.threshold == 0.01 + + +# ------------------------------------------------------------------ +# greater_than / less_than +# ------------------------------------------------------------------ + + +class TestComparisons: + def test_greater_than_accepts_larger_value(self): + greater_than(2.0, 1.0) + + def test_greater_than_rejects_smaller_value(self): + with pytest.raises(ValueError, match="must be greater than"): + greater_than(0.5, 1.0) + + def test_greater_than_is_strict(self): + with pytest.raises(ValueError, match="must be greater than"): + greater_than(1.0, 1.0) + + def test_less_than_accepts_smaller_value(self): + less_than(1.0, 2.0, name="lower") + + def test_less_than_rejects_larger_value(self): + with pytest.raises(ValueError, match="must be less than"): + less_than(3.0, 2.0, name="lower") + + def test_less_than_is_strict(self): + with pytest.raises(ValueError, match="must be less than"): + less_than(2.0, 2.0, name="lower") + + def test_message_uses_names_when_given(self): + with pytest.raises(ValueError) as excinfo: + less_than(3, 2, name="x1", threshold_name="x2") + message = str(excinfo.value) + assert "x1 (3)" in message + assert "x2 (2)" in message + + def test_message_uses_bare_values_when_unnamed(self): + with pytest.raises(ValueError) as excinfo: + greater_than(1, 2) + assert str(excinfo.value) == "1 must be greater than 2" + + +# ------------------------------------------------------------------ +# unique_values / UniqueStrList +# ------------------------------------------------------------------ + + +class TestUniqueValues: + def test_returns_input_when_unique(self): + values = ["nose", "tail"] + assert unique_values(values) == values + + def test_rejects_duplicates(self): + with pytest.raises(ValueError, match="must be unique"): + unique_values(["nose", "nose"]) + + def test_empty_list_is_allowed(self): + assert unique_values([]) == [] + + def test_enforced_through_annotated_type(self): + assert ListConfig(bodyparts=["nose", "tail"]).bodyparts == ["nose", "tail"] + with pytest.raises(ValidationError): + ListConfig(bodyparts=["nose", "nose"]) + + +# ------------------------------------------------------------------ +# validate_crop_bounds +# ------------------------------------------------------------------ + + +class TestValidateCropBounds: + def test_all_none_is_allowed(self): + validate_crop_bounds(x1=None, x2=None, y1=None, y2=None) + + def test_ordered_bounds_are_allowed(self): + validate_crop_bounds(x1=0, x2=100, y1=0, y2=50) + + @pytest.mark.parametrize( + "bounds", + [ + {"x1": 0, "x2": None, "y1": None, "y2": None}, + {"x1": 0, "x2": 100, "y1": 0, "y2": None}, + {"x1": None, "x2": None, "y1": 0, "y2": 50}, + ], + ids=["one-set", "three-set", "y-only"], + ) + def test_partial_bounds_are_rejected(self, bounds): + with pytest.raises(ValueError, match="all be set or all be omitted"): + validate_crop_bounds(**bounds) + + def test_rejects_unordered_x(self): + with pytest.raises(ValueError, match="x1.*must be less than.*x2"): + validate_crop_bounds(x1=100, x2=0, y1=0, y2=50) + + def test_rejects_unordered_y(self): + with pytest.raises(ValueError, match="y1.*must be less than.*y2"): + validate_crop_bounds(x1=0, x2=100, y1=50, y2=0) + + def test_rejects_zero_width_crop(self): + with pytest.raises(ValueError, match="must be less than"): + validate_crop_bounds(x1=10, x2=10, y1=0, y2=50) + + +# ------------------------------------------------------------------ +# BodypartPair +# ------------------------------------------------------------------ + + +class TestBodypartPair: + def test_accepts_a_pair(self): + assert ListConfig(skeleton=[["nose", "tail"]]).skeleton == [["nose", "tail"]] + + @pytest.mark.parametrize("pair", [["nose"], ["nose", "tail", "ear"]], ids=["one", "three"]) + def test_rejects_wrong_length(self, pair): + with pytest.raises(ValidationError, match="exactly two bodyparts"): + ListConfig(skeleton=[pair]) + + def test_rejects_duplicate_bodyparts(self): + with pytest.raises(ValidationError, match="must be unique"): + ListConfig(skeleton=[["nose", "nose"]]) + + +# ------------------------------------------------------------------ +# NDArrayInt +# ------------------------------------------------------------------ + + +class TestNDArrayInt: + def test_coerces_list_to_int_array(self): + indices = ListConfig(indices=[1, 2, 3]).indices + assert isinstance(indices, np.ndarray) + assert indices.dtype == int + np.testing.assert_array_equal(indices, [1, 2, 3]) + + def test_coerces_nested_list(self): + indices = ListConfig(indices=[[1, 2], [3, 4]]).indices + assert indices.shape == (2, 2) + + def test_existing_array_is_passed_through_unchanged(self): + # The coercion short-circuits on ndarray, so a float array keeps its dtype. + array = np.array([1.5, 2.5]) + indices = ListConfig(indices=array).indices + assert indices is array + assert indices.dtype == np.float64 diff --git a/tests/io/config/test_versioning.py b/tests/io/config/test_versioning.py new file mode 100644 index 0000000..46eae95 --- /dev/null +++ b/tests/io/config/test_versioning.py @@ -0,0 +1,921 @@ +# +# DeepLabCut Toolbox (deeplabcut.org) +# © A. & M.W. Mathis Labs +# https://github.com/DeepLabCut/DeepLabCut +# +# Please see AUTHORS for contributors. +# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS +# +# Licensed under GNU Lesser General Public License v3.0 +# +"""Tests for the configuration versioning and migration system.""" + +import logging + +import pytest +from pydantic import Field, ValidationError + +from dlclibrary.io.config import DLCVersionedConfig, versioning +from dlclibrary.io.config.versioning import ( + get_config_version, + migrate_config, + register_migration, +) +from dlclibrary.deprecation import DLCDeprecationWarning + +_LOGGER_NAME = "dlclibrary.io.config.versioning" + + +# ----------------------------------------------------------------------------- +# Toy migrations (registered only for tests; v98 <-> v99) +# ----------------------------------------------------------------------------- + +_TOY_VERSION_OLD = 98 +_TOY_VERSION_NEW = 99 + +_LEGACY_FIELD = "toy_legacy_field" +_NEW_FIELD = "toy_new_field" + + +_TOY_CONFIG_TYPE = "ToyVersionedConfig" + + +@pytest.fixture(autouse=True) +def _toy_versioned_config_migrations(register_toy_migrations): + """Default v98 <-> v99 migrations for ToyVersionedConfig tests in this module.""" + register_toy_migrations(_TOY_CONFIG_TYPE) + + +# ----------------------------------------------------------------------------- +# get_config_version +# ----------------------------------------------------------------------------- + + +def test_get_config_version_(): + assert get_config_version({}) == 0 + assert get_config_version({"Task": "mytask"}) == 0 + assert get_config_version({"config_version": 1}) == 1 + assert get_config_version({"config_version": 0}) == 0 + + +# ----------------------------------------------------------------------------- +# migrate_config (no-op, errors) +# ----------------------------------------------------------------------------- + + +def test_migrate_config_same_version_returns_unchanged(): + cfg = {"a": 1, "b": 2} + assert migrate_config(cfg, config_type=_TOY_CONFIG_TYPE, target_version=0) is cfg + + +def test_migrate_config_does_not_mutate_input(): + cfg = {"config_version": _TOY_VERSION_OLD, _LEGACY_FIELD: "x"} + result = migrate_config(cfg, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_NEW) + assert cfg[_LEGACY_FIELD] == "x" + assert result is not cfg + + +def test_migrate_config_target_exceeds_current_raises(monkeypatch): + monkeypatch.setattr(versioning, "CURRENT_CONFIG_VERSION", 99) + with pytest.raises(ValueError, match="Target version .* exceeds current"): + migrate_config({"config_version": 0}, config_type=_TOY_CONFIG_TYPE, target_version=100) + + +# ----------------------------------------------------------------------------- +# Toy migration round-trip (v98 <-> v99) +# ----------------------------------------------------------------------------- + + +def test_migration_v98_to_v99_renames_toy_field(): + """Upgrade v98 -> v99: toy_legacy_field becomes toy_new_field.""" + v98 = {"config_version": _TOY_VERSION_OLD, _LEGACY_FIELD: "value", "other": 42} + v99 = migrate_config(v98, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_NEW) + assert get_config_version(v99) == _TOY_VERSION_NEW + assert _NEW_FIELD in v99 + assert v99[_NEW_FIELD] == "value" + assert _LEGACY_FIELD not in v99 + assert v99["other"] == 42 + + +def test_migration_v99_to_v98_renames_toy_field_back(): + """Downgrade v99 -> v98: toy_new_field becomes toy_legacy_field.""" + v99 = {"config_version": _TOY_VERSION_NEW, _NEW_FIELD: "value", "other": 42} + v98 = migrate_config(v99, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_OLD) + assert get_config_version(v98) == _TOY_VERSION_OLD + assert _LEGACY_FIELD in v98 + assert v98[_LEGACY_FIELD] == "value" + assert _NEW_FIELD not in v98 + assert v98["other"] == 42 + + +def test_roundtrip_v98_to_v99_to_v98_preserves_content(): + """Round-trip v98 -> v99 -> v98: content matches original v98.""" + original = { + "config_version": _TOY_VERSION_OLD, + _LEGACY_FIELD: "test_value", + "Task": "mytask", + "extra": [1, 2], + } + v99 = migrate_config(original, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_NEW) + back = migrate_config(v99, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_OLD) + assert get_config_version(back) == _TOY_VERSION_OLD + assert back[_LEGACY_FIELD] == original[_LEGACY_FIELD] + assert back["Task"] == original["Task"] + assert back["extra"] == original["extra"] + + +def test_roundtrip_v99_to_v98_to_v99_preserves_content(): + """Round-trip v99 -> v98 -> v99: content matches original v99.""" + original = { + "config_version": _TOY_VERSION_NEW, + _NEW_FIELD: "test_value", + "Task": "mytask", + "extra": [1, 2], + } + v98 = migrate_config(original, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_OLD) + back = migrate_config(v98, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_NEW) + assert get_config_version(back) == _TOY_VERSION_NEW + assert back[_NEW_FIELD] == original[_NEW_FIELD] + assert back["Task"] == original["Task"] + assert back["extra"] == original["extra"] + + +def test_roundtrip_v98_without_toy_field_unchanged(): + """Config without toy field is unchanged by v98 -> v99 -> v98.""" + original = { + "config_version": _TOY_VERSION_OLD, + "Task": "mytask", + "bodyparts": ["a", "b"], + } + v99 = migrate_config(original, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_NEW) + back = migrate_config(v99, config_type=_TOY_CONFIG_TYPE, target_version=_TOY_VERSION_OLD) + assert get_config_version(back) == _TOY_VERSION_OLD + assert back["Task"] == original["Task"] + assert back["bodyparts"] == original["bodyparts"] + + +# ----------------------------------------------------------------------------- +# DLCVersionedConfig: migration prevents validation error for renamed field +# ----------------------------------------------------------------------------- + + +@pytest.fixture +def ToyConfigWithValidField(): + class ToyConfig(DLCVersionedConfig): + config_version: int = _TOY_VERSION_NEW + valid_project_config_field: str = "" + + return ToyConfig + + +def test_config_with_legacy_field_raises_without_migration(monkeypatch, ToyConfigWithValidField): + """Legacy unknown field raises validation error when no migration is registered.""" + monkeypatch.setattr(versioning, "CURRENT_CONFIG_VERSION", _TOY_VERSION_OLD) + config_with_legacy_field = { + "config_version": _TOY_VERSION_OLD, + "this_fieldname_is_not_in_project_config": "some_value", + } + with pytest.raises((ValidationError, TypeError)): + ToyConfigWithValidField(**config_with_legacy_field) + + +def test_config_after_migration_accepts_renamed_field(monkeypatch, ToyConfigWithValidField): + """Registering a migration renames the legacy field so the model accepts the config.""" + monkeypatch.setattr(versioning, "CURRENT_CONFIG_VERSION", _TOY_VERSION_NEW) + config_with_legacy_field = { + "config_version": _TOY_VERSION_OLD, + "this_fieldname_is_not_in_project_config": "some_value", + } + + @register_migration(_TOY_VERSION_OLD, _TOY_VERSION_NEW, config_type="ToyConfig") + def _toy_migrate_legacy_to_valid_field(config: dict) -> dict: + """Test-only: rename legacy field -> valid_project_config_field.""" + if "this_fieldname_is_not_in_project_config" in config: + config["valid_project_config_field"] = config.pop("this_fieldname_is_not_in_project_config") + return config + + cfg = ToyConfigWithValidField.from_dict(config_with_legacy_field) + assert cfg.valid_project_config_field == "some_value" + + +# ----------------------------------------------------------------------------- +# Migration before alias resolution on DLCVersionedConfig +# ----------------------------------------------------------------------------- + + +class _MigrateThenAliasConfig(DLCVersionedConfig): + config_version: int = _TOY_VERSION_NEW + toy_new_field: str = Field( + default="", + json_schema_extra={"aliases": ["deprecated_alias"]}, + ) + + +@pytest.fixture +def _migrate_then_alias_current(register_toy_migrations): + register_toy_migrations("_MigrateThenAliasConfig") + + +def test_migration_then_alias_legacy_version_key(_migrate_then_alias_current): + """v98 legacy key is renamed by migration, then validated.""" + cfg = _MigrateThenAliasConfig.from_dict( + { + "config_version": _TOY_VERSION_OLD, + _LEGACY_FIELD: "from_migration", + } + ) + assert cfg.toy_new_field == "from_migration" + + +def test_migration_then_alias_same_version_deprecated_name(_migrate_then_alias_current): + """Same-version deprecated alias is resolved after migration (with a warning).""" + with pytest.warns(DLCDeprecationWarning, match="deprecated_alias"): + cfg = _MigrateThenAliasConfig.model_validate( + { + "config_version": _TOY_VERSION_NEW, + "deprecated_alias": "from_alias", + } + ) + assert cfg.toy_new_field == "from_alias" + + +def test_migration_then_alias_on_old_version_with_deprecated_alias( + _migrate_then_alias_current, +): + """Migrate to current version, then resolve alias when v98 used the alias key.""" + with pytest.warns(DLCDeprecationWarning, match="deprecated_alias"): + cfg = _MigrateThenAliasConfig.model_validate( + { + "config_version": _TOY_VERSION_OLD, + "deprecated_alias": "both_steps", + } + ) + assert cfg.toy_new_field == "both_steps" + + +def test_from_dict_runs_migration(_migrate_then_alias_current): + """Legacy dicts loaded via from_dict are migrated before validation.""" + + class _KwOnlyVersioned(DLCVersionedConfig): + config_version: int = _TOY_VERSION_NEW + toy_new_field: str = "" + + @register_migration(_TOY_VERSION_OLD, _TOY_VERSION_NEW, config_type="_KwOnlyVersioned") + def _kw_migrate_v98_to_v99(config: dict) -> dict: + if _LEGACY_FIELD in config: + config[_NEW_FIELD] = config.pop(_LEGACY_FIELD) + return config + + cfg = _KwOnlyVersioned.from_dict( + { + "config_version": _TOY_VERSION_OLD, + _LEGACY_FIELD: "kwargs_value", + } + ) + assert cfg.toy_new_field == "kwargs_value" + assert cfg.config_version == _TOY_VERSION_NEW + + +def test_constructor_does_not_run_migration(_migrate_then_alias_current): + """Typed construction must not run the legacy migration chain.""" + + class _KwOnlyVersioned(DLCVersionedConfig): + config_version: int = _TOY_VERSION_NEW + toy_new_field: str = "" + + @register_migration(_TOY_VERSION_OLD, _TOY_VERSION_NEW, config_type="_KwOnlyVersioned") + def _kw_migrate_v98_to_v99(config: dict) -> dict: + if _LEGACY_FIELD in config: + config[_NEW_FIELD] = config.pop(_LEGACY_FIELD) + return config + + with pytest.raises(ValidationError): + _KwOnlyVersioned( + config_version=_TOY_VERSION_OLD, + **{_LEGACY_FIELD: "kwargs_value"}, + ) + + +# ============================================================================= +# Multi-step migration chain (v50 → v51 → v52 → v53) +# +# Each step renames one field, so we can verify individual steps, full chains, +# downgrades, missing intermediates, missing fields, and mid-chain errors. +# ============================================================================= + +_V50, _V51, _V52, _V53 = 50, 51, 52, 53 +_CHAIN_CONFIG_TYPE = "ChainConfig" + + +def _register_chain_migrations(): + """Register a v50↔v53 migration chain (upgrades and downgrades).""" + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _chain_v50_to_v51(config: dict) -> dict: + if "field_a_old" in config: + config["field_a_new"] = config.pop("field_a_old") + return config + + @register_migration(_V51, _V52, config_type=_CHAIN_CONFIG_TYPE) + def _chain_v51_to_v52(config: dict) -> dict: + if "field_b_old" in config: + config["field_b_new"] = config.pop("field_b_old") + return config + + @register_migration(_V52, _V53, config_type=_CHAIN_CONFIG_TYPE) + def _chain_v52_to_v53(config: dict) -> dict: + if "field_c_old" in config: + config["field_c_new"] = config.pop("field_c_old") + return config + + @register_migration(_V51, _V50, config_type=_CHAIN_CONFIG_TYPE) + def _chain_v51_to_v50(config: dict) -> dict: + if "field_a_new" in config: + config["field_a_old"] = config.pop("field_a_new") + return config + + @register_migration(_V52, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _chain_v52_to_v51(config: dict) -> dict: + if "field_b_new" in config: + config["field_b_old"] = config.pop("field_b_new") + return config + + @register_migration(_V53, _V52, config_type=_CHAIN_CONFIG_TYPE) + def _chain_v53_to_v52(config: dict) -> dict: + if "field_c_new" in config: + config["field_c_old"] = config.pop("field_c_new") + return config + + +@pytest.fixture +def chain_migrations(monkeypatch): + """Register v50↔v53 chain and set CURRENT_CONFIG_VERSION=53.""" + monkeypatch.setattr(versioning, "CURRENT_CONFIG_VERSION", _V53) + _register_chain_migrations() + + +# ----------------------------------------------------------------------------- +# Multi-step upgrade chains +# ----------------------------------------------------------------------------- + + +class TestMultiStepUpgrade: + """Tests for chained upgrade migrations across multiple versions.""" + + def test_upgrade_v50_to_v53_applies_all_steps(self, chain_migrations): + """Full chain v50→v53 renames all three fields.""" + cfg = { + "config_version": _V50, + "field_a_old": "a", + "field_b_old": "b", + "field_c_old": "c", + "untouched": 42, + } + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert result["config_version"] == _V53 + assert result["field_a_new"] == "a" + assert result["field_b_new"] == "b" + assert result["field_c_new"] == "c" + assert result["untouched"] == 42 + assert "field_a_old" not in result + assert "field_b_old" not in result + assert "field_c_old" not in result + + def test_upgrade_v50_to_v52_partial_chain(self, chain_migrations): + """Partial chain v50→v52 renames only fields a and b.""" + cfg = { + "config_version": _V50, + "field_a_old": "a", + "field_b_old": "b", + "field_c_old": "c", + } + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V52) + assert result["config_version"] == _V52 + assert result["field_a_new"] == "a" + assert result["field_b_new"] == "b" + assert result["field_c_old"] == "c" # untouched — v52→v53 not applied + + def test_upgrade_single_step_v51_to_v52(self, chain_migrations): + """Single step in the middle of the chain works.""" + cfg = {"config_version": _V51, "field_b_old": "b"} + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V52) + assert result["config_version"] == _V52 + assert result["field_b_new"] == "b" + + def test_upgrade_does_not_mutate_original(self, chain_migrations): + """Original config dict is untouched after multi-step upgrade.""" + cfg = { + "config_version": _V50, + "field_a_old": "a", + "field_b_old": "b", + "field_c_old": "c", + } + original_copy = cfg.copy() + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert cfg == original_copy + + +# ----------------------------------------------------------------------------- +# Multi-step downgrade chains +# ----------------------------------------------------------------------------- + + +class TestMultiStepDowngrade: + """Tests for chained downgrade migrations across multiple versions.""" + + def test_downgrade_v53_to_v50_applies_all_steps(self, chain_migrations): + """Full downgrade v53→v50 reverses all three field renames.""" + cfg = { + "config_version": _V53, + "field_a_new": "a", + "field_b_new": "b", + "field_c_new": "c", + "untouched": 42, + } + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + assert result["config_version"] == _V50 + assert result["field_a_old"] == "a" + assert result["field_b_old"] == "b" + assert result["field_c_old"] == "c" + assert result["untouched"] == 42 + + def test_downgrade_v53_to_v51_partial(self, chain_migrations): + """Partial downgrade v53→v51 reverses only fields c and b.""" + cfg = { + "config_version": _V53, + "field_a_new": "a", + "field_b_new": "b", + "field_c_new": "c", + } + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert result["config_version"] == _V51 + assert result["field_a_new"] == "a" # untouched — v51→v50 not applied + assert result["field_b_old"] == "b" + assert result["field_c_old"] == "c" + + def test_downgrade_does_not_mutate_original(self, chain_migrations): + """Original config dict is untouched after multi-step downgrade.""" + cfg = { + "config_version": _V53, + "field_a_new": "a", + "field_b_new": "b", + "field_c_new": "c", + } + original_copy = cfg.copy() + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + assert cfg == original_copy + + +# ----------------------------------------------------------------------------- +# Multi-step round-trips +# ----------------------------------------------------------------------------- + + +class TestMultiStepRoundTrip: + """Verify that upgrade→downgrade and downgrade→upgrade round-trips preserve data.""" + + def test_roundtrip_v50_to_v53_and_back(self, chain_migrations): + original = { + "config_version": _V50, + "field_a_old": "a", + "field_b_old": "b", + "field_c_old": "c", + "extra": [1, 2, 3], + } + upgraded = migrate_config(original, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + back = migrate_config(upgraded, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + assert back["config_version"] == _V50 + assert back["field_a_old"] == "a" + assert back["field_b_old"] == "b" + assert back["field_c_old"] == "c" + assert back["extra"] == [1, 2, 3] + + def test_roundtrip_v53_to_v50_and_back(self, chain_migrations): + original = { + "config_version": _V53, + "field_a_new": "a", + "field_b_new": "b", + "field_c_new": "c", + } + downgraded = migrate_config(original, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + back = migrate_config(downgraded, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert back["config_version"] == _V53 + assert back["field_a_new"] == "a" + assert back["field_b_new"] == "b" + assert back["field_c_new"] == "c" + + +# ----------------------------------------------------------------------------- +# Missing intermediate migrations +# ----------------------------------------------------------------------------- + + +class TestMissingIntermediateMigration: + """Verify that gaps in the migration chain produce clear errors.""" + + def test_missing_middle_step_raises_with_context(self, chain_migrations): + """v50→v53 with v51→v52 removed raises naming the missing step.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V51, _V52)] + cfg = {"config_version": _V50} + with pytest.raises(ValueError, match=r"No migration registered for 'ChainConfig' v51 -> v52"): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + + def test_missing_first_step_raises(self, chain_migrations): + """v50→v53 with v50→v51 removed raises naming the first step.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V50, _V51)] + cfg = {"config_version": _V50} + with pytest.raises(ValueError, match=r"No migration registered for 'ChainConfig' v50 -> v51"): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + + def test_missing_downgrade_step_raises(self, chain_migrations): + """v53→v50 with v52→v51 removed raises naming the missing step.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V52, _V51)] + cfg = {"config_version": _V53} + with pytest.raises(ValueError, match=r"No migration registered for 'ChainConfig' v52 -> v51"): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + + +# ----------------------------------------------------------------------------- +# Migration with missing / extra fields +# ----------------------------------------------------------------------------- + + +class TestMissingFields: + """Migration functions should tolerate configs that lack optional fields.""" + + def test_upgrade_with_no_renameable_fields(self, chain_migrations): + """Config without any of the fields the migrations rename still upgrades.""" + cfg = {"config_version": _V50, "unrelated": "data"} + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert result["config_version"] == _V53 + assert result["unrelated"] == "data" + assert "field_a_new" not in result + assert "field_b_new" not in result + assert "field_c_new" not in result + + def test_upgrade_with_partial_fields(self, chain_migrations): + """Only the fields present are renamed; absent ones are not invented.""" + cfg = {"config_version": _V50, "field_a_old": "a"} + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert result["field_a_new"] == "a" + assert "field_b_new" not in result + assert "field_c_new" not in result + + def test_downgrade_with_missing_fields(self, chain_migrations): + """Downgrade with missing fields: absent fields are simply absent.""" + cfg = {"config_version": _V53, "field_c_new": "c"} + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + assert result["config_version"] == _V50 + assert result["field_c_old"] == "c" + assert "field_a_old" not in result + assert "field_b_old" not in result + + def test_extra_unknown_fields_are_preserved(self, chain_migrations): + """Fields not touched by any migration survive the full chain.""" + cfg = { + "config_version": _V50, + "field_a_old": "a", + "totally_unknown": {"nested": True}, + } + result = migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert result["totally_unknown"] == {"nested": True} + + +# ----------------------------------------------------------------------------- +# Error handling mid-migration +# ----------------------------------------------------------------------------- + + +class TestMidMigrationError: + """If a migration function raises, the error should identify the failing step.""" + + def test_error_in_second_step_reports_step(self, chain_migrations): + """An exception in v51→v52 wraps with the step identifier.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V51, _V52)] + + @register_migration(_V51, _V52, config_type=_CHAIN_CONFIG_TYPE) + def _broken_v51_to_v52(config: dict) -> dict: + raise RuntimeError("something broke in v51→v52") + + cfg = {"config_version": _V50} + with pytest.raises( + RuntimeError, + match=r"Migration for 'ChainConfig' v51 -> v52 failed.*something broke", + ): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + + def test_error_preserves_original_config(self, chain_migrations): + """Original config is not mutated even when a mid-chain step fails.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V52, _V53)] + + @register_migration(_V52, _V53, config_type=_CHAIN_CONFIG_TYPE) + def _broken_v52_to_v53(config: dict) -> dict: + raise ValueError("boom") + + cfg = { + "config_version": _V50, + "field_a_old": "a", + "field_b_old": "b", + "field_c_old": "c", + } + original_copy = cfg.copy() + with pytest.raises(ValueError, match="boom"): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert cfg == original_copy + + def test_error_preserves_exception_type(self, chain_migrations): + """The re-raised exception keeps its original type (not generic RuntimeError).""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V51, _V52)] + + @register_migration(_V51, _V52, config_type=_CHAIN_CONFIG_TYPE) + def _type_error_step(config: dict) -> dict: + raise TypeError("wrong type for field X") + + cfg = {"config_version": _V50} + with pytest.raises(TypeError, match="Migration for 'ChainConfig' v51 -> v52 failed"): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + + +# ----------------------------------------------------------------------------- +# register_migration validation +# ----------------------------------------------------------------------------- + + +class TestRegisterMigrationValidation: + """Edge-case validation for the registration decorator.""" + + def test_negative_from_version_raises(self): + with pytest.raises(ValueError, match="non-negative"): + + @register_migration(-1, 0, config_type=_CHAIN_CONFIG_TYPE) + def _bad(config): + return config + + def test_negative_to_version_raises(self): + with pytest.raises(ValueError, match="non-negative"): + + @register_migration(0, -1, config_type=_CHAIN_CONFIG_TYPE) + def _bad(config): + return config + + def test_same_version_raises(self): + with pytest.raises(ValueError, match="must differ"): + + @register_migration(5, 5, config_type=_CHAIN_CONFIG_TYPE) + def _bad(config): + return config + + def test_duplicate_registration_raises(self): + """Registering the same (config_type, from, to) triple twice raises.""" + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _first(config): + return config + + with pytest.raises(ValueError, match="Duplicate migration"): + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _second(config): + return config + + +# ----------------------------------------------------------------------------- +# Logging +# ----------------------------------------------------------------------------- + + +class TestMigrationLogging: + """Verify that migration logging reports field changes.""" + + def test_upgrade_logs_info_start_and_complete(self, caplog, chain_migrations): + """INFO logs report migration start (with config type, direction) and completion.""" + cfg = {"config_version": _V50, "field_a_old": "a"} + with caplog.at_level(logging.INFO, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert "Migrating ChainConfig from version 50 to 53 (upgrade)" in caplog.text + assert "Migration complete: ChainConfig is now at version 53" in caplog.text + + def test_downgrade_logs_info_direction(self, caplog, chain_migrations): + """INFO logs report 'downgrade' direction.""" + cfg = {"config_version": _V53, "field_c_new": "c"} + with caplog.at_level(logging.INFO, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + assert "Migrating ChainConfig from version 53 to 50 (downgrade)" in caplog.text + assert "Migration complete: ChainConfig is now at version 50" in caplog.text + + def test_same_version_no_log(self, caplog, chain_migrations): + """When source == target, nothing is logged.""" + cfg = {"config_version": _V50} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + assert caplog.text == "" + + def test_debug_logs_field_rename(self, caplog, chain_migrations): + """DEBUG logs report removed and added fields for a rename.""" + cfg = {"config_version": _V50, "field_a_old": "a"} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "Removed field 'field_a_old'" in caplog.text + assert "Added field 'field_a_new'" in caplog.text + + def test_debug_logs_updated_field(self, caplog, chain_migrations): + """DEBUG logs report a field whose value changed in-place.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V50, _V51)] + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _mutate_field(config: dict) -> dict: + config["keep_me"] = config.get("keep_me", 0) + 100 + return config + + cfg = {"config_version": _V50, "keep_me": 1} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "Updated field 'keep_me': 1 -> 101" in caplog.text + + def test_debug_logs_no_field_changes(self, caplog, chain_migrations): + """When migration changes nothing, DEBUG reports 'No field changes'.""" + cfg = {"config_version": _V50} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "No field changes" in caplog.text + + def test_debug_logs_applying_migration_function_name(self, caplog, chain_migrations): + """DEBUG logs include the config type and qualified name of each migration function.""" + cfg = {"config_version": _V50, "field_a_old": "a"} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "Applying migration ChainConfig v50 -> v51" in caplog.text + + def test_multi_step_upgrade_logs_all_steps(self, caplog, chain_migrations): + """A multi-step upgrade logs each intermediate step.""" + cfg = { + "config_version": _V50, + "field_a_old": "a", + "field_b_old": "b", + "field_c_old": "c", + } + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V53) + assert "Applying migration ChainConfig v50 -> v51" in caplog.text + assert "Applying migration ChainConfig v51 -> v52" in caplog.text + assert "Applying migration ChainConfig v52 -> v53" in caplog.text + + def test_info_not_shown_when_no_migration(self, caplog, chain_migrations): + """When no migration is needed, no INFO messages are emitted.""" + cfg = {"config_version": _V50} + with caplog.at_level(logging.INFO, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V50) + info_records = [r for r in caplog.records if r.levelno >= logging.INFO] + assert len(info_records) == 0 + + def test_debug_logs_nested_dict_added_key(self, caplog, chain_migrations): + """DEBUG logs report a key added inside a nested dict.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V50, _V51)] + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _add_nested(config: dict) -> dict: + config.setdefault("video_sets", {}) + config["video_sets"]["new_key"] = "new_value" + return config + + cfg = {"config_version": _V50, "video_sets": {}} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "Added field 'video_sets.new_key' = 'new_value'" in caplog.text + + def test_debug_logs_nested_dict_removed_key(self, caplog, chain_migrations): + """DEBUG logs report a key removed inside a nested dict.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V50, _V51)] + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _remove_nested(config: dict) -> dict: + config["opts"].pop("old_opt") + return config + + cfg = {"config_version": _V50, "opts": {"old_opt": 1, "keep": 2}} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "Removed field 'opts.old_opt' (was: 1)" in caplog.text + + def test_debug_logs_nested_dict_updated_value(self, caplog, chain_migrations): + """DEBUG logs report a value changed inside a nested dict.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V50, _V51)] + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _update_nested(config: dict) -> dict: + config["settings"]["threshold"] = 0.9 + return config + + cfg = {"config_version": _V50, "settings": {"threshold": 0.5}} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "Updated field 'settings.threshold': 0.5 -> 0.9" in caplog.text + + def test_debug_logs_deeply_nested_change(self, caplog, chain_migrations): + """DEBUG logs report changes multiple levels deep.""" + del versioning._MIGRATIONS[(_CHAIN_CONFIG_TYPE, _V50, _V51)] + + @register_migration(_V50, _V51, config_type=_CHAIN_CONFIG_TYPE) + def _deep_change(config: dict) -> dict: + config["a"]["b"]["c"] = "new" + return config + + cfg = {"config_version": _V50, "a": {"b": {"c": "old"}}} + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + migrate_config(cfg, config_type=_CHAIN_CONFIG_TYPE, target_version=_V51) + assert "Updated field 'a.b.c': 'old' -> 'new'" in caplog.text + + +# ----------------------------------------------------------------------------- +# DLCVersionedConfig: from_yaml / to_yaml integration with migration +# ----------------------------------------------------------------------------- + + +class _FileVersionedConfig(DLCVersionedConfig): + config_version: int = _TOY_VERSION_NEW + toy_new_field: str = "" + other: str = "default" + + +class TestFromYamlMigration: + """from_yaml and to_yaml must interact correctly with the migration system.""" + + @pytest.fixture(autouse=True) + def _file_versioned_migrations(self, register_toy_migrations): + register_toy_migrations("_FileVersionedConfig") + + def test_from_yaml_with_old_version_runs_migration(self, tmp_path): + """A YAML file at an old config_version is migrated on load.""" + path = tmp_path / "config.yaml" + path.write_text(f"config_version: {_TOY_VERSION_OLD}\n{_LEGACY_FIELD}: migrated_value\nother: kept\n") + cfg = _FileVersionedConfig.from_yaml(path) + assert cfg.toy_new_field == "migrated_value" + assert cfg.other == "kept" + assert cfg.config_version == _TOY_VERSION_NEW + + def test_from_yaml_current_version_no_migration(self, tmp_path): + """A YAML file already at the current version loads without migration.""" + path = tmp_path / "config.yaml" + path.write_text(f"config_version: {_TOY_VERSION_NEW}\ntoy_new_field: already_current\n") + cfg = _FileVersionedConfig.from_yaml(path) + assert cfg.toy_new_field == "already_current" + + def test_to_yaml_persists_new_config_version(self, tmp_path): + """After migration, to_yaml writes the new config_version to disk.""" + path = tmp_path / "config.yaml" + path.write_text(f"config_version: {_TOY_VERSION_OLD}\n{_LEGACY_FIELD}: value\n") + cfg = _FileVersionedConfig.from_yaml(path) + cfg.to_yaml(path) + + from dlclibrary.io.config.utils import read_config_as_dict + + saved = read_config_as_dict(path) + assert saved["config_version"] == _TOY_VERSION_NEW + assert "toy_new_field" in saved + assert _LEGACY_FIELD not in saved + + def test_to_yaml_then_from_yaml_roundtrip(self, tmp_path): + """Writing a migrated config and reloading it should be a perfect no-op.""" + path = tmp_path / "config.yaml" + path.write_text(f"config_version: {_TOY_VERSION_OLD}\n{_LEGACY_FIELD}: round_trip\nother: preserved\n") + cfg_first = _FileVersionedConfig.from_yaml(path) + cfg_first.to_yaml(path) + cfg_second = _FileVersionedConfig.from_yaml(path) + assert cfg_second.toy_new_field == "round_trip" + assert cfg_second.other == "preserved" + assert cfg_second.config_version == _TOY_VERSION_NEW + assert not cfg_second.is_dirty + + +# ----------------------------------------------------------------------------- +# DLCVersionedConfig compatibility with validate_assignment +# ----------------------------------------------------------------------------- + + +class ValidatedMigratingConfig(DLCVersionedConfig): + name: str = "default" + count: int = 0 + + +class TestVersionedConfigValidateAssignment: + """Versioned config must not interfere with validate_assignment (regression).""" + + @pytest.fixture(autouse=True) + def _default_version(self, monkeypatch): + monkeypatch.setattr(versioning, "CURRENT_CONFIG_VERSION", 0) + + def test_valid_assignment_takes_effect(self): + cfg = ValidatedMigratingConfig() + cfg.count = 42 + assert cfg.count == 42 + + def test_invalid_assignment_raises(self): + cfg = ValidatedMigratingConfig() + with pytest.raises(ValidationError): + cfg.count = "not-an-int" + assert cfg.count == 0 + + def test_coercion_applied_on_assignment(self): + cfg = ValidatedMigratingConfig() + cfg.count = True # coerced to 1 + assert cfg.count == 1 From ad8a6a4e9470aa3faabd889714b9cec19b24ede7 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 15 Oct 2026 10:51:33 +0200 Subject: [PATCH 4/5] Delete test_owed.py --- tests/io/config/test_owed.py | 30 ------------------------------ 1 file changed, 30 deletions(-) delete mode 100644 tests/io/config/test_owed.py diff --git a/tests/io/config/test_owed.py b/tests/io/config/test_owed.py deleted file mode 100644 index 970d241..0000000 --- a/tests/io/config/test_owed.py +++ /dev/null @@ -1,30 +0,0 @@ -# -# DeepLabCut Toolbox (deeplabcut.org) -# © A. & M.W. Mathis Labs -# https://github.com/DeepLabCut/DeepLabCut -# -# Please see AUTHORS for contributors. -# https://github.com/DeepLabCut/DeepLabCut/blob/master/AUTHORS -# -# Licensed under GNU Lesser General Public License v3.0 -# -"""Owed changes relative to DeepLabCut's copy of the config code.""" - -from pathlib import Path - -import pytest - -from dlclibrary.io.config import ProjectConfig - - -@pytest.mark.xfail( - strict=True, - raises=AssertionError, - reason="ProjectConfig._post_yaml_load_updates replaces project_path on every from_yaml, as in DeepLabCut", -) -def test_a_plain_read_leaves_project_path_as_stored(tmp_path): - stored = tmp_path / "elsewhere" - config_path = tmp_path / "config.yaml" - config_path.write_text(f"project_path: {stored.as_posix()}\n") - - assert ProjectConfig.from_yaml(config_path).project_path == Path(stored.as_posix()) From 7c01ce70393bff9db3076a20ccdef1bade0921c4 Mon Sep 17 00:00:00 2001 From: Cyril Achard Date: Thu, 15 Oct 2026 16:14:21 +0200 Subject: [PATCH 5/5] Update pyproject.toml --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index bd0bf55..c647b82 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ classifiers = [ dependencies = [ "huggingface_hub>=0.11.1", "numpy>=1.22.4", - "pydantic>=2,<3", + "pydantic>=2.1,<3", "ruamel.yaml>=0.16", "typing_extensions", ]