app/backend/tests/flow/test_dashboards.py

"""Dashboards: documents beside the flows, and the values their widgets move."""

import pytest

from fluksio.flow.dashboards import (
    DashboardDef,
    DashboardNotFound,
    DashboardStore,
    SettingDef,
    WidgetDef,
    default_dashboard,
    results_dashboard,
)
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.nodes import Node
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, StaleVersion


@pytest.fixture
def store(tmp_path) -> DashboardStore:
    return DashboardStore(FlowStore(tmp_path / "flows"))


def chart(message: str, points: int) -> WidgetDef:
    return WidgetDef(
        id="temp",
        type="chart",
        config={"series": [{"message": message}], "history": {"points": points}},
    )


def test_a_dashboard_survives_a_round_trip(store: DashboardStore):
    saved = store.write(default_dashboard("house"))

    read = store.read("house")

    assert read.name == "house"
    assert read.widgets == []
    assert read.version == saved.version


def test_dashboards_are_invisible_to_the_flow_listing(store: DashboardStore):
    """They share the repository; they are not flows."""
    store.write(default_dashboard("house"))

    assert store.flows.list_flows() == []
    assert [d.name for d in store.list()] == ["house"]


def test_a_save_based_on_a_version_someone_moved_past_is_refused(
    store: DashboardStore,
):
    first = store.write(default_dashboard("house"))
    store.write(first, first.version)

    with pytest.raises(StaleVersion):
        store.write(first, first.version)


def test_an_edit_reaches_a_panel_only_once_it_is_published(store: DashboardStore):
    """A wall panel reads the published file; the editor writes beside it."""
    published = store.write(default_dashboard("house"))

    draft = store.write_draft(
        published.model_copy(update={"title": "Kitchen"}), published.version
    )

    assert store.has_draft("house")
    assert store.read("house").title == published.title
    assert store.read("house", draft=True).title == "Kitchen"

    store.publish("house", draft.version)

    assert store.read("house").title == "Kitchen"
    assert not store.has_draft("house")


def test_discarding_leaves_what_is_published(store: DashboardStore):
    published = store.write(default_dashboard("house"))
    store.write_draft(
        published.model_copy(update={"title": "Kitchen"}), published.version
    )

    assert store.discard_draft("house").title == published.title
    assert not store.has_draft("house")


def test_a_new_dashboard_is_a_draft_until_it_is_published(store: DashboardStore):
    """Creating one does not put it on a wall; publishing is what does."""
    created = store.write_draft(default_dashboard("house"), 0)

    assert not store.is_published("house")
    with pytest.raises(DashboardNotFound):
        store.read("house")
    assert store.read("house", draft=True).title == created.title
    # Listed all the same, so the editor can find what it just made.
    assert [(d.name, d.has_draft, d.version) for d in store.list()] == [
        ("house", True, created.version)
    ]

    store.publish("house", created.version)

    assert store.is_published("house")
    assert store.read("house").title == created.title


def test_an_unpublished_dashboard_can_be_renamed_and_deleted(store: DashboardStore):
    """And renaming it does not put it on a wall either."""
    store.write_draft(default_dashboard("house"), 0)

    store.rename("house", "home")

    assert not store.exists("house")
    assert store.has_draft("home") and not store.is_published("home")

    store.delete("home")
    assert not store.exists("home")


def test_deleting_and_renaming(store: DashboardStore):
    store.write(default_dashboard("house"))

    renamed = store.rename("house", "home")
    assert renamed.name == "home"
    assert not store.exists("house")

    store.delete("home")
    with pytest.raises(DashboardNotFound):
        store.read("home")


def test_the_deepest_chart_decides_how_much_past_is_kept(store: DashboardStore):
    store.write(
        DashboardDef(
            name="house",
            widgets=[
                chart("heating.temp", 400),
                chart("heating.temp", 900),
                chart("solar.watts", 100),
            ],
        )
    )

    assert store.history_requirements() == {"heating.temp": 900, "solar.watts": 100}


def test_a_chart_cannot_ask_for_an_unbounded_series(store: DashboardStore):
    store.write(
        DashboardDef(
            name="house",
            widgets=[chart("heating.temp", 10**9)],
        )
    )

    assert store.history_requirements() == {"heating.temp": 5000}


