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

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

from fastapi.testclient import TestClient

from app.cloud import config as cloud_config
from app.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
    assert (
        client.get(
            f"{settings.API_V1_STR}/messages/", headers=panel_headers
        ).status_code
        == 200
    )

    # 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


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 app.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 app.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