feat(engine): fixedvalueshandler + route legacy dispatch through it

This commit is contained in:
DumiFlex
2026-04-26 16:30:49 +03:00
parent 6626aca3d7
commit 2eeb90052e
4 changed files with 126 additions and 11 deletions
+20 -11
View File
@@ -1,18 +1,20 @@
"""Per-module-type handlers. One function per module type.
Handler signature is ``(module, ctx, rng) -> Context``. Returns the (possibly
mutated) context. ``rng`` is a ``random.Random`` supplied by the pipeline —
``fixed_values`` ignores it; seed-consuming handlers (wildcard, derivation)
use it in later specs.
mutated) context. As of this commit, the actual resolution lives in
``engine.modules.dispatcher``; the function below is now a thin bridge that
coerces the legacy dataclass shape, calls ``resolve_module``, and writes the
result back into ``ctx``.
"""
import dataclasses
import logging
import random
from collections.abc import Callable
from typing import Any, TypeAlias
from .context import Context
from .modules import FixedValueModule
from .modules import FixedValueModule, coerce_legacy_module, resolve_module
logger = logging.getLogger(__name__)
@@ -24,17 +26,24 @@ def handle_fixed_values(
ctx: Context,
_rng: random.Random,
) -> Context:
"""Write each entry's value into ``ctx``. Mutates and returns ``ctx``.
"""Write each entry's value into ``ctx`` via the new dispatcher bridge.
``_rng`` is accepted for signature parity with future seed-consuming
handlers (wildcard, derivation); ``fixed_values`` doesn't use randomness.
``_rng`` is accepted for signature parity with seed-consuming handlers.
"""
for entry in module.entries:
name = entry.variable_name.lstrip("$")
raw = (
dataclasses.asdict(module)
if dataclasses.is_dataclass(module)
else dict(module)
)
snapshot = coerce_legacy_module(raw)
bindings = resolve_module(snapshot, ctx=None)
for var_name, value in bindings.items():
name = var_name.lstrip("$")
if not name:
logger.warning(
"Skipping entry with empty variable_name in module %s", module.id
"Skipping binding with empty variable_name in module %s",
module.id,
)
continue
ctx[name] = entry.value
ctx[name] = value
return ctx
+6
View File
@@ -49,3 +49,9 @@ __all__ += [
"RecursionLimitExceeded",
"WildcardHandler",
]
from engine.modules.fixed_values_handler import FixedValuesHandler # noqa: E402
register_handler(FixedValuesHandler)
__all__ += ["FixedValuesHandler"]
+30
View File
@@ -0,0 +1,30 @@
"""Fixed-values module resolver. Emits one binding per named value."""
from __future__ import annotations
from typing import Any
from engine.modules.dispatcher import ModuleHandler
class FixedValuesHandler(ModuleHandler):
type_id = "fixed_values"
@classmethod
def resolve(
cls,
payload: dict[str, Any],
instance: dict[str, Any],
ctx: Any,
) -> dict[str, str]:
values: list[dict[str, Any]] = list(payload.get("values", []))
enabled = instance.get("enabled_options")
if enabled is not None:
allowed = set(enabled)
values = [v for v in values if v.get("id") in allowed]
out: dict[str, str] = {}
for v in values:
name = (v.get("name") or "").strip()
if not name:
continue
out[name] = str(v.get("value", ""))
return out
@@ -0,0 +1,70 @@
"""Tests for FixedValuesHandler — emits one mapping per named value."""
from engine.modules.fixed_values_handler import FixedValuesHandler
def _payload(values):
return {"values": values}
def test_resolve_returns_all_values():
payload = _payload([
{"id": "v1", "name": "$lens", "value": "85mm"},
{"id": "v2", "name": "$angle", "value": "wide"},
])
out = FixedValuesHandler.resolve(payload, instance={}, ctx=None)
assert out == {"$lens": "85mm", "$angle": "wide"}
def test_resolve_empty_values_returns_empty_dict():
out = FixedValuesHandler.resolve(_payload([]), instance={}, ctx=None)
assert out == {}
def test_resolve_filters_by_enabled_options():
payload = _payload([
{"id": "v1", "name": "$a", "value": "1"},
{"id": "v2", "name": "$b", "value": "2"},
])
out = FixedValuesHandler.resolve(
payload, instance={"enabled_options": ["v2"]}, ctx=None,
)
assert out == {"$b": "2"}
def test_resolve_skips_blank_names():
payload = _payload([
{"id": "v1", "name": "", "value": "x"},
{"id": "v2", "name": " ", "value": "y"},
{"id": "v3", "name": "$ok", "value": "z"},
])
out = FixedValuesHandler.resolve(payload, instance={}, ctx=None)
assert out == {"$ok": "z"}
def test_resolve_handles_missing_value_key():
"""A value without 'value' resolves to empty string."""
payload = _payload([{"id": "v1", "name": "$x"}])
out = FixedValuesHandler.resolve(payload, instance={}, ctx=None)
assert out == {"$x": ""}
def test_resolve_coerces_value_to_string():
payload = _payload([{"id": "v1", "name": "$n", "value": 42}])
out = FixedValuesHandler.resolve(payload, instance={}, ctx=None)
assert out == {"$n": "42"}
def test_handler_type_id_is_fixed_values():
assert FixedValuesHandler.type_id == "fixed_values"
def test_resolve_via_dispatcher_after_import():
"""Importing engine.modules auto-registers FixedValuesHandler."""
from engine.modules import resolve_module
snap = {
"type": "fixed_values",
"payload": _payload([{"id": "v1", "name": "$lens", "value": "85mm"}]),
"instance": {},
}
out = resolve_module(snap, ctx=None)
assert out == {"$lens": "85mm"}