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_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_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["pages"] = [{"id": "main", "sections": [{"id": "main", "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.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 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 installation
# --------------------------------------------------------------------------


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 installation 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 installation-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["pages"] = [
        {
            "id": "main",
            "title": "Overview",
            "sections": [
                {
                    "id": "main",
                    "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