Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 68 additions & 5 deletions netengine/config/spec_config.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,21 @@
"""Infrastructure specification configuration and loading."""

from pathlib import Path
from typing import Any, Optional
from typing import TYPE_CHECKING, Any, Optional

from omegaconf import OmegaConf

from netengine.config.loader import ConfigLoader

if TYPE_CHECKING:
from netengine.spec.models import NetEngineSpec


class SpecConfig:
"""Load and manage infrastructure specifications with composition support."""

@staticmethod
def load(
def load_raw(
spec_path: Path | str,
base_path: Optional[Path | str] = None,
overrides: Optional[dict[str, Any]] = None,
Expand All @@ -25,7 +28,7 @@ def load(
overrides: Optional overrides dictionary

Returns:
Loaded and merged specification as dictionary
Loaded and merged raw specification as dictionary
"""
spec_path = Path(spec_path)

Expand All @@ -42,7 +45,34 @@ def load(
return spec_dict

@staticmethod
def load_environment_variants(
def load(
spec_path: Path | str,
base_path: Optional[Path | str] = None,
overrides: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
"""Load spec as a raw dictionary.

This backwards-compatible alias preserves the historical ``SpecConfig.load``
behavior. Prefer ``load_raw`` for new raw merge callers or
``load_validated`` when a validated ``NetEngineSpec`` is required.
"""
return SpecConfig.load_raw(spec_path, base_path=base_path, overrides=overrides)

@staticmethod
def load_validated(
spec_path: Path | str,
base_path: Optional[Path | str] = None,
overrides: Optional[dict[str, Any]] = None,
validate_feature_states: bool = True,
) -> "NetEngineSpec":
"""Load, compose, and validate a spec into a ``NetEngineSpec``."""
from netengine.spec.loader import validate_spec_data

spec_dict = SpecConfig.load_raw(spec_path, base_path=base_path, overrides=overrides)
return validate_spec_data(spec_dict, validate_feature_states=validate_feature_states)

@staticmethod
def load_environment_variants_raw(
base_spec: Path | str,
environment: str = "dev",
overrides: Optional[dict[str, Any]] = None,
Expand All @@ -59,7 +89,7 @@ def load_environment_variants(
overrides: Optional additional overrides

Returns:
Merged specification
Merged raw specification dictionary
"""
base_path = Path(base_spec)
base_dir = base_path.parent
Expand All @@ -78,6 +108,39 @@ def load_environment_variants(

return spec_dict

@staticmethod
def load_environment_variants(
base_spec: Path | str,
environment: str = "dev",
overrides: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
"""Load environment variants as a raw dictionary.

This backwards-compatible alias preserves the historical
``SpecConfig.load_environment_variants`` behavior. Prefer
``load_environment_variants_raw`` for new raw merge callers or
``load_environment_variant_validated`` when a validated ``NetEngineSpec``
is required.
"""
return SpecConfig.load_environment_variants_raw(
base_spec, environment=environment, overrides=overrides
)

@staticmethod
def load_environment_variant_validated(
base_spec: Path | str,
environment: str = "dev",
overrides: Optional[dict[str, Any]] = None,
validate_feature_states: bool = True,
) -> "NetEngineSpec":
"""Load an environment variant and validate it into a ``NetEngineSpec``."""
from netengine.spec.loader import validate_spec_data

spec_dict = SpecConfig.load_environment_variants_raw(
base_spec, environment=environment, overrides=overrides
)
return validate_spec_data(spec_dict, validate_feature_states=validate_feature_states)

@staticmethod
def to_dict(spec_obj: Any) -> dict[str, Any]:
"""Convert OmegaConf spec object to dictionary.
Expand Down
33 changes: 31 additions & 2 deletions netengine/spec/loader.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
"""YAML spec loading and validation with OmegaConf composition support."""

import ipaddress
import netengine.logs as logs
from collections.abc import Iterator
from pathlib import Path
from typing import Any, Optional

import yaml
from pydantic import ValidationError

import netengine.logs as logs
from netengine.config.loader import ConfigLoader
from netengine.spec.models import SUPPORTED_SPEC_SCHEMA_VERSIONS, NetEngineSpec
from netengine.spec.types import GatewayRealInternetMode
Expand Down Expand Up @@ -218,6 +218,35 @@ def _validate_spec_data(
return spec


def load_spec(yaml_path: str | Path, *, validate_feature_states: bool = True) -> NetEngineSpec:
"""Load and validate a NetEngine YAML specification.

Args:
data: Composed raw specification dictionary
validate_feature_states: Whether to enforce feature-state metadata

Returns:
Validated, immutable NetEngineSpec object

Raises:
SpecLoadError: If the composed spec data is invalid
"""
yaml_path = Path(yaml_path)

if not yaml_path.exists():
raise SpecLoadError(f"Spec file not found: {yaml_path}")

try:
with open(yaml_path) as f:
data = yaml.safe_load(f)
except yaml.YAMLError as e:
raise SpecLoadError(f"Failed to parse YAML: {e}")
except IOError as e:
raise SpecLoadError(f"Failed to read file: {e}")

return _validate_spec_data(data, validate_feature_states=validate_feature_states)


def load_spec(yaml_path: str | Path, *, validate_feature_states: bool = True) -> NetEngineSpec:
"""Load and validate a NetEngine YAML specification.

Expand All @@ -243,7 +272,7 @@ def load_spec(yaml_path: str | Path, *, validate_feature_states: bool = True) ->
except IOError as e:
raise SpecLoadError(f"Failed to read file: {e}")

return _validate_spec_data(data, validate_feature_states=validate_feature_states)
return validate_spec_data(data, validate_feature_states=validate_feature_states)


def load_spec_with_composition(
Expand Down
57 changes: 57 additions & 0 deletions tests/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
load_spec_with_composition,
load_spec_with_environment,
)
from netengine.spec.models import NetEngineSpec


@pytest.fixture
Expand Down Expand Up @@ -260,6 +261,34 @@ def test_load_spec(self, base_spec_file: Path) -> None:
spec = SpecConfig.load(base_spec_file)
assert spec["metadata"]["name"] == "test-network"

def test_load_raw_returns_dict(self, base_spec_file: Path) -> None:
"""Test explicit raw spec loading returns a dictionary."""
spec = SpecConfig.load_raw(base_spec_file)

assert isinstance(spec, dict)
assert spec["metadata"]["name"] == "test-network"

def test_load_validated_returns_netengine_spec(self, base_spec_file: Path) -> None:
"""Test validated spec loading returns a NetEngineSpec."""
spec = SpecConfig.load_validated(base_spec_file)

assert isinstance(spec, NetEngineSpec)
assert spec.metadata.name == "test-network"

def test_load_validated_rejects_invalid_specs(self, base_spec_file: Path) -> None:
"""Test validated spec loading rejects invalid composed specs."""
overrides = {
"substrate": {
"networks": {
"platform": {"type": "bridge", "subnet": "10.0.0.0/8"},
"core": {"type": "bridge", "subnet": "10.1.0.0/16"},
}
}
}

with pytest.raises(SpecLoadError, match="subnet overlap"):
SpecConfig.load_validated(base_spec_file, overrides=overrides)

def test_load_with_base(self, base_spec_file: Path, prod_spec_file: Path) -> None:
"""Test loading spec with base composition."""
spec = SpecConfig.load(prod_spec_file, base_path=base_spec_file)
Expand All @@ -281,6 +310,34 @@ def test_load_environment_variants_dev(self, base_spec_file: Path, dev_spec_file
# Dev spec sets environment field
assert spec["metadata"].get("environment") == "dev"

def test_load_environment_variants_raw_returns_dict(
self, base_spec_file: Path, dev_spec_file: Path
) -> None:
"""Test explicit raw environment loading returns a dictionary."""
spec = SpecConfig.load_environment_variants_raw(base_spec_file, environment="dev")

assert isinstance(spec, dict)
assert spec["metadata"].get("environment") == "dev"

def test_load_environment_variant_validated_returns_netengine_spec(
self, base_spec_file: Path, dev_spec_file: Path
) -> None:
"""Test validated environment loading returns a NetEngineSpec."""
spec = SpecConfig.load_environment_variant_validated(base_spec_file, environment="dev")

assert isinstance(spec, NetEngineSpec)
assert spec.metadata.name == "test-network"
assert spec.metadata.environment == "dev"

def test_load_environment_variant_validated_rejects_invalid_specs(
self, base_spec_file: Path
) -> None:
"""Test validated environment loading rejects invalid composed specs."""
overrides = {"substrate": {"networks": {"bad": {"type": "bridge", "subnet": "bad"}}}}

with pytest.raises(SpecLoadError, match="invalid CIDR"):
SpecConfig.load_environment_variant_validated(base_spec_file, overrides=overrides)

def test_load_environment_variants_prod(
self, base_spec_file: Path, prod_spec_file: Path
) -> None:
Expand Down