app/backend/tests/api/routes/test_devices.py

"""Devices: assignment, pairing, and what a paired credential may reach."""

import threading

from fastapi.testclient import TestClient

from fluksio.api.routes import devices as devices_route
from fluksio.cloud import config as cloud_config
from fluksio.core.config import settings
from fluksio.flow.devices import DeviceState
from fluksio.main import app

PREFIX = f"{settings.API_V1_STR}/devices"
#: What the openpanel firmware calls, kept as aliases.
PANELS = f"{settings.API_V1_STR}/panels"
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"


def _panels(client: TestClient, headers: dict[str, str], config: dict) -> None:
    """Make the stored devices exactly ``config["panels"]``, one call each."""
    wanted = {device["id"]: device for device in config["panels"]}
    stored = client.get(f"{PREFIX}/", headers=headers).json()["data"]
    for device in stored:
        if device["id"] not in wanted:
            gone = client.delete(f"{PREFIX}/{device['id']}", headers=headers)
            assert gone.status_code == 200, gone.text
    versions = {device["id"]: device["version"] for device in stored}
    for device_id, device in wanted.items():
        if device_id not in versions:
            created = client.post(f"{PREFIX}/{device_id}", headers=headers)
            assert created.status_code == 200, created.text
            versions[device_id] = created.json()["version"]
        saved = client.put(
            f"{PREFIX}/{device_id}",
            headers=headers,
            json={**device, "version": versions[device_id]},
        )
        assert saved.status_code == 200, saved.text


def _dashboard(client: TestClient, headers: dict[str, str], name: str) -> dict:
    """Create a dashboard and publish it.

    A new one is a draft, and a panel is only ever shown what is published.
    """
    created = client.post(f"{DASHBOARDS}/{name}", headers=headers)
    assert created.status_code == 200, created.text
    published = client.post(
        f"{DASHBOARDS}/{name}/publish",
        headers=headers,
        json={"version": created.json()["version"]},
    )
    assert published.status_code == 200, published.text
    return published.json()


def _pair(client: TestClient, headers: dict[str, str], panel: str) -> dict[str, str]:
    """Walk a device through pairing and return the header it ends up with."""
    started = client.post(f"{PREFIX}/pair").json()
    waiting = client.get(
        f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
    )
    assert waiting.json()["access_token"] is None

    approved = client.post(
        f"{PREFIX}/{panel}/pair", headers=headers, json={"code": started["code"]}
    )
    assert approved.status_code == 200, approved.text

    collected = client.get(
        f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
    ).json()
    assert collected["device"] == collected["panel"] == panel
    return {"Authorization": f"Bearer {collected['access_token']}"}


def test_devices_require_authentication(client: TestClient) -> None:
    assert client.get(f"{PREFIX}/").status_code == 401
    assert client.put(f"{PREFIX}/hall", json={"id": "hall"}).status_code == 401
    assert client.get(f"{PREFIX}/pending").status_code == 401


