fix(resources): resolve AmbiguousStep, wire registry context into _validate_model, add depth limit
- Fix AmbiguousStep: rename step decorator from
'the creation should fail with "{fragment1}" or "{fragment2}"' to
'the creation should fail with either "{fragment1}" or "{fragment2}"'
so behave parse does not treat it as ambiguous with the single-arg form;
this was causing all 8 features to error on step load, failing CI.
- Wire registry into _validate_model(): add ValidationInfo parameter and
extract type_registry from Pydantic validation context so multi-level
cycle detection (A->B->A, A->B->C->A) runs through the production code
path, not just as a pre-creation standalone call.
- Update BDD steps to use ResourceTypeSpec.model_validate(..., context=
{"type_registry": registry}) instead of calling detect_inheritance_cycles
directly before construction, so tests validate the actual fix path.
- Add MAX_INHERITANCE_DEPTH = 100 constant and depth counter in
detect_inheritance_cycles() while loop to guard against DoS via
pathologically deep chains.
- Consolidate five separate import blocks from _resource_type_validation
into a single grouped import in resource_type.py.
- Add depth-limit scenario and step covering the new MAX_INHERITANCE_DEPTH
guard to ensure new lines are covered by diff-coverage.
This commit is contained in:
@@ -17,7 +17,7 @@ Feature: ResourceTypeSpec inheritance cycle detection for multi-level cycles
|
||||
Scenario: Two-level cycle A→B→A is detected
|
||||
Given a registered type "acme/alpha" inheriting from "acme/beta"
|
||||
When I create a ResourceTypeSpec with name "acme/beta" inheriting "acme/alpha"
|
||||
Then the creation should fail with "circular" or "cycle"
|
||||
Then the creation should fail with either "circular" or "cycle"
|
||||
|
||||
# ── Three-level cycles (new behavior) ────────────────────────────────────
|
||||
|
||||
@@ -25,7 +25,7 @@ Feature: ResourceTypeSpec inheritance cycle detection for multi-level cycles
|
||||
Given a registered type "acme/alpha" inheriting from "acme/beta"
|
||||
And a registered type "acme/beta" inheriting from "acme/gamma"
|
||||
When I create a ResourceTypeSpec with name "acme/gamma" inheriting "acme/alpha"
|
||||
Then the creation should fail with "circular" or "cycle"
|
||||
Then the creation should fail with either "circular" or "cycle"
|
||||
|
||||
# ── Valid multi-level inheritance (should succeed) ──────────────────────
|
||||
|
||||
@@ -39,3 +39,9 @@ Feature: ResourceTypeSpec inheritance cycle detection for multi-level cycles
|
||||
Given a registered type "acme/alpha" inheriting from "acme/beta"
|
||||
When I create a ResourceTypeSpec with name "acme/beta" inheriting nothing
|
||||
Then the creation should succeed
|
||||
|
||||
# ── Depth limit (DoS guard) ───────────────────────────────────────────────
|
||||
|
||||
Scenario: Inheritance chain exceeding the depth limit is rejected
|
||||
When I create a ResourceTypeSpec whose ancestry chain exceeds the depth limit
|
||||
Then the creation should fail with "exceeds"
|
||||
|
||||
@@ -11,12 +11,10 @@ from typing import Any
|
||||
from behave import given, then, when
|
||||
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
detect_inheritance_cycles,
|
||||
MAX_INHERITANCE_DEPTH,
|
||||
)
|
||||
from cleveragents.domain.models.core.resource_type import (
|
||||
ResourceKind,
|
||||
ResourceTypeSpec,
|
||||
SandboxStrategy,
|
||||
)
|
||||
|
||||
|
||||
@@ -43,21 +41,21 @@ def step_register_type(context: Any, name: str, parent: str) -> None:
|
||||
def step_create_with_inherits(context: Any, name: str, parent: str) -> None:
|
||||
"""Attempt to create a ResourceTypeSpec with the given inheritance.
|
||||
|
||||
Also performs multi-level cycle detection using the test registry
|
||||
(if populated) since ResourceTypeSpec._validate_model only has access
|
||||
to the model's own fields, not the broader type registry.
|
||||
Passes the test registry as Pydantic validation context so that
|
||||
ResourceTypeSpec._validate_model() performs multi-level cycle detection
|
||||
through the production code path.
|
||||
"""
|
||||
registry: dict[str, Any] = getattr(context, "type_registry", {})
|
||||
ctx = {"type_registry": registry} if registry else None
|
||||
try:
|
||||
# Perform multi-level cycle detection using the registry first.
|
||||
# This mirrors what a registry-aware service would do before
|
||||
# persisting a new type definition.
|
||||
detect_inheritance_cycles(name, parent, registry if registry else None)
|
||||
spec = ResourceTypeSpec(
|
||||
name=name,
|
||||
resource_kind=ResourceKind.PHYSICAL,
|
||||
sandbox_strategy=SandboxStrategy.NONE,
|
||||
inherits=parent,
|
||||
spec = ResourceTypeSpec.model_validate(
|
||||
{
|
||||
"name": name,
|
||||
"resource_kind": "physical",
|
||||
"sandbox_strategy": "none",
|
||||
"inherits": parent,
|
||||
},
|
||||
context=ctx,
|
||||
)
|
||||
context.created_spec = spec # type: ignore[attr-defined]
|
||||
context.creation_error = None # type: ignore[attr-defined]
|
||||
@@ -69,11 +67,13 @@ def step_create_with_inherits(context: Any, name: str, parent: str) -> None:
|
||||
def step_create_without_inherits(context: Any, name: str) -> None:
|
||||
"""Attempt to create a ResourceTypeSpec without inheritance."""
|
||||
try:
|
||||
spec = ResourceTypeSpec(
|
||||
name=name,
|
||||
resource_kind=ResourceKind.PHYSICAL,
|
||||
sandbox_strategy=SandboxStrategy.NONE,
|
||||
inherits=None,
|
||||
spec = ResourceTypeSpec.model_validate(
|
||||
{
|
||||
"name": name,
|
||||
"resource_kind": "physical",
|
||||
"sandbox_strategy": "none",
|
||||
"inherits": None,
|
||||
}
|
||||
)
|
||||
context.created_spec = spec # type: ignore[attr-defined]
|
||||
context.creation_error = None # type: ignore[attr-defined]
|
||||
@@ -91,7 +91,7 @@ def step_creation_failed(context: Any, fragment: str) -> None:
|
||||
)
|
||||
|
||||
|
||||
@then('the creation should fail with "{fragment1}" or "{fragment2}"')
|
||||
@then('the creation should fail with either "{fragment1}" or "{fragment2}"')
|
||||
def step_creation_failed_either(context: Any, fragment1: str, fragment2: str) -> None:
|
||||
"""Assert that creation failed with an error containing either fragment."""
|
||||
err = context.creation_error # type: ignore[attr-defined]
|
||||
@@ -108,3 +108,26 @@ def step_creation_succeeded(context: Any) -> None:
|
||||
err = context.creation_error # type: ignore[attr-defined]
|
||||
assert err is None, f"Expected creation to succeed but got error: {err}"
|
||||
assert hasattr(context, "created_spec"), "Expected created_spec to be set"
|
||||
|
||||
|
||||
@when("I create a ResourceTypeSpec whose ancestry chain exceeds the depth limit")
|
||||
def step_create_exceeds_depth(context: Any) -> None:
|
||||
"""Build a registry chain longer than MAX_INHERITANCE_DEPTH and attempt creation."""
|
||||
chain_len = MAX_INHERITANCE_DEPTH + 1
|
||||
registry: dict[str, Any] = {
|
||||
f"chain/t{i}": {"inherits": f"chain/t{i + 1}"} for i in range(chain_len)
|
||||
}
|
||||
try:
|
||||
spec = ResourceTypeSpec.model_validate(
|
||||
{
|
||||
"name": "chain/new",
|
||||
"resource_kind": "physical",
|
||||
"sandbox_strategy": "none",
|
||||
"inherits": "chain/t0",
|
||||
},
|
||||
context={"type_registry": registry},
|
||||
)
|
||||
context.created_spec = spec # type: ignore[attr-defined]
|
||||
context.creation_error = None # type: ignore[attr-defined]
|
||||
except (ValueError, TypeError) as exc:
|
||||
context.creation_error = exc # type: ignore[attr-defined]
|
||||
|
||||
@@ -14,6 +14,7 @@ _BUILTIN_NAME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")
|
||||
_NAMESPACED_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*/[a-zA-Z][a-zA-Z0-9_-]*$")
|
||||
|
||||
MAX_SCAN_DEPTH = 10
|
||||
MAX_INHERITANCE_DEPTH = 100
|
||||
|
||||
# All known built-in type names. Extracted here so that
|
||||
# ``ResourceTypeSpec.BUILTIN_NAMES`` stays a one-liner reference.
|
||||
@@ -303,8 +304,15 @@ def detect_inheritance_cycles(
|
||||
if registry is not None and inherits is not None:
|
||||
visited: set[str] = {name}
|
||||
current = inherits
|
||||
depth = 0
|
||||
|
||||
while current is not None:
|
||||
depth += 1
|
||||
if depth > MAX_INHERITANCE_DEPTH:
|
||||
raise ValueError(
|
||||
f"Inheritance chain for '{name}' exceeds the maximum "
|
||||
f"allowed depth of {MAX_INHERITANCE_DEPTH}."
|
||||
)
|
||||
if current in visited:
|
||||
# Cycle detected
|
||||
raise ValueError(
|
||||
|
||||
@@ -22,21 +22,20 @@ import re
|
||||
from enum import StrEnum
|
||||
from typing import Any, ClassVar
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
||||
from pydantic import (
|
||||
BaseModel,
|
||||
ConfigDict,
|
||||
Field,
|
||||
ValidationInfo,
|
||||
field_validator,
|
||||
model_validator,
|
||||
)
|
||||
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
BUILTIN_TYPE_NAMES,
|
||||
)
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
detect_inheritance_cycles as _detect_inheritance_cycles,
|
||||
)
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
validate_auto_discovery as _validate_auto_discovery,
|
||||
)
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
validate_self_referential as _validate_self_referential,
|
||||
)
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
validate_virtual_type as _validate_virtual_type,
|
||||
)
|
||||
|
||||
@@ -311,7 +310,7 @@ class ResourceTypeSpec(BaseModel):
|
||||
# -- Cross-field validation -----------------------------------------------
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_model(self) -> ResourceTypeSpec:
|
||||
def _validate_model(self, info: ValidationInfo) -> ResourceTypeSpec:
|
||||
"""Cross-field validation for resource type constraints."""
|
||||
# Custom types must be namespaced
|
||||
if not self.built_in and "/" not in self.name:
|
||||
@@ -342,7 +341,8 @@ class ResourceTypeSpec(BaseModel):
|
||||
)
|
||||
|
||||
# ADR-042 rule 3: no cycles (self-inheritance and multi-level cycles)
|
||||
_detect_inheritance_cycles(self.name, self.inherits)
|
||||
_registry = info.context.get("type_registry") if info.context else None
|
||||
_detect_inheritance_cycles(self.name, self.inherits, _registry)
|
||||
# ADR-042 rule 4: built-in must not inherit from custom
|
||||
if self.built_in and self.inherits is not None and "/" in self.inherits:
|
||||
raise ValueError(
|
||||
|
||||
Reference in New Issue
Block a user