def test_history_is_kept_to_the_depth_a_chart_asked_for():
    state = MemoryState()
    limits = {"f.temp": 300}

    for i in range(400):
        state.append_history({"f.temp": float(i)}, float(i), limits)

    assert len(state.history("f.temp")) == 300


# ---------------------------------------------------------------------------
# What an input widget does
# ---------------------------------------------------------------------------


def test_publishing_a_value_runs_what_consumes_it():
    """A slider is a value arriving; the graph should not care who sent it."""
    seen: list[float] = []

    def consume(setpoint, params):
        seen.append(setpoint)
        return {"applied": setpoint}

    node = Node(
        f=consume,
        requires=[MessageSpec(name="setpoint", port="setpoint", dtype=DType.FLOAT)],
        provides=[MessageSpec(name="applied", port="applied", dtype=DType.FLOAT)],
        name="thermostat",
    )
    node.assign_flow("heating", "thermostat")
    state = MemoryState()
    pipeline = Pipeline(nodes=[node], state=state)

    pipeline.publish({"heating.setpoint": 21.5})

    assert seen == [21.5]
    assert state["heating.applied"] == 21.5


def test_publishing_nothing_does_nothing():
    pipeline = Pipeline(nodes=[], state=MemoryState())

    pipeline.publish({})

    assert pipeline.values() == {}


def query_chart(**config) -> dict:
    return {
        "source": "query",
        "request": "heating.chart_req",
        "request_dtype": "record",
        "message": "heating.chart_series",
        "dtype": "series",
        "refresh_s": 30,
        **config,
    }


def test_a_widget_refuses_a_dtype_it_cannot_carry():
    WidgetDef(id="s", type="switch", config={"message": "a.b", "dtype": "bool"})
    # A document written before the editor recorded types binds anything.
    WidgetDef(id="s", type="switch", config={"message": "a.b"})

    with pytest.raises(ValueError):
        WidgetDef(id="s", type="switch", config={"message": "a.b", "dtype": "float"})


def test_the_structured_widgets_bind_their_shapes():
    WidgetDef(id="a", type="agenda", config={"message": "a.b", "dtype": "list"})
    WidgetDef(id="n", type="notification", config={"message": "a.b", "dtype": "record"})
    WidgetDef(id="f", type="forecast", config={"message": "a.b", "dtype": "list"})

    with pytest.raises(ValueError):
        WidgetDef(id="a", type="agenda", config={"message": "a.b", "dtype": "json"})
    with pytest.raises(ValueError):
        WidgetDef(
            id="n", type="notification", config={"message": "a.b", "dtype": "list"}
        )
    with pytest.raises(ValueError):
        WidgetDef(id="f", type="forecast", config={"message": "a.b", "dtype": "json"})


def test_a_media_widget_binds_media_and_nothing_else():
    for dtype in ("image", "audio", "video", "artifact"):
        WidgetDef(id="m", type="media", config={"message": "cam.frame", "dtype": dtype})

    for dtype in ("float", "record"):
        with pytest.raises(ValueError):
            WidgetDef(
                id="m", type="media", config={"message": "cam.frame", "dtype": dtype}
            )


def test_a_bar_nests_a_second_number():
    WidgetDef(
        id="b",
        type="bar",
        config={
            "message": "a.in",
            "dtype": "float",
            "inner": "a.pv",
            "inner_dtype": "int",
        },
    )

    with pytest.raises(ValueError):
        WidgetDef(
            id="b",
            type="bar",
            config={"message": "a.in", "dtype": "float", "inner_dtype": "bool"},
        )


def test_a_bar_is_drawn_on_both_readings_it_nests():
    widget = WidgetDef(id="b", type="bar", config={"message": "a.in", "inner": "a.pv"})

    assert widget.messages == ["a.in", "a.pv"]


