app/backend/tests/flow/test_node_settings.py

"""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 "")


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):
    store.write_draft(flow, 0)
    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 "")