From 61611f5de6b0c2a3b59425c64811bafc7290cb8a Mon Sep 17 00:00:00 2001 From: Brad Edwards Date: Thu, 16 Jul 2026 04:12:41 +0200 Subject: [PATCH] refactor: split scenario instantiation helpers --- .../aces_sdl/_scenario_instantiation.py | 38 ++++++++++++++++++ .../python/packages/aces_sdl/scenario.py | 39 ++----------------- 2 files changed, 42 insertions(+), 35 deletions(-) create mode 100644 implementations/python/packages/aces_sdl/_scenario_instantiation.py diff --git a/implementations/python/packages/aces_sdl/_scenario_instantiation.py b/implementations/python/packages/aces_sdl/_scenario_instantiation.py new file mode 100644 index 000000000..a84b8391b --- /dev/null +++ b/implementations/python/packages/aces_sdl/_scenario_instantiation.py @@ -0,0 +1,38 @@ +"""Private helpers for validating instantiated scenarios.""" + +from collections.abc import Mapping + +from ._base import VARIABLE_TOKEN_RE + + +def collect_variable_tokens(value: object) -> list[str]: + """Return the names of every ``${name}`` token found in string values. + + Mirrors ``instantiate._substitute_value``: every string is a substitution + site and mapping keys are not. An instantiated scenario is fully concrete, + so no token may survive in any string value. + """ + found: list[str] = [] + if isinstance(value, Mapping): + for nested in value.values(): + found.extend(collect_variable_tokens(nested)) + elif isinstance(value, (list, tuple)): + for item in value: + found.extend(collect_variable_tokens(item)) + elif isinstance(value, str): + found.extend(VARIABLE_TOKEN_RE.findall(value)) + return found + + +def resolve_json_pointer(payload: object, pointer: str) -> object: + """Resolve an RFC 6901-style JSON pointer against a scenario payload.""" + current = payload + for raw_segment in pointer.split("/")[1:]: + segment = raw_segment.replace("~1", "/").replace("~0", "~") + if isinstance(current, Mapping): + current = current[segment] + elif isinstance(current, (list, tuple)): + current = current[int(segment)] + else: + raise TypeError("JSON Pointer traverses a scalar value") + return current diff --git a/implementations/python/packages/aces_sdl/scenario.py b/implementations/python/packages/aces_sdl/scenario.py index 77f9f8ac9..b5e5ec754 100644 --- a/implementations/python/packages/aces_sdl/scenario.py +++ b/implementations/python/packages/aces_sdl/scenario.py @@ -16,7 +16,7 @@ from pydantic import ConfigDict, Field, PrivateAttr, model_validator -from ._base import VARIABLE_TOKEN_RE, SDLModel +from ._base import SDLModel from ._errors import SDLParseDiagnostic from ._identifiers import ( PortableIdentifier, @@ -25,6 +25,7 @@ require_portable_identifier, ) from ._mapping_scopes import HASHMAP_SECTIONS +from ._scenario_instantiation import collect_variable_tokens, resolve_json_pointer from ._stateful_resource_references import validate_stateful_resource_references from .accounts import Account from .agents import Agent @@ -100,25 +101,6 @@ def _constraint_node_refs( return refs -def _collect_variable_tokens(value: object) -> list[str]: - """Return the names of every ``${name}`` token found in string *values*. - - Mirrors ``instantiate._substitute_value``: every string is a substitution - site and mapping keys are not. An :class:`InstantiatedScenario` is fully - concrete, so no token may survive in any string value. - """ - found: list[str] = [] - if isinstance(value, Mapping): - for nested in value.values(): - found.extend(_collect_variable_tokens(nested)) - elif isinstance(value, (list, tuple)): - for item in value: - found.extend(_collect_variable_tokens(item)) - elif isinstance(value, str): - found.extend(VARIABLE_TOKEN_RE.findall(value)) - return found - - def _validate_declaration_identifier( identifier: object, *, @@ -465,7 +447,7 @@ def explicitness(self) -> dict[str, ExplicitnessRecord]: @model_validator(mode="after") def _reject_unresolved_variable_references(self) -> "InstantiatedScenario": payload = self.model_dump(mode="json", by_alias=True) - tokens = sorted(set(_collect_variable_tokens(payload))) + tokens = sorted(set(collect_variable_tokens(payload))) if tokens: joined = ", ".join(tokens) raise ValueError(f"InstantiatedScenario must not contain unresolved variable references: {joined}") @@ -475,7 +457,7 @@ def _reject_unresolved_variable_references(self) -> "InstantiatedScenario": binding_values[(*imported.namespace, *binding.parameter)] = binding.value for constraint in self.instantiation_provenance.capability_constraints: try: - concrete_value = _resolve_json_pointer(payload, constraint.field_pointer) + concrete_value = resolve_json_pointer(payload, constraint.field_pointer) except (KeyError, IndexError, TypeError, ValueError) as exc: raise ValueError( f"Capability constraint field_pointer does not resolve: {constraint.field_pointer}" @@ -485,16 +467,3 @@ def _reject_unresolved_variable_references(self) -> "InstantiatedScenario": if not _json_value_equal(concrete_value, binding_values[constraint.parameter]): raise ValueError("Capability constraint binding does not match the concrete field value") return self - - -def _resolve_json_pointer(payload: object, pointer: str) -> object: - current = payload - for raw_segment in pointer.split("/")[1:]: - segment = raw_segment.replace("~1", "/").replace("~0", "~") - if isinstance(current, Mapping): - current = current[segment] - elif isinstance(current, (list, tuple)): - current = current[int(segment)] - else: - raise TypeError("JSON Pointer traverses a scalar value") - return current