def test_a_bar_reads_its_rows():
    widget = WidgetDef(
        id="b",
        type="bar",
        config={
            "rows": [
                {"message": "a.load", "dtype": "float", "label": "House"},
                {"message": "a.pv", "dtype": "int"},
            ],
            "min": 0,
            "max": 9,
        },
    )

    assert widget.messages == ["a.load", "a.pv"]

    with pytest.raises(ValueError):
        WidgetDef(
            id="b",
            type="bar",
            config={"rows": [{"message": "a.on", "dtype": "bool"}]},
        )

    with pytest.raises(ValueError):
        WidgetDef(
            id="b",
            type="bar",
            config={"rows": [{"message": f"a.m{n}"} for n in range(9)]},
        )


def test_a_clock_reads_nothing_and_publishes_nothing():
    widget = WidgetDef(id="c", type="clock", config={"format": "24h"})

    assert widget.messages == []
    assert widget.target == ""


def test_a_querying_chart_asks_with_a_record_and_draws_a_series():
    widget = WidgetDef(id="c", type="chart", config=query_chart())

    # It publishes its request and reads the answer, so the canvas draws both.
    assert widget.target == "heating.chart_req"
    assert widget.messages == ["heating.chart_series"]

    with pytest.raises(ValueError):
        WidgetDef(id="c", type="chart", config=query_chart(dtype="float"))
    with pytest.raises(ValueError):
        WidgetDef(id="c", type="chart", config=query_chart(request_dtype="json"))


def test_a_colour_widget_publishes_the_shape_its_format_names():
    """The format picks the payload, so the binding is held to that one."""
    wheel = WidgetDef(
        id="lamp",
        type="color",
        config={"target": "a.color", "dtype": "list", "format": "hsv"},
    )

    assert wheel.target == "a.color"
    # It publishes rather than reads: nothing is drawn from a message.
    assert wheel.messages == []
    # Hex is the same widget speaking a string, and rgb is still a list.
    WidgetDef(
        id="l", type="color", config={"target": "a.c", "dtype": "str", "format": "hex"}
    )
    WidgetDef(
        id="l", type="color", config={"target": "a.c", "dtype": "list", "format": "rgb"}
    )
    # Nothing recorded binds anything, as everywhere else.
    WidgetDef(id="l", type="color", config={"target": "a.c"})

    with pytest.raises(ValueError):
        WidgetDef(
            id="l",
            type="color",
            config={"target": "a.c", "dtype": "str", "format": "hsv"},
        )
    with pytest.raises(ValueError):
        WidgetDef(
            id="l",
            type="color",
            config={"target": "a.c", "dtype": "list", "format": "hex"},
        )
    with pytest.raises(ValueError):
        WidgetDef(id="l", type="color", config={"target": "a.c", "dtype": "float"})


def test_a_querying_chart_keeps_no_ring():
    """The answer carries its own past; a ring would store it twice."""
    widget = WidgetDef(
        id="c", type="chart", config=query_chart(history={"points": 900})
    )

    assert widget.history_points == 0


def test_a_dashboard_is_cut_into_a_sane_number_of_columns():
    for columns in (0, 49):
        with pytest.raises(ValueError):
            DashboardDef(name="house", columns=columns)


def test_a_setting_refuses_a_message_it_cannot_carry():
    """The same rule a widget binding is held to, from the document alone."""
    DashboardDef(
        name="house",
        settings={"theme": SettingDef(value="dark", message="a.b", dtype="str")},
    )
    DashboardDef(
        name="house",
        settings={"locked": SettingDef(value=False, message="a.b", dtype="bool")},
    )

    with pytest.raises(ValueError):
        DashboardDef(
            name="house",
            settings={"theme": SettingDef(value="dark", message="a.b", dtype="float")},
        )
    with pytest.raises(ValueError):
        DashboardDef(
            name="house",
            settings={"locked": SettingDef(value=False, message="a.b", dtype="str")},
        )


def test_an_unbound_setting_is_just_its_value():
    """The static case, which is what makes a dark panel cost no flow.

    No message means nothing to type-check and nothing for a panel to be
    entitled to — and a name this build does not know is left alone rather
    than refused, so an older installation reads a newer document.
    """
    defn = DashboardDef(
        name="house",
        settings={
            "theme": SettingDef(value="dark"),
            "someday": SettingDef(value=7, message="a.b", dtype="int"),
        },
    )

    assert defn.settings["theme"].value == "dark"
    assert defn.setting_messages == ["a.b"]


