Promote CLI override parsing to config loader

netengine/cli/main.py

@@ -10,8 +10,6 @@
 from typing import Any
 
 import click
-import yaml
-
 from netengine.cli.doctor import doctor
 from netengine.diagnostic.preflight import (
     DoctorCheckResult,
@@ -20,6 +18,7 @@
     run_preflight,
 )
 from netengine.cli.env import db_url_from_env
+from netengine.config.loader import ConfigOverrideError, parse_dotted_overrides
 from netengine.core.migrations import MigrationService, MigrationStatus
 from netengine.core.orchestrator import Orchestrator
 from netengine.core.state import RuntimeState, get_state_file
@@ -48,35 +47,10 @@
 
 def _parse_set_overrides(set_values: tuple[str, ...]) -> dict[str, Any]:
     """Convert repeatable dotted key=value CLI overrides into a nested dictionary."""
-    overrides: dict[str, Any] = {}
-
-    for item in set_values:
-        key, separator, raw_value = item.partition("=")
-        if not separator:
-            raise click.BadParameter("must be in key=value form", param_hint="--set")
-
-        parts = key.split(".")
-        if any(part == "" for part in parts):
-            raise click.BadParameter("keys must be non-empty dotted paths", param_hint="--set")
-
-        value = yaml.safe_load(raw_value)
-        cursor = overrides
-        for part in parts[:-1]:
-            existing = cursor.get(part)
-            if existing is None:
-                nested: dict[str, Any] = {}
-                cursor[part] = nested
-                cursor = nested
-            elif isinstance(existing, dict):
-                cursor = existing
-            else:
-                raise click.BadParameter(
-                    f"cannot set nested key under non-object path '{part}'",
-                    param_hint="--set",
-                )
-        cursor[parts[-1]] = value
-
-    return overrides
+    try:
+        return parse_dotted_overrides(set_values)
+    except ConfigOverrideError as exc:
+        raise click.BadParameter(str(exc), param_hint="--set") from exc
 
 
 def _load_spec_for_cli(

netengine/config/loader.py

@@ -1,11 +1,62 @@
 """Configuration loading and merging utilities."""
 
+from collections.abc import Iterable
 from pathlib import Path
 from typing import Any, Optional, Type, TypeVar, cast
 
 import yaml
 from omegaconf import OmegaConf
 
+class ConfigOverrideError(ValueError):
+    """Raised when a dotted configuration override cannot be parsed."""
+
+
+def parse_dotted_overrides(values: Iterable[str]) -> dict[str, Any]:
+    """Convert dotted ``key=value`` overrides into a nested dictionary.
+
+    Values are parsed with ``yaml.safe_load`` so CLI strings such as
+    ``true``, ``42``, or ``[a, b]`` become YAML-native scalars/lists.
+
+    Args:
+        values: Override strings in dotted ``key=value`` form.
+
+    Returns:
+        Nested dictionary suitable for merging into a configuration.
+
+    Raises:
+        ConfigOverrideError: If an override is malformed or attempts to nest
+            beneath a scalar value already set by another override.
+    """
+    overrides: dict[str, Any] = {}
+
+    for item in values:
+        key, separator, raw_value = item.partition("=")
+        if not separator:
+            raise ConfigOverrideError("must be in key=value form")
+
+        parts = key.split(".")
+        if any(part == "" for part in parts):
+            raise ConfigOverrideError("keys must be non-empty dotted paths")
+
+        value = yaml.safe_load(raw_value)
+        cursor = overrides
+        for part in parts[:-1]:
+            existing = cursor.get(part)
+            if existing is None:
+                nested: dict[str, Any] = {}
+                cursor[part] = nested
+                cursor = nested
+            elif isinstance(existing, dict):
+                cursor = existing
+            else:
+                raise ConfigOverrideError(
+                    f"cannot set nested key under non-object path '{part}'"
+                )
+        cursor[parts[-1]] = value
+
+    return overrides
+
+
 T = TypeVar("T")
 
 

tests/test_config.py

@@ -1,13 +1,12 @@
 """Tests for configuration management."""
 
-import os
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
 import pytest
 import yaml
 
-from netengine.config.loader import ConfigLoader
+from netengine.config.loader import ConfigLoader, ConfigOverrideError, parse_dotted_overrides
 from netengine.config.spec_config import SpecConfig
 from netengine.spec.loader import (
     SpecLoadError,
@@ -169,6 +168,58 @@ def dev_spec_file(temp_spec_dir: Path) -> Path:
     return spec_file
 
 
+class TestParseDottedOverrides:
+    """Tests for reusable dotted override parsing."""
+
+    def test_nested_overrides(self) -> None:
+        """Dotted paths should become nested dictionaries."""
+        overrides = parse_dotted_overrides([
+            "metadata.environment=dev",
+            "operator.api.port=9090",
+        ])
+
+        assert overrides == {
+            "metadata": {"environment": "dev"},
+            "operator": {"api": {"port": 9090}},
+        }
+
+    def test_yaml_scalar_and_list_parsing(self) -> None:
+        """Override values should be parsed with YAML semantics."""
+        overrides = parse_dotted_overrides([
+            "feature.enabled=true",
+            "limits.retries=3",
+            "dns.servers=[1.1.1.1, 8.8.8.8]",
+            "empty=null",
+        ])
+
+        assert overrides["feature"]["enabled"] is True
+        assert overrides["limits"]["retries"] == 3
+        assert overrides["dns"]["servers"] == ["1.1.1.1", "8.8.8.8"]
+        assert overrides["empty"] is None
+
+    @pytest.mark.parametrize(
+        "value, message",
+        [
+            ("metadata.name", "must be in key=value form"),
+            ("metadata..name=test", "keys must be non-empty dotted paths"),
+            (".metadata.name=test", "keys must be non-empty dotted paths"),
+            ("metadata.name.=test", "keys must be non-empty dotted paths"),
+        ],
+    )
+    def test_malformed_inputs(self, value: str, message: str) -> None:
+        """Malformed override strings should be rejected."""
+        with pytest.raises(ConfigOverrideError, match=message):
+            parse_dotted_overrides([value])
+
+    def test_scalar_nested_conflict(self) -> None:
+        """Nested keys cannot be added under an existing scalar value."""
+        with pytest.raises(
+            ConfigOverrideError,
+            match="cannot set nested key under non-object path 'metadata'",
+        ):
+            parse_dotted_overrides(["metadata=scalar", "metadata.name=test-network"])
+
+
 class TestConfigLoader:
     """Tests for ConfigLoader utilities."""