stroblme
99534a82ee
A wheel whose top-level module is `app` collides with anything else in a user's venv, so the package that is about to be published takes the name it is published under. Only the Python package moves; the repo, the Docker WORKDIR and the compose project keep theirs.
64 lines
2.0 KiB
Python
64 lines
2.0 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
|
|
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
|
|
|
|
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):
|
|
store.write_draft(a_flow(reading=3), 0)
|
|
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 "")
|