def test_a_bound_setting_is_drawn_on_the_canvas(store: DashboardStore):
    """A dashboard consuming a message is an endpoint like a tile is."""
    store.write(
        DashboardDef(
            name="house",
            settings={"theme": SettingDef(value="system", message="home.theme")},
        )
    )

    (binding,) = store.bindings_for("home")
    assert binding["widget"] == "settings.theme"
    assert binding["type"] == "setting"
    assert binding["requires"] == ["home.theme"]
    assert not binding["provides"]
    assert store.bindings_for("other") == []


def training_flow() -> FlowDef:
    """A batch flow shaped like an experiment: a curve and two results."""
    return FlowDef(
        name="study",
        mode="batch",
        outputs=["accuracy", "report"],
        nodes=[
            NodeDef(
                id="fit",
                provides=[
                    MessageSpec(name="loss", dtype=DType.FLOAT, stream=True),
                    MessageSpec(name="accuracy", dtype=DType.FLOAT),
                    MessageSpec(name="report", dtype=DType.RECORD),
                    MessageSpec(name="weights", dtype=DType.ARTIFACT),
                ],
            )
        ],
    )


def test_a_generated_dashboard_charts_the_curves_and_states_the_results():
    """The ports are the whole specification; nothing else is guessed."""
    defn = results_dashboard(training_flow())
    kinds = [(w.type, w.config.get("message")) for w in defn.widgets]

    assert [w.type for w in defn.widgets] == ["chart", "stat", "notification"]
    assert defn.widgets[0].config["series"][0]["message"] == "study.loss"
    assert ("stat", "study.accuracy") in kinds
    assert ("notification", "study.report") in kinds


def test_an_artifact_output_gets_no_widget():
    """A checkpoint has no single reading to draw."""
    defn = results_dashboard(
        training_flow().model_copy(update={"outputs": ["weights"]})
    )

    assert [w.type for w in defn.widgets] == ["chart"]


def test_a_flow_declaring_no_outputs_draws_no_stats():
    """Empty outputs means "everything", which is not a set to enumerate."""
    defn = results_dashboard(training_flow().model_copy(update={"outputs": []}))

    assert [w.type for w in defn.widgets] == ["chart"]


def runs_chart(**runs) -> WidgetDef:
    return WidgetDef(id="curves", type="chart", config={"source": "runs", "runs": runs})


def test_a_chart_pinned_to_runs_reads_no_live_message():
    """Its series is in the run tables; the engine has nothing to keep for it."""
    widget = runs_chart(metric="study.loss", flow="study", latest=3)

    assert widget.messages == []
    assert widget.history_points == 0


def test_a_chart_of_runs_must_say_which_runs_and_which_metric():
    with pytest.raises(ValueError, match="metric"):
        runs_chart(flow="study")
    with pytest.raises(ValueError, match="a flow, a sweep, or run ids"):
        runs_chart(metric="study.loss")
    with pytest.raises(ValueError, match="between 1 and 5"):
        runs_chart(metric="study.loss", flow="study", latest=9)


def test_a_document_written_as_pages_is_read_as_one_grid():
    """Stored dashboards live in each installation's repository.

    So the old shape is normalised on the way in rather than migrated, and a
    placed second section keeps its arrangement instead of piling onto the
    first — which is how the viewer always drew it.
    """
    old = {
        "name": "house",
        "pages": [
            {
                "id": "main",
                "sections": [
                    {
                        "id": "a",
                        "widgets": [
                            {
                                "id": "top",
                                "type": "stat",
                                "layout": {"lg": {"x": 0, "y": 0, "w": 3, "h": 2}},
                            }
                        ],
                    },
                    {
                        "id": "b",
                        "widgets": [
                            {
                                "id": "under",
                                "type": "stat",
                                "layout": {"lg": {"x": 0, "y": 1, "w": 3, "h": 2}},
                            }
                        ],
                    },
                ],
            }
        ],
    }

    read = DashboardDef.model_validate(old)

    assert [w.id for w in read.widgets] == ["top", "under"]
    # The first section is two rows deep, so the second one starts under it.
    assert read.widgets[1].layout["lg"].y == 3
    assert "pages" not in read.model_dump()