alexei.dolgolyov
6de61b965e
A new `template` value source evaluates a hardened, sandboxed Jinja expression over the live values of other value sources — the system's first float combinator. Backend: - Shared engine (utils/template_expr.py): ImmutableSandboxedEnvironment with filters/tests and auto-injected globals stripped; only min/max/abs/round/ clamp exposed; rejects **, string/collection-literal repetition, attribute access and non-global calls; NaN/inf-safe result coercion. - TemplateValueSource model + TemplateValueStream runtime: compile-once, primitives-only eval context, raw[name] exposure, eval_interval throttle, ref-counted input acquire/release, rename-safe hot-update. - Validation: unbound-variable + reserved-name rejection, reference cycle/depth guards (depth-only at create, full cycle at update), runtime acquire() depth backstop, and delete referential-integrity. - API: Create/Update/Response schemas + discriminated unions, _RESPONSE_MAP, and an advisory POST /value-sources/validate-template endpoint. - Demo seed: a static source plus a template combinator example. Frontend: - Editor modal section: repeatable inputs list (EntitySelect rows), a zero-dependency Jinja syntax highlighter, a hints/reference panel, and a debounced live validator that gates Save (stale-response-safe). - Graph editor: read-only template node with one edge per input. - i18n (en/ru/zh), icon, and card rendering. Tests: engine, stream, factory/cycle, validate endpoint, and demo seed.
232 lines
8.6 KiB
Python
232 lines
8.6 KiB
Python
"""Tests for the template value source: model, factory, cycle/depth, refs."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from ledgrab.storage.value_source import TemplateValueSource
|
|
|
|
|
|
class TestModelRoundTrip:
|
|
def _make(self, **over):
|
|
now = datetime.now(timezone.utc)
|
|
defaults = dict(
|
|
id="vs_t1",
|
|
name="Combo",
|
|
source_type="template",
|
|
created_at=now,
|
|
updated_at=now,
|
|
template="min(a * 2, 1)",
|
|
inputs=[{"name": "a", "value_source_id": "vs_a"}],
|
|
default_value=0.2,
|
|
eval_interval=1.5,
|
|
)
|
|
defaults.update(over)
|
|
return TemplateValueSource(**defaults)
|
|
|
|
def test_to_from_dict_idempotent(self):
|
|
src = self._make()
|
|
rebuilt = TemplateValueSource.from_dict(src.to_dict())
|
|
assert rebuilt.template == src.template
|
|
assert rebuilt.inputs == src.inputs
|
|
assert rebuilt.default_value == src.default_value
|
|
assert rebuilt.eval_interval == src.eval_interval
|
|
assert rebuilt.to_dict()["return_type"] == "float"
|
|
|
|
def test_old_row_deserializes_with_defaults(self):
|
|
"""A row written before template fields existed must load safely."""
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
src = TemplateValueSource.from_dict(
|
|
{
|
|
"id": "vs_old",
|
|
"name": "Old",
|
|
"source_type": "template",
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
assert src.template == ""
|
|
assert src.inputs == []
|
|
assert src.default_value == 0.0
|
|
assert src.eval_interval is None
|
|
|
|
def test_dirty_scalars_coerce_to_defaults(self):
|
|
"""Non-numeric stored scalars must not drop the whole row on load."""
|
|
src = TemplateValueSource.from_dict(
|
|
{
|
|
"id": "x",
|
|
"name": "n",
|
|
"source_type": "template",
|
|
"template": "a",
|
|
"default_value": "not-a-number",
|
|
"eval_interval": "bad",
|
|
}
|
|
)
|
|
assert src.default_value == 0.0
|
|
assert src.eval_interval is None
|
|
|
|
def test_inputs_normalized_from_dirty_data(self):
|
|
src = TemplateValueSource.from_dict(
|
|
{
|
|
"id": "x",
|
|
"name": "n",
|
|
"source_type": "template",
|
|
"inputs": [{"name": "a", "value_source_id": "vs_a"}, "junk", {"bad": 1}],
|
|
}
|
|
)
|
|
# non-dict entries dropped; dict entries coerced to {name, value_source_id}
|
|
assert src.inputs == [
|
|
{"name": "a", "value_source_id": "vs_a"},
|
|
{"name": "", "value_source_id": ""},
|
|
]
|
|
|
|
|
|
class TestFactoryCreate:
|
|
def test_create_valid(self, value_source_store):
|
|
src = value_source_store.create_source(
|
|
"Combo",
|
|
"template",
|
|
template="min(a * 2, 1)",
|
|
inputs=[{"name": "a", "value_source_id": ""}],
|
|
default_value=0.3,
|
|
)
|
|
assert isinstance(src, TemplateValueSource)
|
|
assert src.id.startswith("vs_")
|
|
assert src.default_value == 0.3
|
|
|
|
def test_empty_template_rejected(self, value_source_store):
|
|
with pytest.raises(ValueError):
|
|
value_source_store.create_source("X", "template", template=" ", inputs=[])
|
|
|
|
def test_compile_error_rejected(self, value_source_store):
|
|
with pytest.raises(ValueError):
|
|
value_source_store.create_source("X", "template", template="a +", inputs=[])
|
|
|
|
def test_cost_bomb_rejected(self, value_source_store):
|
|
with pytest.raises(ValueError):
|
|
value_source_store.create_source("X", "template", template="10 ** 10", inputs=[])
|
|
|
|
def test_reserved_input_name_rejected(self, value_source_store):
|
|
with pytest.raises(ValueError):
|
|
value_source_store.create_source(
|
|
"X",
|
|
"template",
|
|
template="min(0, 1)",
|
|
inputs=[{"name": "min", "value_source_id": "vs_a"}],
|
|
)
|
|
|
|
def test_duplicate_input_name_rejected(self, value_source_store):
|
|
with pytest.raises(ValueError):
|
|
value_source_store.create_source(
|
|
"X",
|
|
"template",
|
|
template="a",
|
|
inputs=[
|
|
{"name": "a", "value_source_id": "vs_a"},
|
|
{"name": "a", "value_source_id": "vs_b"},
|
|
],
|
|
)
|
|
|
|
def test_default_value_out_of_range_rejected(self, value_source_store):
|
|
with pytest.raises(ValueError):
|
|
value_source_store.create_source(
|
|
"X",
|
|
"template",
|
|
template="a",
|
|
inputs=[{"name": "a", "value_source_id": ""}],
|
|
default_value=5.0,
|
|
)
|
|
|
|
def test_unbound_variable_rejected(self, value_source_store):
|
|
# 'ha_enti' is referenced but only 'ha_entity' is bound (typo) → reject.
|
|
with pytest.raises(ValueError):
|
|
value_source_store.create_source(
|
|
"X",
|
|
"template",
|
|
template="ha_enti",
|
|
inputs=[{"name": "ha_entity", "value_source_id": ""}],
|
|
)
|
|
|
|
|
|
class TestFactoryUpdate:
|
|
def test_partial_update_template_only(self, value_source_store):
|
|
src = value_source_store.create_source(
|
|
"X",
|
|
"template",
|
|
template="a",
|
|
inputs=[{"name": "a", "value_source_id": ""}],
|
|
default_value=0.1,
|
|
)
|
|
updated = value_source_store.update_source(src.id, template="clamp(a * 3)")
|
|
assert updated.template == "clamp(a * 3)"
|
|
assert updated.default_value == 0.1 # unchanged
|
|
|
|
def test_update_invalid_template_rejected(self, value_source_store):
|
|
src = value_source_store.create_source("X", "template", template="clamp(0.5)", inputs=[])
|
|
with pytest.raises(ValueError):
|
|
value_source_store.update_source(src.id, template="a |")
|
|
|
|
|
|
class TestCycleAndDepth:
|
|
def test_self_reference_rejected(self, value_source_store):
|
|
t = value_source_store.create_source("T", "template", template="clamp(0.5)", inputs=[])
|
|
with pytest.raises(ValueError):
|
|
value_source_store.update_source(t.id, inputs=[{"name": "x", "value_source_id": t.id}])
|
|
|
|
def test_circular_reference_rejected(self, value_source_store):
|
|
t1 = value_source_store.create_source("T1", "template", template="clamp(0.5)", inputs=[])
|
|
t2 = value_source_store.create_source(
|
|
"T2",
|
|
"template",
|
|
template="x",
|
|
inputs=[{"name": "x", "value_source_id": t1.id}],
|
|
)
|
|
# t1 -> t2 -> t1 would be a cycle
|
|
with pytest.raises(ValueError):
|
|
value_source_store.update_source(
|
|
t1.id, inputs=[{"name": "x", "value_source_id": t2.id}]
|
|
)
|
|
|
|
def test_deep_chain_rejected(self, value_source_store):
|
|
prev = value_source_store.create_source("L0", "template", template="clamp(0.5)", inputs=[])
|
|
created = 1
|
|
with pytest.raises(ValueError):
|
|
for i in range(1, 12):
|
|
node = value_source_store.create_source(
|
|
f"L{i}",
|
|
"template",
|
|
template="x",
|
|
inputs=[{"name": "x", "value_source_id": prev.id}],
|
|
)
|
|
prev = node
|
|
created += 1
|
|
# Should have rejected before building an unbounded chain.
|
|
assert created <= 8
|
|
|
|
def test_get_transitive_dependencies(self, value_source_store):
|
|
leaf = value_source_store.create_source(
|
|
"leaf", "template", template="clamp(0.5)", inputs=[]
|
|
)
|
|
mid = value_source_store.create_source(
|
|
"mid", "template", template="x", inputs=[{"name": "x", "value_source_id": leaf.id}]
|
|
)
|
|
top = value_source_store.create_source(
|
|
"top", "template", template="x", inputs=[{"name": "x", "value_source_id": mid.id}]
|
|
)
|
|
deps = value_source_store.get_transitive_dependencies(top.id)
|
|
assert deps == {mid.id, leaf.id}
|
|
|
|
|
|
class TestReferencingSources:
|
|
def test_find_referencing_sources(self, value_source_store):
|
|
base = value_source_store.create_source("Base", "static", value=0.5)
|
|
tmpl = value_source_store.create_source(
|
|
"Uses",
|
|
"template",
|
|
template="b",
|
|
inputs=[{"name": "b", "value_source_id": base.id}],
|
|
)
|
|
refs = value_source_store.find_referencing_sources(base.id)
|
|
assert tmpl.name in refs
|
|
assert value_source_store.find_referencing_sources(tmpl.id) == []
|