bambuddy/backend/tests/integration/test_camwall_api.py

"""Integration tests for the token-authenticated Cam Wall feed (#2531).

The feature's whole reason for existing as a separate endpoint (rather than
letting a token through to ``GET /printers``) is that a kiosk URL is not a
secret. So the tests that matter here are the negative ones: what a Cam Wall
token *cannot* reach, and what the payload *does not* contain.
"""

from __future__ import annotations

import pytest
from httpx import AsyncClient

from backend.app.core.printer_scope import ALL_PRINTERS

pytestmark = [pytest.mark.asyncio, pytest.mark.integration]


async def _setup_admin(async_client: AsyncClient, *, suffix: str) -> str:
    await async_client.post(
        "/api/v1/auth/setup",
        json={
            "auth_enabled": True,
            "admin_username": f"camwalladmin{suffix}",
            "admin_password": "AdminPass1!",
        },
    )
    login = await async_client.post(
        "/api/v1/auth/login",
        json={"username": f"camwalladmin{suffix}", "password": "AdminPass1!"},
    )
    return login.json()["access_token"]


async def _mint(async_client: AsyncClient, jwt: str, *, scope: str, name: str = "kiosk") -> str:
    response = await async_client.post(
        "/api/v1/auth/tokens",
        headers={"Authorization": f"Bearer {jwt}"},
        json={"name": name, "expires_in_days": 30, "scope": scope},
    )
    assert response.status_code == 201, response.text
    assert response.json()["scope"] == scope
    return response.json()["token"]


@pytest.fixture
async def printer_row(db_session):
    """Insert the printer straight into the DB.

    POST /printers probes the real device before it will store a row, and there
    is no printer on the other end of a test run.
    """
    from backend.app.models.printer import Printer

    printer = Printer(
        name="Wall P1S",
        ip_address="192.168.1.77",
        access_code="12345678",
        serial_number="01P00A000000001",
        model="P1S",
    )
    db_session.add(printer)
    await db_session.commit()
    return printer


class TestCamWallFeedAuth:
    async def test_no_token_is_rejected(self, async_client: AsyncClient):
        await _setup_admin(async_client, suffix="_notoken")
        response = await async_client.get("/api/v1/camwall/printers")
        assert response.status_code == 401

    async def test_garbage_token_is_rejected(self, async_client: AsyncClient):
        await _setup_admin(async_client, suffix="_garbage")
        response = await async_client.get("/api/v1/camwall/printers?token=bblt_aaaaaaaa_nope")
        assert response.status_code == 401

    async def test_camera_stream_token_cannot_reach_the_feed(self, async_client: AsyncClient):
        """The point of the separate scope.

        ``camera_stream`` tokens are already in the wild, minted by users who
        agreed to hand out *video*. Shipping the Cam Wall must not retroactively
        grant them the ability to enumerate printers by name.
        """
        jwt = await _setup_admin(async_client, suffix="_wrongscope")
        stream_token = await _mint(async_client, jwt, scope="camera_stream")

        response = await async_client.get(f"/api/v1/camwall/printers?token={stream_token}")
        assert response.status_code == 401

    async def test_camwall_token_reaches_the_feed(self, async_client: AsyncClient, printer_row):
        jwt = await _setup_admin(async_client, suffix="_rightscope")
        camwall_token = await _mint(async_client, jwt, scope="camwall")

        response = await async_client.get(f"/api/v1/camwall/printers?token={camwall_token}")
        assert response.status_code == 200, response.text
        body = response.json()
        assert len(body) == 1
        assert body[0]["name"] == "Wall P1S"

    async def test_revoked_camwall_token_is_rejected(self, async_client: AsyncClient):
        jwt = await _setup_admin(async_client, suffix="_revoked")
        created = await async_client.post(
            "/api/v1/auth/tokens",
            headers={"Authorization": f"Bearer {jwt}"},
            json={"name": "kiosk", "expires_in_days": 30, "scope": "camwall"},
        )
        camwall_token = created.json()["token"]
        await async_client.delete(
            f"/api/v1/auth/tokens/{created.json()['id']}",
            headers={"Authorization": f"Bearer {jwt}"},
        )

        response = await async_client.get(f"/api/v1/camwall/printers?token={camwall_token}")
        assert response.status_code == 401


