stroblmeandClaude Opus 5
60d7ec81c0
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. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
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 "")
|