From f82df562e43518be86d5efc06e9b5701cce63e03 Mon Sep 17 00:00:00 2001 From: aidankhogg Date: Mon, 29 Jun 2026 23:12:40 +0100 Subject: [PATCH] Refactor spec loader validation --- netengine/spec/loader.py | 105 ++++++++++++--------------------------- tests/test_config.py | 37 ++++++++++++++ 2 files changed, 70 insertions(+), 72 deletions(-) diff --git a/netengine/spec/loader.py b/netengine/spec/loader.py index 79832e5..630fe5e 100644 --- a/netengine/spec/loader.py +++ b/netengine/spec/loader.py @@ -188,6 +188,36 @@ def _cross_validate(spec: NetEngineSpec) -> None: ) +def _validate_spec_data( + data: dict[str, Any], *, validate_feature_states: bool +) -> NetEngineSpec: + """Validate loaded spec data and return a NetEngineSpec model.""" + if not isinstance(data, dict): + raise SpecLoadError("Spec must be a YAML object (dict)") + + metadata = data.get("metadata") + if not isinstance(metadata, dict): + raise SpecLoadError("Spec metadata must be a YAML object (dict)") + schema_version = metadata.get("schema_version") + if schema_version is not None and schema_version not in SUPPORTED_SPEC_SCHEMA_VERSIONS: + supported = ", ".join(sorted(SUPPORTED_SPEC_SCHEMA_VERSIONS)) + raise SpecLoadError( + f"Unsupported spec metadata.schema_version {schema_version!r}; " + f"supported versions: {supported}. Export with the older NetEngine version " + "or migrate the spec before booting this release." + ) + + try: + spec = NetEngineSpec(**data) + except ValidationError as e: + raise SpecLoadError(f"Spec validation failed: {e}") + + _cross_validate(spec) + if validate_feature_states: + _validate_feature_states(spec) + return spec + + def load_spec(yaml_path: str | Path, *, validate_feature_states: bool = True) -> NetEngineSpec: """Load and validate a NetEngine YAML specification. @@ -213,30 +243,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}") - if not isinstance(data, dict): - raise SpecLoadError("Spec must be a YAML object (dict)") - - metadata = data.get("metadata") - if not isinstance(metadata, dict): - raise SpecLoadError("Spec metadata must be a YAML object (dict)") - schema_version = metadata.get("schema_version") - if schema_version is not None and schema_version not in SUPPORTED_SPEC_SCHEMA_VERSIONS: - supported = ", ".join(sorted(SUPPORTED_SPEC_SCHEMA_VERSIONS)) - raise SpecLoadError( - f"Unsupported spec metadata.schema_version {schema_version!r}; " - f"supported versions: {supported}. Export with the older NetEngine version " - "or migrate the spec before booting this release." - ) - - try: - spec = NetEngineSpec(**data) - except ValidationError as e: - raise SpecLoadError(f"Spec validation failed: {e}") - - _cross_validate(spec) - if validate_feature_states: - _validate_feature_states(spec) - return spec + return _validate_spec_data(data, validate_feature_states=validate_feature_states) def load_spec_with_composition( @@ -287,30 +294,7 @@ def load_spec_with_composition( except IOError as e: raise SpecLoadError(f"Failed to read file: {e}") - if not isinstance(data, dict): - raise SpecLoadError("Spec must be a YAML object (dict)") - - metadata = data.get("metadata") - if not isinstance(metadata, dict): - raise SpecLoadError("Spec metadata must be a YAML object (dict)") - schema_version = metadata.get("schema_version") - if schema_version is not None and schema_version not in SUPPORTED_SPEC_SCHEMA_VERSIONS: - supported = ", ".join(sorted(SUPPORTED_SPEC_SCHEMA_VERSIONS)) - raise SpecLoadError( - f"Unsupported spec metadata.schema_version {schema_version!r}; " - f"supported versions: {supported}. Export with the older NetEngine version " - "or migrate the spec before booting this release." - ) - - try: - spec = NetEngineSpec(**data) - except ValidationError as e: - raise SpecLoadError(f"Spec validation failed: {e}") - - _cross_validate(spec) - if validate_feature_states: - _validate_feature_states(spec) - return spec + return _validate_spec_data(data, validate_feature_states=validate_feature_states) def load_spec_with_environment( @@ -359,27 +343,4 @@ def load_spec_with_environment( except IOError as e: raise SpecLoadError(f"Failed to read file: {e}") - if not isinstance(data, dict): - raise SpecLoadError("Spec must be a YAML object (dict)") - - metadata = data.get("metadata") - if not isinstance(metadata, dict): - raise SpecLoadError("Spec metadata must be a YAML object (dict)") - schema_version = metadata.get("schema_version") - if schema_version is not None and schema_version not in SUPPORTED_SPEC_SCHEMA_VERSIONS: - supported = ", ".join(sorted(SUPPORTED_SPEC_SCHEMA_VERSIONS)) - raise SpecLoadError( - f"Unsupported spec metadata.schema_version {schema_version!r}; " - f"supported versions: {supported}. Export with the older NetEngine version " - "or migrate the spec before booting this release." - ) - - try: - spec = NetEngineSpec(**data) - except ValidationError as e: - raise SpecLoadError(f"Spec validation failed: {e}") - - _cross_validate(spec) - if validate_feature_states: - _validate_feature_states(spec) - return spec + return _validate_spec_data(data, validate_feature_states=validate_feature_states) diff --git a/tests/test_config.py b/tests/test_config.py index fdf5aa0..9f453af 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -260,6 +260,12 @@ def test_load_environment_with_overrides( class TestSpecLoaderIntegration: """Integration tests for spec loader with OmegaConf.""" + def _write_spec(self, tmp_dir: Path, name: str, spec: dict) -> Path: + path = tmp_dir / name + with open(path, "w") as f: + yaml.dump(spec, f) + return path + def test_load_spec_backward_compatibility(self, base_spec_file: Path) -> None: """Test that original load_spec function still works.""" spec = load_spec(base_spec_file) @@ -294,6 +300,37 @@ def test_spec_invalid_yaml(self, temp_spec_dir: Path) -> None: with pytest.raises(SpecLoadError, match="Failed to parse YAML"): load_spec(spec_file) + def test_load_spec_rejects_unsupported_schema_version(self, temp_spec_dir: Path) -> None: + """Test schema-version rejection in the direct loader path.""" + data = _minimal_spec() + data["metadata"]["schema_version"] = "netengine.spec.future" + spec_file = self._write_spec(temp_spec_dir, "spec.yaml", data) + + with pytest.raises(SpecLoadError, match="Unsupported spec metadata.schema_version"): + load_spec(spec_file) + + def test_load_spec_with_composition_rejects_unsupported_schema_version( + self, temp_spec_dir: Path + ) -> None: + """Test schema-version rejection in the composition loader path.""" + base_file = self._write_spec(temp_spec_dir, "spec.base.yaml", _minimal_spec()) + override = {"metadata": {"schema_version": "netengine.spec.future"}} + override_file = self._write_spec(temp_spec_dir, "spec.prod.yaml", override) + + with pytest.raises(SpecLoadError, match="Unsupported spec metadata.schema_version"): + load_spec_with_composition(override_file, base_path=base_file) + + def test_load_spec_with_environment_rejects_unsupported_schema_version( + self, temp_spec_dir: Path + ) -> None: + """Test schema-version rejection in the environment loader path.""" + base_file = self._write_spec(temp_spec_dir, "spec.base.yaml", _minimal_spec()) + env_override = {"metadata": {"schema_version": "netengine.spec.future"}} + self._write_spec(temp_spec_dir, "spec.dev.yaml", env_override) + + with pytest.raises(SpecLoadError, match="Unsupported spec metadata.schema_version"): + load_spec_with_environment(base_file, environment="dev") + class TestSpecCrossValidation: """Tests for cross-field spec validation (CIDR overlap, name uniqueness, etc.)."""