class TestCamWallFeedPayload:
    async def test_payload_withholds_secrets_and_filenames(self, async_client: AsyncClient, printer_row):
        """A URL taped to a TV must not disclose more than the picture does.

        Serial number and IP ride along on the ordinary printer list even for
        non-secret callers, and the filename names the customer's part. None of
        the three may appear here.
        """
        jwt = await _setup_admin(async_client, suffix="_payload")
        camwall_token = await _mint(async_client, jwt, scope="camwall")

        response = await async_client.get(f"/api/v1/camwall/printers?token={camwall_token}")
        assert response.status_code == 200
        entry = response.json()[0]

        for leaked in ("serial_number", "ip_address", "access_code", "subtask_name", "gcode_file"):
            assert leaked not in entry, f"{leaked} must not be served to a kiosk token"

        assert set(entry) == {
            "id",
            "name",
            "camera_rotation",
            "connected",
            "state",
            "progress",
            "remaining_time",
            "layer_num",
            "total_layers",
            "hms_errors",
        }

    async def test_disconnected_printer_reports_connected_false(self, async_client: AsyncClient, printer_row):
        """No MQTT client is running in tests, so the printer has no state at
        all — the tile must render as offline rather than blank.
        """
        jwt = await _setup_admin(async_client, suffix="_offline")
        camwall_token = await _mint(async_client, jwt, scope="camwall")

        response = await async_client.get(f"/api/v1/camwall/printers?token={camwall_token}")
        entry = response.json()[0]
        assert entry["connected"] is False
        assert entry["state"] is None
        assert entry["hms_errors"] == []


class TestCamWallTokenReachesTheVideo:
    """A wall that can list the tiles but not fill them is useless — the same
    token has to satisfy the camera-stream gate.
    """

    async def test_camwall_token_passes_the_camera_stream_gate(self, async_client: AsyncClient):
        from backend.app.core.auth import verify_camera_stream_token

        jwt = await _setup_admin(async_client, suffix="_video")
        camwall_token = await _mint(async_client, jwt, scope="camwall")

        assert await verify_camera_stream_token(camwall_token) == ALL_PRINTERS  # admin-owned: every printer (#1727)

    async def test_camera_stream_token_still_passes_its_own_gate(self, async_client: AsyncClient):
        """Regression guard on #1108: widening the accepted scopes must not have
        broken the tokens that were already working.
        """
        from backend.app.core.auth import verify_camera_stream_token

        jwt = await _setup_admin(async_client, suffix="_video_legacy")
        stream_token = await _mint(async_client, jwt, scope="camera_stream")

        assert await verify_camera_stream_token(stream_token) == ALL_PRINTERS  # admin-owned: every printer (#1727)

    async def test_camwall_gate_rejects_a_camera_stream_token(self, async_client: AsyncClient):
        from backend.app.core.auth import verify_camwall_token

        jwt = await _setup_admin(async_client, suffix="_gate_narrow")
        stream_token = await _mint(async_client, jwt, scope="camera_stream")

        assert await verify_camwall_token(stream_token) is None


class TestScopeValidation:
    async def test_unknown_scope_is_rejected_at_mint(self, async_client: AsyncClient):
        jwt = await _setup_admin(async_client, suffix="_badscope")
        response = await async_client.post(
            "/api/v1/auth/tokens",
            headers={"Authorization": f"Bearer {jwt}"},
            json={"name": "x", "expires_in_days": 30, "scope": "printers_write"},
        )
        assert response.status_code == 400
        assert "unsupported scope" in response.json()["detail"].lower()