def test_assign_and_read_back(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    for name in ("hall_a", "hall_b"):
        client.post(f"{DASHBOARDS}/{name}", headers=superuser_token_headers)

    _panels(
        client,
        superuser_token_headers,
        {
            "panels": [
                {"id": "hall", "title": "Hall", "dashboards": ["hall_a", "hall_b"]}
            ]
        },
    )

    stored = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()
    hall = next(device for device in stored["data"] if device["id"] == "hall")
    assert hall["dashboards"] == ["hall_a", "hall_b"]
    assert hall["online"] is False
    # The address a device is pointed at comes from the server, because the
    # browser's own origin is the portal's when someone administers remotely.
    assert stored["frontend_host"] == settings.FRONTEND_HOST.rstrip("/")
    assert (
        client.get(f"{PREFIX}/hall", headers=superuser_token_headers).json()["title"]
        == "Hall"
    )
    assert (
        client.get(f"{PREFIX}/nowhere", headers=superuser_token_headers).status_code
        == 404
    )


def test_touch_belongs_to_the_device(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Whether a screen is touched is a fact about the screen, not the document.

    The same dashboard may hang on a hallway tablet and on a desk browser, so
    the flag rides on the panel and each one answers for itself.
    """
    client.post(f"{DASHBOARDS}/shared", headers=superuser_token_headers)
    _panels(
        client,
        superuser_token_headers,
        {
            "panels": [
                {"id": "wall", "dashboards": ["shared"], "touch": True},
                {"id": "desk", "dashboards": ["shared"]},
            ]
        },
    )

    by_id = {
        device["id"]: device
        for device in client.get(f"{PREFIX}/", headers=superuser_token_headers).json()[
            "data"
        ]
    }
    assert by_id["wall"]["touch"] is True
    # Absent is pointed at.
    assert by_id["desk"]["touch"] is False


def test_creating_twice_or_a_reserved_id_is_refused(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    assert (
        client.post(f"{PREFIX}/twice", headers=superuser_token_headers).status_code
        == 200
    )
    assert (
        client.post(f"{PREFIX}/twice", headers=superuser_token_headers).status_code
        == 409
    )
    assert (
        client.post(f"{PREFIX}/pending", headers=superuser_token_headers).status_code
        == 422
    )


def test_concurrent_saves_of_one_version_give_one_409(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """The race `PUT /panels/` had: one of two saves of the same version wins."""
    created = client.post(
        f"{PREFIX}/racy", headers=superuser_token_headers, json={"title": "Racy"}
    ).json()
    statuses: list[int] = []

    def save(title: str) -> None:
        statuses.append(
            client.put(
                f"{PREFIX}/racy",
                headers=superuser_token_headers,
                json={**created, "title": title},
            ).status_code
        )

    threads = [threading.Thread(target=save, args=(t,)) for t in ("A", "B")]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert sorted(statuses) == [200, 409]

    stored = client.get(f"{PREFIX}/racy", headers=superuser_token_headers).json()
    assert stored["version"] == created["version"] + 1
    assert stored["title"] in ("A", "B")


def test_polling_needs_the_secret(client: TestClient) -> None:
    started = client.post(f"{PREFIX}/pair").json()
    assert len(started["code"]) == 6
    assert (
        client.get(
            f"{PREFIX}/pair/{started['code']}", params={"secret": "wrong"}
        ).status_code
        == 404
    )
    assert (
        client.get(f"{PREFIX}/pair/ZZZZZZ", params={"secret": "x"}).status_code == 404
    )


def test_approving_an_unknown_code_or_panel_is_refused(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    _panels(client, superuser_token_headers, {"panels": [{"id": "hall"}]})
    assert (
        client.post(
            f"{PREFIX}/hall/pair",
            headers=superuser_token_headers,
            json={"code": "ZZZZZZ"},
        ).status_code
        == 404
    )
    started = client.post(f"{PREFIX}/pair").json()
    assert (
        client.post(
            f"{PREFIX}/nowhere/pair",
            headers=superuser_token_headers,
            json={"code": started["code"]},
        ).status_code
        == 404
    )


def test_paired_panel_reaches_only_what_it_shows(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    for name in ("panel_shown", "panel_hidden"):
        _dashboard(client, superuser_token_headers, name)
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "hall", "dashboards": ["panel_shown"]}]},
    )

    panel_headers = _pair(client, superuser_token_headers, "hall")

    # What it was assigned, published, plus its own definition and the messages
    # its widgets speak.
    assert (
        client.get(f"{DASHBOARDS}/panel_shown", headers=panel_headers).status_code
        == 200
    )
    assert client.get(f"{PREFIX}/hall", headers=panel_headers).status_code == 200
    # Where the firmware from before devices reads itself.
    assert client.get(f"{PANELS}/hall", headers=panel_headers).status_code == 200
    # Not another device.
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "hall", "dashboards": ["panel_shown"]}, {"id": "other"}]},
    )
    assert client.get(f"{PREFIX}/other", headers=panel_headers).status_code == 403
    # But not the catalogue, which is the whole namespace at once and no
    # screen's business.
    assert (
        client.get(
            f"{settings.API_V1_STR}/messages/", headers=panel_headers
        ).status_code
        == 403
    )

    # The generated client spells the default out, so `?draft=false` is what a
    # browser actually asks with for "the published one".
    for spelling in ("false", "0", "off"):
        assert (
            client.get(
                f"{DASHBOARDS}/panel_shown",
                headers=panel_headers,
                params={"draft": spelling},
            ).status_code
            == 200
        ), spelling

    # And nothing else.
    assert (
        client.get(f"{DASHBOARDS}/panel_hidden", headers=panel_headers).status_code
        == 403
    )
    # Fail-closed: a spelling neither side recognises counts as a draft.
    for spelling in ("true", "1", "yes", "maybe"):
        assert (
            client.get(
                f"{DASHBOARDS}/panel_shown",
                headers=panel_headers,
                params={"draft": spelling},
            ).status_code
            == 403
        ), spelling
    assert client.get(f"{DASHBOARDS}/", headers=panel_headers).status_code == 403
    assert (
        client.get(f"{settings.API_V1_STR}/flows/", headers=panel_headers).status_code
        == 403
    )
    assert (
        client.delete(f"{DASHBOARDS}/panel_shown", headers=panel_headers).status_code
        == 403
    )
    assert client.get(f"{PREFIX}/", headers=panel_headers).status_code == 403


#: One of each input widget, a querying chart and a plain reading — the set a
#: seeded example draws on. Between them they name every message a panel is
#: entitled to speak, and nothing names ``demo.unrelated``.
PANEL_WIDGETS = [
    {"id": "w_button", "type": "button", "config": {"target": "demo.button"}},
    {
        "id": "w_switch",
        "type": "switch",
        "config": {
            "target": "demo.switch",
            "message": "demo.switch_state",
            "dtype": "bool",
        },
    },
    {
        "id": "w_slider",
        "type": "slider",
        "config": {
            "target": "demo.slider",
            "dtype": "float",
            "reading": {"message": "demo.slider_reading", "dtype": "float"},
        },
    },
    {"id": "w_input", "type": "input", "config": {"target": "demo.input"}},
    {"id": "w_dropdown", "type": "dropdown", "config": {"target": "demo.dropdown"}},
    {
        "id": "w_buttons",
        "type": "buttons",
        "config": {
            "target": "demo.buttons",
            "buttons": [{"label": "One", "value": 1}],
        },
    },
    {
        "id": "w_query",
        "type": "chart",
        "config": {
            "source": "query",
            "request": "demo.query_request",
            "request_dtype": "record",
            "message": "demo.query_series",
            "dtype": "series",
        },
    },
    {"id": "w_stat", "type": "stat", "config": {"message": "demo.temperature"}},
]


def _dashboard_with(
    client: TestClient, headers: dict[str, str], name: str, widgets: list[dict]
) -> None:
    """A published dashboard carrying these widgets."""
    saved = _dashboard(client, headers, name)
    saved["widgets"] = widgets
    written = client.put(f"{DASHBOARDS}/{name}", headers=headers, json=saved)
    assert written.status_code == 200, written.text
    published = client.post(
        f"{DASHBOARDS}/{name}/publish",
        headers=headers,
        json={"version": written.json()["version"]},
    )
    assert published.status_code == 200, published.text


def test_a_panel_speaks_only_its_own_widgets_messages(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """The allowlist is a walk of the panel's widgets, not the whole namespace.

    404 rather than 200 for the ones it may publish: the gate lets them
    through and the engine then refuses them because no flow in this suite
    declares them. What matters here is that it is not 403.
    """
    _dashboard_with(client, superuser_token_headers, "panel_controls", PANEL_WIDGETS)
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "workshop", "dashboards": ["panel_controls"]}]},
    )
    panel_headers = _pair(client, superuser_token_headers, "workshop")
    messages = f"{settings.API_V1_STR}/messages"

    # Every input widget publishes, and so does a querying chart — its request
    # is a value it puts into the graph, and a wall panel with no way to send
    # it would draw nothing.
    for target in (
        "demo.button",
        "demo.switch",
        "demo.slider",
        "demo.input",
        "demo.dropdown",
        "demo.buttons",
        "demo.query_request",
    ):
        assert (
            client.post(
                f"{messages}/{target}", headers=panel_headers, json={"value": 1}
            ).status_code
            == 404
        ), target

    # And reads the history of everything its widgets bind to, but writes none
    # of what they only draw.
    for name in (
        "demo.temperature",
        "demo.switch_state",
        "demo.slider_reading",
        "demo.query_series",
    ):
        assert (
            client.get(f"{messages}/{name}/history", headers=panel_headers).status_code
            == 200
        ), name
        assert (
            client.post(
                f"{messages}/{name}", headers=panel_headers, json={"value": 1}
            ).status_code
            == 403
        ), name

    # A message no tile on this panel names is not its business, in either
    # direction.
    assert (
        client.post(
            f"{messages}/demo.unrelated", headers=panel_headers, json={"value": 1}
        ).status_code
        == 403
    )
    assert (
        client.get(
            f"{messages}/demo.unrelated/history", headers=panel_headers
        ).status_code
        == 403
    )


def _publish_settings(
    client: TestClient, headers: dict[str, str], name: str, settings_: dict
) -> None:
    """Give a published dashboard these settings and publish it again."""
    saved = client.get(f"{DASHBOARDS}/{name}?draft=true", headers=headers).json()
    saved["settings"] = settings_
    written = client.put(f"{DASHBOARDS}/{name}", headers=headers, json=saved)
    assert written.status_code == 200, written.text
    published = client.post(
        f"{DASHBOARDS}/{name}/publish",
        headers=headers,
        json={"version": written.json()["version"]},
    )
    assert published.status_code == 200, published.text


def test_a_locked_dashboard_entitles_a_panel_to_read_and_not_to_publish(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """`locked` bounds what a screen may send, not what it may draw.

    It used to bound neither: the client stopped offering the control and the
    server took the publish from anything that asked anyway.

    Unions like the allowlist itself, which is what the second dashboard here
    is for — a message a locked dashboard shows and an unlocked one controls
    stays writable, because the unlocked one is what entitles the panel to it.
    """
    _dashboard_with(client, superuser_token_headers, "panel_locked", PANEL_WIDGETS)
    _publish_settings(
        client, superuser_token_headers, "panel_locked", {"locked": {"value": True}}
    )
    _dashboard_with(
        client,
        superuser_token_headers,
        "panel_open",
        [{"id": "w_button", "type": "button", "config": {"target": "demo.button"}}],
    )
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "foyer", "dashboards": ["panel_locked", "panel_open"]}]},
    )
    panel_headers = _pair(client, superuser_token_headers, "foyer")
    messages = f"{settings.API_V1_STR}/messages"

    # Only the locked dashboard names it, so there is nothing left to send it.
    assert (
        client.post(
            f"{messages}/demo.slider", headers=panel_headers, json={"value": 1}
        ).status_code
        == 403
    )
    # 404, not 403: past the gate, and refused by an engine that knows no such
    # message. The unlocked dashboard binds this one too.
    assert (
        client.post(
            f"{messages}/demo.button", headers=panel_headers, json={"value": 1}
        ).status_code
        == 404
    )
    # A querying chart's request survives the lock: publishing it is how that
    # tile *reads*, and a locked dashboard whose charts cannot ask goes blank
    # rather than read-only.
    assert (
        client.post(
            f"{messages}/demo.query_request",
            headers=panel_headers,
            json={"value": {"range_s": 3600, "interval_s": 60}},
        ).status_code
        == 404
    )
    # And what the locked dashboard draws is still readable — read-only is not
    # blind, and a wall panel showing stale numbers is the failure to avoid.
    assert (
        client.get(
            f"{messages}/demo.temperature/history", headers=panel_headers
        ).status_code
        == 200
    )


FLOWS = f"{settings.API_V1_STR}/flows"


def _device_node(device: str, node: str = "porch", **ports: list) -> dict:
    """A device node: ``provides`` its ``in`` ports, ``requires`` its ``out`` ones."""
    return {
        "id": node,
        "type": "device",
        "params": {"device": device, "config": ports.pop("config", {})},
        "provides": ports.get("provides", []),
        "requires": ports.get("requires", []),
    }


def _publish_flow(
    client: TestClient, headers: dict[str, str], name: str, nodes: list, **more
) -> dict:
    """Write a flow and publish it, replacing whatever was there."""
    client.delete(f"{FLOWS}/{name}", headers=headers)
    saved = client.put(
        f"{FLOWS}/{name}",
        headers=headers,
        json={"name": name, "nodes": nodes, "version": 0, **more},
    )
    assert saved.status_code == 200, saved.text
    published = client.post(
        f"{FLOWS}/{name}/publish",
        headers=headers,
        json={"version": saved.json()["definition"]["version"]},
    )
    assert published.status_code == 200, published.text
    return published.json()


#: A doorbell button and the backlight, as a device node in flow ``porch``.
PORCH_NODE = _device_node(
    "porch_io",
    provides=[{"name": "doorbell", "port": "gpio.4", "dtype": "bool"}],
    requires=[
        {"name": "backlight", "port": "backlight", "dtype": "int", "trigger": False}
    ],
    config={"gpio.4": {"pull": "up", "debounce_ms": 50}},
)


def test_a_devices_io_comes_from_its_device_nodes(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """What the device reads as ``io`` is what published flows wire, in port order.

    The same bytes for as long as nothing changes: the firmware compares the
    whole document and rebuilds its screen on any difference.
    """
    _panels(client, superuser_token_headers, {"panels": [{"id": "porch_io"}]})
    _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE])
    # A draft wires nothing until it is published.
    client.put(
        f"{FLOWS}/porch_draft",
        headers=superuser_token_headers,
        json={
            "name": "porch_draft",
            "nodes": [
                _device_node(
                    "porch_io",
                    provides=[{"name": "mic", "port": "mic", "dtype": "audio"}],
                )
            ],
            "version": 0,
        },
    )

    first = client.get(f"{PREFIX}/porch_io", headers=superuser_token_headers)
    read = first.json()
    assert read["io"] == [
        {
            "port": "backlight",
            "message": "porch.backlight",
            "direction": "out",
            "config": {},
        },
        {
            "port": "gpio.4",
            "message": "porch.doorbell",
            "direction": "in",
            "config": {"pull": "up", "debounce_ms": 50},
        },
    ]
    assert [(b["flow"], b["node"], b["port"]) for b in read["bindings"]] == [
        ("porch", "porch", "backlight"),
        ("porch", "porch", "gpio.4"),
    ]
    again = client.get(f"{PANELS}/porch_io", headers=superuser_token_headers)
    assert again.content == first.content
    # The list carries the same, for the Devices tab.
    listed = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()
    assert next(d for d in listed["data"] if d["id"] == "porch_io")["io"] == read["io"]
    for flow in ("porch", "porch_draft"):
        client.delete(f"{FLOWS}/{flow}", headers=superuser_token_headers)


def test_a_devices_own_ports_publish_through_a_lock(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """An ``in`` port is the device's own sensor, which a dashboard lock is not about.

    An ``out`` port is the device following a message, so it reads and does
    not write, and only an ``in`` port's name opens the artifact upload. A
    value published on a bound port is the device node's, whatever the device
    says it is from.
    """
    _dashboard_with(client, superuser_token_headers, "panel_io_locked", PANEL_WIDGETS)
    _publish_settings(
        client, superuser_token_headers, "panel_io_locked", {"locked": {"value": True}}
    )
    _panels(
        client,
        superuser_token_headers,
        {
            "panels": [
                {"id": "porch_io", "title": "Porch", "dashboards": ["panel_io_locked"]}
            ]
        },
    )
    _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE])
    panel_headers = _pair(client, superuser_token_headers, "porch_io")
    messages = f"{settings.API_V1_STR}/messages"

    controller = app.state.flow_controller
    sources = []
    original = controller.publish_message

    def recorded(name, value, source=None):  # type: ignore[no-untyped-def]
        sources.append(source)
        return original(name, value, source)

    controller.publish_message = recorded
    try:
        published = client.post(
            f"{messages}/porch.doorbell",
            headers=panel_headers,
            json={"value": True, "source_kind": "panel", "source_id": "panel:x:y"},
        )
    finally:
        del controller.publish_message
    assert published.status_code == 200, published.text
    assert sources[0].model_dump() == {
        "kind": "node",
        "id": "porch.porch",
        "label": "Porch",
        "detail": "gpio.4",
    }

    assert (
        client.post(
            f"{messages}/porch.backlight", headers=panel_headers, json={"value": 50}
        ).status_code
        == 403
    )
    assert (
        client.get(
            f"{messages}/porch.backlight/history", headers=panel_headers
        ).status_code
        == 200
    )
    assert (
        client.post(
            f"{messages}/porch.unrelated", headers=panel_headers, json={"value": 1}
        ).status_code
        == 403
    )

    artifacts = f"{settings.API_V1_STR}/artifacts"
    for name, status in (
        ("porch.doorbell", 200),
        ("porch.backlight", 403),
        ("porch.unrelated", 403),
    ):
        uploaded = client.put(
            artifacts,
            headers=panel_headers,
            params={"name": name, "media_type": "audio/wav", "volatile": True},
            content=b"RIFF",
        )
        assert uploaded.status_code == status, name

    # Unwired, the port is nothing the credential reaches any more.
    _publish_flow(client, superuser_token_headers, "porch", [])
    assert (
        client.post(
            f"{messages}/porch.doorbell", headers=panel_headers, json={"value": True}
        ).status_code
        == 403
    )
    client.delete(f"{FLOWS}/porch", headers=superuser_token_headers)


def test_a_port_wired_twice_or_an_unknown_device_is_an_issue(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    _panels(client, superuser_token_headers, {"panels": [{"id": "porch_io"}]})
    _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE])
    _publish_flow(
        client,
        superuser_token_headers,
        "porch_too",
        [
            _device_node(
                "porch_io",
                provides=[{"name": "bell", "port": "gpio.4", "dtype": "bool"}],
            ),
            _device_node("ghost", node="ghost"),
        ],
    )

    def codes(flow: str) -> set[tuple[str, str | None, bool]]:
        issues = client.get(f"{FLOWS}/{flow}", headers=superuser_token_headers).json()[
            "issues"
        ]
        return {
            (i["code"], i["port"], i["advisory"])
            for i in issues
            if i["code"] in ("device_port", "unknown_device")
        }

    assert codes("porch") == {("device_port", "gpio.4", False)}
    assert codes("porch_too") == {
        ("device_port", "gpio.4", False),
        ("unknown_device", None, False),
    }
    # Creating the device is enough, nothing has to be published again.
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "porch_io"}, {"id": "ghost"}]},
    )
    assert codes("porch_too") == {("device_port", "gpio.4", False)}

    # With a report a port it does not list is a warning, and one it lists the
    # other way round an error. Seen in a draft, which is validated as it is.
    store = app.state.device_store
    with store._state_lock:
        states = store._read_states()
        states.setdefault("ghost", DeviceState()).report = {
            "schema": 1,
            "ports": [{"port": "touch", "direction": "in", "dtype": "bool"}],
        }
        store._write_states(states)
    client.put(
        f"{FLOWS}/porch_too",
        headers=superuser_token_headers,
        json={
            "name": "porch_too",
            "version": client.get(
                f"{FLOWS}/porch_too", headers=superuser_token_headers
            ).json()["definition"]["version"],
            "nodes": [
                _device_node(
                    "ghost",
                    node="ghost",
                    provides=[{"name": "relay", "port": "relay", "dtype": "bool"}],
                    requires=[
                        {
                            "name": "touch",
                            "port": "touch",
                            "dtype": "bool",
                            "trigger": False,
                        }
                    ],
                )
            ],
        },
    )
    assert codes("porch_too") == {
        ("device_port", "relay", True),
        ("device_port", "touch", False),
    }
    for flow in ("porch", "porch_too"):
        client.delete(f"{FLOWS}/{flow}", headers=superuser_token_headers)


def test_publishing_a_flow_with_device_nodes_tells_the_devices(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """An empty dashboard name is what a device refetches its document on."""
    _panels(client, superuser_token_headers, {"panels": [{"id": "porch_io"}]})
    controller = app.state.flow_controller
    sent = []
    original = controller._publish

    def recorded(event):  # type: ignore[no-untyped-def]
        sent.append(event)
        original(event)

    controller._publish = recorded
    try:
        _publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE])
        told = [e for e in sent if e["type"] == "dashboard_changed"]
        assert told and told[-1]["dashboard"] == ""
        sent.clear()
        # A flow with no device in it moves nothing a device reads.
        _publish_flow(client, superuser_token_headers, "porch_plain", [])
        assert not [e for e in sent if e["type"] == "dashboard_changed"]
        client.delete(f"{FLOWS}/porch", headers=superuser_token_headers)
        assert [e for e in sent if e["type"] == "dashboard_changed"]
    finally:
        del controller._publish
        client.delete(f"{FLOWS}/porch_plain", headers=superuser_token_headers)


def test_a_panel_may_read_its_dashboards_own_settings(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A bound setting is no widget's binding, and a panel still needs it.

    The allowlist is a walk of what the dashboards name, so the walk has to
    reach the dashboard itself: a theme driven over a message is the one thing
    a screen on a wall cannot be told any other way, and a panel refused that
    message would leave the feature silently doing nothing on the only surface
    it exists for.
    """
    from fluksio.api.routes.flows import panel_scope

    _dashboard_with(
        client, superuser_token_headers, "panel_themed", [PANEL_WIDGETS[-1]]
    )
    saved = client.get(
        f"{DASHBOARDS}/panel_themed?draft=true", headers=superuser_token_headers
    ).json()
    saved["settings"] = {
        "theme": {"value": "dark", "message": "demo.panel_theme", "dtype": "str"},
        # Bound to nothing: a static setting entitles a panel to nothing.
        "locked": {"value": False},
    }
    written = client.put(
        f"{DASHBOARDS}/panel_themed", headers=superuser_token_headers, json=saved
    )
    assert written.status_code == 200, written.text
    published = client.post(
        f"{DASHBOARDS}/panel_themed/publish",
        headers=superuser_token_headers,
        json={"version": written.json()["version"]},
    )
    assert published.status_code == 200, published.text

    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "hall", "dashboards": ["panel_themed"]}]},
    )
    panel_headers = _pair(client, superuser_token_headers, "hall")
    messages = f"{settings.API_V1_STR}/messages"

    # Not 403: the gate lets it through, and the engine then answers for a
    # message no flow in this suite declares.
    assert (
        client.get(
            f"{messages}/demo.panel_theme/history", headers=panel_headers
        ).status_code
        == 200
    )
    # And the socket is bounded by the same walk, so the value actually lands.
    token = panel_headers["Authorization"][7:]
    assert panel_scope(token, client.app) == {
        "demo.panel_theme",
        "demo.temperature",
    }

    # A setting still buys nothing beyond itself.
    assert (
        client.get(
            f"{messages}/demo.unrelated/history", headers=panel_headers
        ).status_code
        == 403
    )


