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

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

from fastapi.testclient import TestClient

from fluksio.cloud import config as cloud_config
from fluksio.core.config import settings

PREFIX = f"{settings.API_V1_STR}/panels"
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"


def _panels(client: TestClient, headers: dict[str, str], config: dict) -> None:
    response = client.put(f"{PREFIX}/", headers=headers, json=config)
    assert response.status_code == 200, response.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["panel"] == panel
    return {"Authorization": f"Bearer {collected['access_token']}"}


def test_panels_require_authentication(client: TestClient) -> None:
    assert client.get(f"{PREFIX}/").status_code == 401
    assert client.put(f"{PREFIX}/", json={"panels": []}).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()
    assert stored["panels"][0]["dashboards"] == ["hall_a", "hall_b"]
    # 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_panel(
    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 = {
        panel["id"]: panel
        for panel in client.get(f"{PREFIX}/", headers=superuser_token_headers).json()[
            "panels"
        ]
    }
    assert by_id["wall"]["touch"] is True
    # Absent is pointed at, which is what a panels file written before this
    # field existed comes back as.
    assert by_id["desk"]["touch"] is False


def test_duplicate_panel_is_refused(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    response = client.put(
        f"{PREFIX}/",
        headers=superuser_token_headers,
        json={"panels": [{"id": "twice"}, {"id": "twice"}]},
    )
    assert response.status_code == 422


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
    # 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"},
    },
    {"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.
    for name in ("demo.temperature", "demo.switch_state", "demo.query_series"):
        assert (
            client.get(f"{messages}/{name}/history", headers=panel_headers).status_code
            == 200
        ), 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
    )


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 client holding an older copy of the panels cannot undo the revocation
    # by writing the nonce back.
    _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 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

    from fluksio.api.routes import panels as panels_route

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

    entry = panels_route._Pending(
        secret="s", expires=time.time() + panels_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 = panels_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
    )


# --------------------------------------------------------------------------
# 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."""
    started = client.post(
        f"{PREFIX}/pair",
        headers={
            "user-agent": "Mozilla/5.0 (X11; CrOS aarch64)",
            "x-forwarded-for": "203.0.113.7, 10.0.0.1",
        },
    ).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 first hop, not the proxy that relayed it.
    assert "203.0.113.7" 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

    from fluksio.api.routes import panels as panels_route

    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(panels_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


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"