240 lines
7.8 KiB
Python
240 lines
7.8 KiB
Python
"""A node's settings are arguments of its function, like its ports.
|
|
|
|
What distinguishes them is where the value comes from: a port carries whatever
|
|
the graph last published, a setting is a constant stored with the flow. Both
|
|
arrive by name, so one name cannot mean both.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fluksio.flow.controller import FlowController, _signature_gap, follow_header
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.pipeline import Pipeline
|
|
from fluksio.flow.schemas import FlowDef, NodeDef
|
|
from fluksio.flow.state import MemoryState
|
|
from fluksio.flow.store import FlowStore
|
|
from tests.utils.stores import created
|
|
|
|
SOURCE = "def process(reading, factor):\n return {'scaled': reading * factor}\n"
|
|
|
|
|
|
def a_flow(**params: object) -> FlowDef:
|
|
return FlowDef(
|
|
name="house",
|
|
nodes=[
|
|
NodeDef(
|
|
id="scale",
|
|
params=dict(params),
|
|
requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
|
|
provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)],
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path: Path) -> FlowStore:
|
|
return FlowStore(tmp_path / "flows")
|
|
|
|
|
|
def test_a_setting_reaches_the_function_as_a_keyword_argument(store: FlowStore):
|
|
store.write_flow(a_flow(factor=3))
|
|
store.write_node_source("house", "scale", SOURCE)
|
|
controller = FlowController(store)
|
|
|
|
nodes, _loaded, _initial, _inputs = controller._build_flows(
|
|
[(store.read_flow("house"), False)]
|
|
)
|
|
pipeline = Pipeline(nodes=nodes, state=MemoryState())
|
|
pipeline.run({"house.reading": 2.0})
|
|
|
|
assert pipeline.values()["house.scaled"]["value"] == 6.0
|
|
|
|
|
|
def test_a_setting_named_after_a_port_is_refused(store: FlowStore):
|
|
created(store, a_flow(reading=3))
|
|
store.write_node_source("house", "scale", SOURCE, draft=True)
|
|
controller = FlowController(store)
|
|
|
|
preview = controller.preview("house")
|
|
|
|
assert [node.status for node in preview.nodes] == ["error"]
|
|
assert "both an input and a setting" in (preview.nodes[0].error or "")
|
|
|
|
|
|
def _flow(requires: list[MessageSpec], **params: object) -> FlowDef:
|
|
return FlowDef(
|
|
name="house",
|
|
nodes=[
|
|
NodeDef(
|
|
id="scale",
|
|
params=dict(params),
|
|
requires=requires,
|
|
provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)],
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
READING = [MessageSpec(name="reading", dtype=DType.FLOAT)]
|
|
|
|
|
|
def _preview(store: FlowStore, flow: FlowDef, source: str | None):
|
|
created(store, flow)
|
|
if source is not None:
|
|
store.write_node_source("house", "scale", source, draft=True)
|
|
return FlowController(store).preview("house").nodes[0]
|
|
|
|
|
|
def test_a_port_the_function_cannot_take_is_refused_at_load(store: FlowStore):
|
|
node = _preview(store, _flow(READING), "def process(value):\n return {}\n")
|
|
|
|
assert node.status == "error"
|
|
assert "'reading' is an input of 'scale'" in (node.error or "")
|
|
assert "process(value)" in (node.error or "")
|
|
|
|
|
|
def test_a_setting_the_function_cannot_take_is_refused_at_load(store: FlowStore):
|
|
node = _preview(store, _flow(READING, gain=2), SOURCE)
|
|
|
|
assert node.status == "error"
|
|
assert "'gain' is a setting of 'scale'" in (node.error or "")
|
|
|
|
|
|
def test_a_function_taking_keywords_accepts_anything(store: FlowStore):
|
|
node = _preview(
|
|
store, _flow(READING, gain=2), "def process(**kw):\n return {}\n"
|
|
)
|
|
|
|
assert node.status == "active"
|
|
|
|
|
|
def test_a_node_never_written_is_not_wrong_yet(store: FlowStore):
|
|
# The template a new node opens with takes nothing; that is missing, not a
|
|
# mismatch.
|
|
node = _preview(store, _flow(READING), None)
|
|
|
|
assert node.status == "active"
|
|
|
|
|
|
def test_a_port_arrives_under_its_own_name(store: FlowStore):
|
|
other = [MessageSpec(name="other.reading", dtype=DType.FLOAT)]
|
|
node = _preview(store, _flow(other, factor=3), SOURCE)
|
|
|
|
assert node.status == "active"
|
|
|
|
|
|
def test_a_shared_body_is_checked_against_each_user(store: FlowStore):
|
|
store.write_flow(_flow(READING, factor=3))
|
|
store.write_node_source("house", "scale", SOURCE)
|
|
store.share_node("house", "scale", "scale_it")
|
|
store.write_flow(
|
|
FlowDef(
|
|
name="garden",
|
|
nodes=[
|
|
NodeDef(
|
|
id="scale",
|
|
source_ref="scale_it",
|
|
requires=[MessageSpec(name="level", dtype=DType.FLOAT)],
|
|
provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)],
|
|
)
|
|
],
|
|
)
|
|
)
|
|
controller = FlowController(store)
|
|
|
|
nodes = controller.preview("garden").nodes
|
|
|
|
assert [n.status for n in nodes] == ["error"]
|
|
assert "'level' is an input of 'scale'" in (nodes[0].error or "")
|
|
|
|
|
|
def _scaffold_header(node: NodeDef) -> str:
|
|
"""The header the canvas writes: input ports in order, then settings."""
|
|
names = [spec.port for spec in node.requires] + [
|
|
key for key in node.params if key != "synchronous"
|
|
]
|
|
return f"def process({', '.join(names)}):\n return {{}}\n"
|
|
|
|
|
|
def test_a_scaffold_header_reads_back_as_the_node_it_came_from():
|
|
node = _flow(
|
|
[*READING, MessageSpec(name="garden.level", port="depth")],
|
|
factor=3,
|
|
synchronous=True,
|
|
).nodes[0]
|
|
|
|
assert follow_header(node, _scaffold_header(node)) == node
|
|
|
|
|
|
def test_a_header_declares_ports_and_settings_that_read_back_the_same():
|
|
header = "def process(reading, level, gain=1.5, mode='eco'):\n return {}\n"
|
|
|
|
node = follow_header(_flow([]).nodes[0], header)
|
|
|
|
assert [spec.port for spec in node.requires] == ["reading", "level"]
|
|
assert node.params == {"gain": 1.5, "mode": "eco"}
|
|
assert _scaffold_header(node).split(":")[0] == (
|
|
"def process(reading, level, gain, mode)"
|
|
)
|
|
|
|
|
|
def test_a_renamed_parameter_keeps_the_message_and_a_dropped_one_goes():
|
|
node = _flow(READING, factor=3, offset=1).nodes[0]
|
|
|
|
renamed = follow_header(node, "def process(value, factor, offset): pass\n")
|
|
dropped = follow_header(node, "def process(reading, factor): pass\n")
|
|
|
|
assert renamed.requires == [READING[0].model_copy(update={"port": "value"})]
|
|
assert renamed.params == {"factor": 3, "offset": 1}
|
|
assert dropped.requires == READING
|
|
assert dropped.params == {"factor": 3}
|
|
|
|
|
|
def test_without_drop_a_missing_parameter_stays_and_the_rest_follows():
|
|
node = _flow(
|
|
[*READING, MessageSpec(name="garden.level", port="depth")], factor=3
|
|
).nodes[0]
|
|
|
|
paused = follow_header(node, "def process(depth): pass\n", drop=False)
|
|
renamed = follow_header(node, "def process(value, depth, factor): pass\n", False)
|
|
added = follow_header(node, "def process(depth, gain=2): pass\n", drop=False)
|
|
|
|
assert paused == node
|
|
assert renamed.requires[0] == READING[0].model_copy(update={"port": "value"})
|
|
assert [spec.port for spec in added.requires] == ["reading", "depth"]
|
|
assert added.params == {"factor": 3, "gain": 2}
|
|
|
|
|
|
def test_a_new_parameter_after_a_setting_is_a_setting():
|
|
node = _flow(READING, factor=3).nodes[0]
|
|
|
|
followed = follow_header(node, "def process(reading, factor, offset): pass\n")
|
|
|
|
assert followed.requires == READING
|
|
assert followed.params == {"factor": 3, "offset": ""}
|
|
|
|
|
|
def test_a_header_it_cannot_read_for_certain_leaves_the_node():
|
|
node = _flow(READING, factor=3).nodes[0]
|
|
|
|
for code in (
|
|
"def process(reading, factor",
|
|
"def process(**kw): pass\n",
|
|
"@cache\ndef process(value): pass\n",
|
|
"def run(value): pass\n",
|
|
):
|
|
assert follow_header(node, code) == node
|
|
|
|
|
|
def test_code_written_for_the_params_dict_is_told_settings_arrive_by_name():
|
|
gap = _signature_gap("def process(a, params): pass\n", "n", {"a"}, {"target"})
|
|
|
|
assert gap is not None
|
|
assert "process(..., target)" in gap
|
|
assert (
|
|
_signature_gap("def process(a, target): pass\n", "n", {"a"}, {"target"}) is None
|
|
)
|