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.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 _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"):
        client.post(f"{DASHBOARDS}/{name}", headers=superuser_token_headers)
    _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:
    client.post(f"{DASHBOARDS}/panel_gone", headers=superuser_token_headers)
    _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 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={"x-fluksio-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={"x-fluksio-via": "portal"}).json()

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