def test_a_dashboard_setting_bound_to_the_wrong_type_is_refused(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """The same answer the server gives a mis-wired widget."""
    saved = _dashboard(client, superuser_token_headers, "panel_mistyped")
    saved["settings"] = {"theme": {"value": "dark", "message": "a.b", "dtype": "float"}}

    refused = client.put(
        f"{DASHBOARDS}/panel_mistyped", headers=superuser_token_headers, json=saved
    )
    assert refused.status_code == 422, refused.text


def test_unpairing_a_screen_leaves_its_panel_standing(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Bumping the nonce revokes one credential and nothing else."""
    _dashboard(client, superuser_token_headers, "panel_kept")
    assigned = {
        "panels": [{"id": "kitchen", "title": "Kitchen", "dashboards": ["panel_kept"]}]
    }
    _panels(client, superuser_token_headers, assigned)
    panel_headers = _pair(client, superuser_token_headers, "kitchen")
    assert (
        client.get(f"{DASHBOARDS}/panel_kept", headers=panel_headers).status_code == 200
    )

    bumped = client.post(f"{PREFIX}/kitchen/unpair", headers=superuser_token_headers)
    assert bumped.status_code == 200, bumped.text

    # 401, not 403: this is no longer a credential, so the device goes back to
    # showing a code rather than retrying what it was refused.
    assert (
        client.get(f"{DASHBOARDS}/panel_kept", headers=panel_headers).status_code == 401
    )

    # The panel, its title and its assignment are exactly where they were, and
    # the next screen pairs to it normally.
    panel = client.get(f"{PREFIX}/kitchen", headers=superuser_token_headers).json()
    assert panel["title"] == "Kitchen"
    assert panel["dashboards"] == ["panel_kept"]
    fresh = _pair(client, superuser_token_headers, "kitchen")
    assert fresh != panel_headers
    assert client.get(f"{DASHBOARDS}/panel_kept", headers=fresh).status_code == 200

    # A save of the device cannot undo the revocation: the nonce is not part
    # of what is saved.
    _panels(client, superuser_token_headers, assigned)
    assert (
        client.get(f"{DASHBOARDS}/panel_kept", headers=panel_headers).status_code == 401
    )
    assert client.get(f"{DASHBOARDS}/panel_kept", headers=fresh).status_code == 200

    assert (
        client.post(
            f"{PREFIX}/nowhere/unpair", headers=superuser_token_headers
        ).status_code
        == 404
    )
    assert client.post(f"{PREFIX}/kitchen/unpair").status_code == 401


class _FakeRedis:
    """Just enough Redis for the pending-code store: strings and one zset.

    Expiry is not enforced here — each entry carries its own, which is what the
    store reads — so this only has to be shared to stand in for two workers.
    """

    def __init__(self) -> None:
        self.strings: dict[str, str] = {}
        self.zset: dict[str, float] = {}

    def set(
        self, key: str, value: str, ex: int | None = None, nx: bool = False
    ) -> bool | None:
        if nx and key in self.strings:
            return None
        self.strings[key] = value
        return True

    def get(self, key: str) -> str | None:
        return self.strings.get(key)

    def delete(self, key: str) -> None:
        self.strings.pop(key, None)

    def zadd(self, key: str, mapping: dict[str, float]) -> None:
        self.zset.update(mapping)

    def zcard(self, key: str) -> int:
        return len(self.zset)

    def zrange(self, key: str, start: int, end: int) -> list[str]:
        return sorted(self.zset, key=self.zset.__getitem__)

    def zrem(self, key: str, member: str) -> None:
        self.zset.pop(member, None)

    def zremrangebyscore(self, key: str, low: str, high: float) -> None:
        for member in [m for m, score in self.zset.items() if score <= high]:
            del self.zset[member]


def test_a_pending_code_is_found_by_whichever_worker_is_polled() -> None:
    """A poll lands wherever the proxy sent it, not on the minting worker."""
    import time

    shared = _FakeRedis()
    minted = devices_route._PendingStore()
    minted._redis = shared  # type: ignore[assignment]
    polled = devices_route._PendingStore()
    polled._redis = shared  # type: ignore[assignment]

    entry = devices_route._Pending(
        secret="s", expires=time.time() + devices_route.PAIR_TTL, device="a tablet"
    )
    assert minted.add("ABC123", entry)
    # The cap and the collision check mean the same thing on both.
    assert not polled.add("ABC123", entry)
    assert polled.count() == minted.count() == 1

    seen = polled.get("ABC123")
    assert seen is not None and seen.device == "a tablet"

    # An approval on one worker is collected from the other.
    seen.token = "minted-over-there"
    polled.save("ABC123", seen)
    collected = minted.get("ABC123")
    assert collected is not None and collected.token == "minted-over-there"

    minted.drop("ABC123")
    assert polled.get("ABC123") is None
    assert polled.count() == 0

    # An entry past its own expiry is nothing, whatever the key still says.
    stale = devices_route._Pending(secret="s", expires=time.time() - 1)
    assert minted.add("STALE1", stale)
    assert polled.get("STALE1") is None
    assert polled.count() == 0


def test_removing_the_panel_revokes_its_credential(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    _dashboard(client, superuser_token_headers, "panel_gone")
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "workshop", "dashboards": ["panel_gone"]}]},
    )
    panel_headers = _pair(client, superuser_token_headers, "workshop")
    assert (
        client.get(f"{DASHBOARDS}/panel_gone", headers=panel_headers).status_code == 200
    )

    _panels(client, superuser_token_headers, {"panels": []})
    # 401, not 403: there is nothing left to be forbidden from, and the device
    # should go back to the pairing screen rather than retry.
    assert (
        client.get(f"{DASHBOARDS}/panel_gone", headers=panel_headers).status_code == 401
    )

    # Nor does a device created again under the same id take it back.
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "workshop", "dashboards": ["panel_gone"]}]},
    )
    assert (
        client.get(f"{DASHBOARDS}/panel_gone", headers=panel_headers).status_code == 401
    )


def test_firmware_from_before_devices_pairs_through_the_panel_aliases(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """The openpanel firmware's exact calls: an empty JSON body, then a poll."""
    _panels(client, superuser_token_headers, {"panels": [{"id": "porch"}]})
    started = client.post(
        f"{PANELS}/pair", content=b"", headers={"Content-Type": "application/json"}
    )
    assert started.status_code == 200, started.text
    code, secret = started.json()["code"], started.json()["secret"]
    approved = client.post(
        f"{PREFIX}/porch/pair", headers=superuser_token_headers, json={"code": code}
    )
    assert approved.status_code == 200, approved.text
    collected = client.get(f"{PANELS}/pair/{code}", params={"secret": secret}).json()
    assert collected["panel"] == "porch"
    headers = {"Authorization": f"Bearer {collected['access_token']}"}
    assert client.get(f"{PANELS}/porch", headers=headers).json()["id"] == "porch"


def test_a_device_on_this_network_is_adopted_from_the_pending_list(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """No code to type for a request from the LAN, but one through the portal."""
    _panels(client, superuser_token_headers, {"panels": [{"id": "knob"}]})
    lan = TestClient(app, client=("192.168.1.50", 50000))
    started = lan.post(
        f"{PREFIX}/pair", json={"name": "knob-3f2a", "board": "knob", "firmware": "1.4"}
    ).json()
    remote = client.post(
        f"{PREFIX}/pair", headers=_via_portal(), json={"name": "x" * 500}
    ).json()

    pending = client.get(f"{PREFIX}/pending", headers=superuser_token_headers).json()
    mine = next(entry for entry in pending if entry["name"] == "knob-3f2a")
    assert mine["board"] == "knob" and mine["firmware"] == "1.4"
    assert mine["needs_code"] is False
    portal = next(entry for entry in pending if entry["remote"])
    assert portal["needs_code"] is True
    assert len(portal["name"]) == devices_route.HELLO_LENGTH
    # The list never hands a code out.
    assert started["code"] not in str(pending) and remote["code"] not in str(pending)

    refused = client.post(
        f"{PREFIX}/knob/pair",
        headers=superuser_token_headers,
        json={"request": portal["request"]},
    )
    assert refused.status_code == 403

    adopted = client.post(
        f"{PREFIX}/knob/pair",
        headers=superuser_token_headers,
        json={"request": mine["request"]},
    )
    assert adopted.status_code == 200, adopted.text
    collected = lan.get(
        f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
    ).json()
    assert collected["device"] == "knob"
    headers = {"Authorization": f"Bearer {collected['access_token']}"}
    assert client.get(f"{PREFIX}/knob", headers=headers).status_code == 200


# --------------------------------------------------------------------------
# A screen that reached the portal but not this instance
# --------------------------------------------------------------------------


def _via_portal() -> dict[str, str]:
    """The header the connector stamps on a request it replayed off the tunnel."""
    return {cloud_config.VIA_HEADER: cloud_config.VIA_PORTAL}


def test_the_device_asking_is_named_before_anyone_approves(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Approving a code adopts whatever holds it, so it is worth a look."""
    # Behind a trusted proxy (127.0.0.1 by default), as the tunnel replays it.
    proxied = TestClient(app, client=("127.0.0.1", 50000))
    started = proxied.post(
        f"{PREFIX}/pair",
        headers={
            "user-agent": "Mozilla/5.0 (X11; CrOS aarch64)",
            "x-forwarded-for": "198.51.100.1, 203.0.113.7",
        },
    ).json()

    looked = client.get(
        f"{PREFIX}/pair/{started['code']}/device", headers=superuser_token_headers
    )
    assert looked.status_code == 200
    assert "CrOS" in looked.json()["device"]
    # The hop the proxy added, not the one the device wrote in front of it.
    assert "203.0.113.7" in looked.json()["device"]
    assert "198.51.100.1" not in looked.json()["device"]
    assert looked.json()["remote"] is False

    # Nobody without an account gets to enumerate what is waiting.
    assert client.get(f"{PREFIX}/pair/{started['code']}/device").status_code == 401
    assert (
        client.get(
            f"{PREFIX}/pair/ZZZZZZ/device", headers=superuser_token_headers
        ).status_code
        == 404
    )


def test_a_remote_device_is_paired_at_the_portal(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    enrolled: object,  # noqa: ARG001 (fixture installs the enrolment)
    monkeypatch,  # type: ignore[no-untyped-def]
) -> None:
    """A device that arrived through the tunnel gets the portal's credential.

    It could never present one this instance signed: the portal verifies
    what crosses it, and it verifies against its own key.
    """
    import httpx

    calls: list[dict[str, object]] = []

    def fake_post(url: str, **kwargs: object) -> httpx.Response:
        calls.append({"url": url, **kwargs})
        return httpx.Response(
            200,
            json={"access_token": "minted-by-the-portal", "expires_in": 31536000},
            request=httpx.Request("POST", url),
        )

    monkeypatch.setattr(devices_route.httpx, "post", fake_post)

    _panels(client, superuser_token_headers, {"panels": [{"id": "hallway"}]})
    started = client.post(f"{PREFIX}/pair", headers=_via_portal()).json()

    looked = client.get(
        f"{PREFIX}/pair/{started['code']}/device", headers=superuser_token_headers
    ).json()
    assert looked["remote"] is True

    approved = client.post(
        f"{PREFIX}/hallway/pair",
        headers=superuser_token_headers,
        json={"code": started["code"]},
    )
    assert approved.status_code == 200, approved.text

    assert calls[0]["url"].endswith("/api/v1/panel-tokens/")  # type: ignore[union-attr]
    assert calls[0]["headers"]["Authorization"] == "Bearer instance-token"  # type: ignore[index]
    assert calls[0]["json"] == {"panel": "hallway"}  # type: ignore[index]

    collected = client.get(
        f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
    ).json()
    assert collected["access_token"] == "minted-by-the-portal"


def test_a_remote_device_needs_an_enrolment(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Unenrolled, there is nowhere to ask — and no token to invent locally."""
    _panels(client, superuser_token_headers, {"panels": [{"id": "shed"}]})
    started = client.post(f"{PREFIX}/pair", headers=_via_portal()).json()

    approved = client.post(
        f"{PREFIX}/shed/pair",
        headers=superuser_token_headers,
        json={"code": started["code"]},
    )
    assert approved.status_code == 409


def test_a_panel_credential_cannot_move_artifacts(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """The artifact store is a person's or a worker's, not a screen's.

    It authenticates for itself rather than through the shared dependency, so
    the scope check has to be handed the request there too — without it a panel
    would act as whoever approved it.
    """
    client.post(f"{DASHBOARDS}/panel_art", headers=superuser_token_headers)
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "hallway", "dashboards": ["panel_art"]}]},
    )
    panel_headers = _pair(client, superuser_token_headers, "hallway")

    assert (
        client.put(
            f"{settings.API_V1_STR}/artifacts",
            headers=panel_headers,
            params={"name": "x"},
            content=b"hello",
        ).status_code
        == 403
    )


def test_a_local_device_cannot_claim_to_be_remote(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Otherwise it would be handed a credential the whole internet accepts."""
    started = client.post(f"{PREFIX}/pair", headers={"x-fluksio-via": "portal"}).json()
    looked = client.get(
        f"{PREFIX}/pair/{started['code']}/device", headers=superuser_token_headers
    )
    assert looked.json()["remote"] is False


def test_a_panels_socket_carries_only_what_it_draws(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A screen may hang anywhere, so it is sent its own values and nothing else.

    The handshake has no route for the scope check to judge, so the bound is
    applied to what goes out: no node status, no logs, no shape of the graph,
    and no value belonging to a dashboard this panel does not show.
    """
    from fluksio.api.routes.flows import (
        event_for_panel,
        panel_scope,
        snapshot_payload,
    )

    saved = _dashboard(client, superuser_token_headers, "panel_socket")
    saved["widgets"] = [
        {
            "id": "w1",
            "type": "stat",
            "config": {"message": "house.kitchen.temperature"},
        }
    ]
    written = client.put(
        f"{DASHBOARDS}/panel_socket", headers=superuser_token_headers, json=saved
    )
    assert written.status_code == 200, written.text
    # A panel reads the published document, so the draft has to be promoted.
    published = client.post(
        f"{DASHBOARDS}/panel_socket/publish",
        headers=superuser_token_headers,
        json={"version": written.json()["version"]},
    )
    assert published.status_code == 200, published.text

    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "hallway", "dashboards": ["panel_socket"]}]},
    )
    token = _pair(client, superuser_token_headers, "hallway")["Authorization"][7:]

    only = panel_scope(token, client.app)
    assert only == {"house.kitchen.temperature"}

    class _Controller:
        def values(self) -> dict[str, dict[str, str]]:
            return {
                "house.kitchen.temperature": {"value": "21"},
                "house.safe.code": {"value": "1234"},
            }

    payload = snapshot_payload(_Controller(), only)  # type: ignore[arg-type]
    assert set(payload["values"]) == {"house.kitchen.temperature"}
    # Emptied rather than dropped: a hanging screen runs the bundle it was
    # paired with, and that bundle reads these keys.
    assert payload["logs"] == [] and payload["nodes"] == []
    assert payload["issues"] == [] and payload["emits"] == {}

    # The stream that follows the snapshot is bounded the same way: a value it
    # draws goes out, one it does not stays here, and nothing else travels.
    assert event_for_panel(
        {"type": "message_value", "name": "house.kitchen.temperature"}, only
    )
    assert not event_for_panel(
        {"type": "message_value", "name": "house.safe.code"}, only
    )
    assert not event_for_panel({"type": "node_log", "text": "a traceback"}, only)
    # Except a dashboard being published: that is how a screen hears the
    # document it draws — or the set of them it was given — has moved.
    assert event_for_panel({"type": "dashboard_changed", "dashboard": "x"}, only)

    # A person's credential is not bounded at all.
    assert panel_scope(superuser_token_headers["Authorization"][7:], client.app) is None

    # It opens the socket with a ticket like any browser, and the ticket
    # carries the panel's bound rather than widening it.
    ticket = client.post(
        f"{settings.API_V1_STR}/flows/ws-ticket",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert ticket.status_code == 200, ticket.text
    with client.websocket_connect(
        f"{settings.API_V1_STR}/flows/ws?ticket={ticket.json()['ticket']}"
    ) as ws:
        assert ws.receive_json()["nodes"] == []


def test_a_panel_fetches_only_the_media_its_tiles_are_showing(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A media tile needs the bytes, and only the ones its message points at."""
    _dashboard_with(
        client,
        superuser_token_headers,
        "panel_camera",
        [
            {
                "id": "w_media",
                "type": "media",
                "config": {"message": "demo.frame", "dtype": "image"},
            }
        ],
    )
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "porch", "dashboards": ["panel_camera"]}]},
    )
    panel_headers = _pair(client, superuser_token_headers, "porch")

    store = client.app.state.artifact_store
    shown = store.put([b"the frame on the wall"], media_type="image/png")
    elsewhere = store.put([b"an artifact of some other run"])
    client.app.state.flow_controller.state.update({"demo.frame": shown})

    artifacts = f"{settings.API_V1_STR}/artifacts"
    assert (
        client.get(f"{artifacts}/{shown['digest']}", headers=panel_headers).status_code
        == 200
    )
    # Knowing a digest is not being entitled to it: a screen reads what it draws.
    assert (
        client.get(
            f"{artifacts}/{elsewhere['digest']}", headers=panel_headers
        ).status_code
        == 403
    )


def test_a_socket_pushes_only_the_frames_it_was_asked_for(tmp_path) -> None:
    """Bytes are the one thing this socket cannot send speculatively.

    A frame a second per subscriber is affordable; every frame to every open
    editor tab is not. So nothing goes until a client names what it is drawing,
    and a panel credential can only name what it was already allowed to see.
    """
    import orjson

    from fluksio.api.routes.flows import media_frames, wanted_names
    from fluksio.flow.artifacts import ArtifactStore, VolatileStore

    store = ArtifactStore(tmp_path / "artifacts")
    store.volatile = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000)
    frame = store.put(
        [b"\x89PNG..."], name="f.png", media_type="image/png", volatile=True
    )
    kept = store.put([b"checkpoint"], name="w.pt")

    asking = orjson.dumps({"type": "media", "names": ["cam.frame", "other.frame"]})
    # A person's socket gets what it asked for; a panel's is intersected with
    # what it draws, so naming a message is not a way around the scope.
    assert wanted_names(asking.decode(), None) == {"cam.frame", "other.frame"}
    assert wanted_names(asking.decode(), {"cam.frame"}) == {"cam.frame"}
    # Anything else on this socket leaves the set alone.
    assert wanted_names('{"type":"ping"}', None) is None
    assert wanted_names("not json", None) is None

    events = [
        {"type": "message_value", "name": "cam.frame", "value": frame, "ts": 1.0},
        {"type": "message_value", "name": "other.frame", "value": frame, "ts": 1.0},
        {"type": "message_value", "name": "cam.model", "value": kept, "ts": 1.0},
        {"type": "node_log", "name": "cam.frame", "value": frame},
    ]
    frames = media_frames(events, {"cam.frame", "cam.model"}, store)

    # One frame: the ring's, for the name that asked. The durable checkpoint is
    # what a fetch is for, and a log is not a value.
    assert len(frames) == 1
    length = int.from_bytes(frames[0][:4], "big")
    header = orjson.loads(frames[0][4 : 4 + length])
    assert header == {
        "type": "media",
        "name": "cam.frame",
        "digest": frame["digest"],
        "media_type": "image/png",
        "ts": 1.0,
    }
    assert frames[0][4 + length :] == b"\x89PNG..."

    # Nothing asked for, nothing sent.
    assert media_frames(events, set(), store) == []


def test_only_the_newest_frame_of_a_batch_is_pushed(tmp_path) -> None:
    """A client that fell behind is not handed frames it would only draw over."""
    import orjson

    from fluksio.api.routes.flows import media_frames
    from fluksio.flow.artifacts import ArtifactStore, VolatileStore

    store = ArtifactStore(tmp_path / "artifacts")
    store.volatile = VolatileStore(tmp_path / "ring", limit_bytes=1_000_000)
    first = store.put([b"one"], media_type="image/png", volatile=True)
    second = store.put([b"two"], media_type="image/png", volatile=True)

    frames = media_frames(
        [
            {"type": "message_value", "name": "cam.frame", "value": first, "ts": 1.0},
            {"type": "message_value", "name": "cam.frame", "value": second, "ts": 2.0},
        ],
        {"cam.frame"},
        store,
    )

    assert len(frames) == 1
    length = int.from_bytes(frames[0][:4], "big")
    assert orjson.loads(frames[0][4 : 4 + length])["digest"] == second["digest"]
    assert frames[0][4 + length :] == b"two"