|
@@ -14,34 +14,9 @@ from __future__ import annotations
|
|
|
import pytest
|
|
import pytest
|
|
|
from httpx import AsyncClient
|
|
from httpx import AsyncClient
|
|
|
|
|
|
|
|
-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"overlayadmin{suffix}",
|
|
|
|
|
- "admin_password": "AdminPass1!",
|
|
|
|
|
- },
|
|
|
|
|
- )
|
|
|
|
|
- login = await async_client.post(
|
|
|
|
|
- "/api/v1/auth/login",
|
|
|
|
|
- json={"username": f"overlayadmin{suffix}", "password": "AdminPass1!"},
|
|
|
|
|
- )
|
|
|
|
|
- return login.json()["access_token"]
|
|
|
|
|
-
|
|
|
|
|
|
|
+from backend.tests.overlay_helpers import mint_token, setup_admin
|
|
|
|
|
|
|
|
-async def _mint(async_client: AsyncClient, jwt: str, *, scope: str, name: str = "obs") -> 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"]
|
|
|
|
|
|
|
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
@pytest.fixture
|
|
@@ -67,12 +42,12 @@ async def printer_row(db_session):
|
|
|
|
|
|
|
|
class TestOverlayFeedAuth:
|
|
class TestOverlayFeedAuth:
|
|
|
async def test_no_token_is_rejected(self, async_client: AsyncClient, printer_row):
|
|
async def test_no_token_is_rejected(self, async_client: AsyncClient, printer_row):
|
|
|
- await _setup_admin(async_client, suffix="_notoken")
|
|
|
|
|
|
|
+ await setup_admin(async_client, suffix="_notoken")
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status")
|
|
|
assert response.status_code == 401
|
|
assert response.status_code == 401
|
|
|
|
|
|
|
|
async def test_garbage_token_is_rejected(self, async_client: AsyncClient, printer_row):
|
|
async def test_garbage_token_is_rejected(self, async_client: AsyncClient, printer_row):
|
|
|
- await _setup_admin(async_client, suffix="_garbage")
|
|
|
|
|
|
|
+ await setup_admin(async_client, suffix="_garbage")
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token=bblt_aaaaaaaa_nope")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token=bblt_aaaaaaaa_nope")
|
|
|
assert response.status_code == 401
|
|
assert response.status_code == 401
|
|
|
|
|
|
|
@@ -81,8 +56,8 @@ class TestOverlayFeedAuth:
|
|
|
acquire the live print status (and filename) just because a new feature
|
|
acquire the live print status (and filename) just because a new feature
|
|
|
shipped.
|
|
shipped.
|
|
|
"""
|
|
"""
|
|
|
- jwt = await _setup_admin(async_client, suffix="_streamscope")
|
|
|
|
|
- stream_token = await _mint(async_client, jwt, scope="camera_stream")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_streamscope")
|
|
|
|
|
+ stream_token = await mint_token(async_client, jwt, scope="camera_stream")
|
|
|
|
|
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={stream_token}")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={stream_token}")
|
|
|
assert response.status_code == 401
|
|
assert response.status_code == 401
|
|
@@ -95,22 +70,22 @@ class TestOverlayFeedAuth:
|
|
|
rejected here — otherwise every wall token silently gains filename
|
|
rejected here — otherwise every wall token silently gains filename
|
|
|
visibility.
|
|
visibility.
|
|
|
"""
|
|
"""
|
|
|
- jwt = await _setup_admin(async_client, suffix="_camwallscope")
|
|
|
|
|
- camwall_token = await _mint(async_client, jwt, scope="camwall")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_camwallscope")
|
|
|
|
|
+ camwall_token = await mint_token(async_client, jwt, scope="camwall")
|
|
|
|
|
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={camwall_token}")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={camwall_token}")
|
|
|
assert response.status_code == 401
|
|
assert response.status_code == 401
|
|
|
|
|
|
|
|
async def test_overlay_token_reaches_the_feed(self, async_client: AsyncClient, printer_row):
|
|
async def test_overlay_token_reaches_the_feed(self, async_client: AsyncClient, printer_row):
|
|
|
- jwt = await _setup_admin(async_client, suffix="_rightscope")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_rightscope")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
|
assert response.status_code == 200, response.text
|
|
assert response.status_code == 200, response.text
|
|
|
assert response.json()["name"] == "Stream P1S"
|
|
assert response.json()["name"] == "Stream P1S"
|
|
|
|
|
|
|
|
async def test_revoked_overlay_token_is_rejected(self, async_client: AsyncClient, printer_row):
|
|
async def test_revoked_overlay_token_is_rejected(self, async_client: AsyncClient, printer_row):
|
|
|
- jwt = await _setup_admin(async_client, suffix="_revoked")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_revoked")
|
|
|
created = await async_client.post(
|
|
created = await async_client.post(
|
|
|
"/api/v1/auth/tokens",
|
|
"/api/v1/auth/tokens",
|
|
|
headers={"Authorization": f"Bearer {jwt}"},
|
|
headers={"Authorization": f"Bearer {jwt}"},
|
|
@@ -132,8 +107,8 @@ class TestOverlayFeedPayload:
|
|
|
that is what distinguishes the scope. Assert the exact key set so the
|
|
that is what distinguishes the scope. Assert the exact key set so the
|
|
|
payload can't silently grow to leak more than the overlay draws.
|
|
payload can't silently grow to leak more than the overlay draws.
|
|
|
"""
|
|
"""
|
|
|
- jwt = await _setup_admin(async_client, suffix="_payload")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_payload")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
|
assert response.status_code == 200
|
|
assert response.status_code == 200
|
|
@@ -165,8 +140,8 @@ class TestOverlayFeedPayload:
|
|
|
"""No MQTT client runs in tests, so the printer has no state — the
|
|
"""No MQTT client runs in tests, so the printer has no state — the
|
|
|
overlay must render its offline state rather than erroring.
|
|
overlay must render its offline state rather than erroring.
|
|
|
"""
|
|
"""
|
|
|
- jwt = await _setup_admin(async_client, suffix="_offline")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_offline")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
|
entry = response.json()
|
|
entry = response.json()
|
|
@@ -210,8 +185,8 @@ class TestOverlayFeedPayload:
|
|
|
|
|
|
|
|
monkeypatch.setattr(pm.printer_manager, "get_status", lambda _pid: _FakeState())
|
|
monkeypatch.setattr(pm.printer_manager, "get_status", lambda _pid: _FakeState())
|
|
|
|
|
|
|
|
- jwt = await _setup_admin(async_client, suffix="_temps")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_temps")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
|
assert response.json()["model"] == "P1S"
|
|
assert response.json()["model"] == "P1S"
|
|
@@ -229,8 +204,8 @@ class TestOverlayFeedPayload:
|
|
|
async def test_unknown_model_is_null(self, async_client: AsyncClient, printer_row, db_session):
|
|
async def test_unknown_model_is_null(self, async_client: AsyncClient, printer_row, db_session):
|
|
|
printer_row.model = None
|
|
printer_row.model = None
|
|
|
await db_session.commit()
|
|
await db_session.commit()
|
|
|
- jwt = await _setup_admin(async_client, suffix="_nomodel")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_nomodel")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
response = await async_client.get(f"/api/v1/printers/{printer_row.id}/overlay-status?token={overlay_token}")
|
|
|
assert response.status_code == 200
|
|
assert response.status_code == 200
|
|
|
assert response.json()["model"] is None
|
|
assert response.json()["model"] is None
|
|
@@ -239,8 +214,8 @@ class TestOverlayFeedPayload:
|
|
|
"""A valid token for a printer id that doesn't exist is a 404 — the token
|
|
"""A valid token for a printer id that doesn't exist is a 404 — the token
|
|
|
passed the gate, the resource simply isn't there.
|
|
passed the gate, the resource simply isn't there.
|
|
|
"""
|
|
"""
|
|
|
- jwt = await _setup_admin(async_client, suffix="_404")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_404")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
|
response = await async_client.get(f"/api/v1/printers/99999/overlay-status?token={overlay_token}")
|
|
response = await async_client.get(f"/api/v1/printers/99999/overlay-status?token={overlay_token}")
|
|
|
assert response.status_code == 404
|
|
assert response.status_code == 404
|
|
@@ -254,17 +229,17 @@ class TestOverlayTokenReachesTheVideo:
|
|
|
async def test_overlay_token_passes_the_camera_stream_gate(self, async_client: AsyncClient):
|
|
async def test_overlay_token_passes_the_camera_stream_gate(self, async_client: AsyncClient):
|
|
|
from backend.app.core.auth import verify_camera_stream_token
|
|
from backend.app.core.auth import verify_camera_stream_token
|
|
|
|
|
|
|
|
- jwt = await _setup_admin(async_client, suffix="_video")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_video")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
|
assert await verify_camera_stream_token(overlay_token) is True
|
|
assert await verify_camera_stream_token(overlay_token) is True
|
|
|
|
|
|
|
|
async def test_overlay_gate_rejects_camera_stream_and_camwall(self, async_client: AsyncClient):
|
|
async def test_overlay_gate_rejects_camera_stream_and_camwall(self, async_client: AsyncClient):
|
|
|
from backend.app.core.auth import verify_overlay_token
|
|
from backend.app.core.auth import verify_overlay_token
|
|
|
|
|
|
|
|
- jwt = await _setup_admin(async_client, suffix="_gate")
|
|
|
|
|
- stream_token = await _mint(async_client, jwt, scope="camera_stream")
|
|
|
|
|
- camwall_token = await _mint(async_client, jwt, scope="camwall", name="wall")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_gate")
|
|
|
|
|
+ stream_token = await mint_token(async_client, jwt, scope="camera_stream")
|
|
|
|
|
+ camwall_token = await mint_token(async_client, jwt, scope="camwall", name="wall")
|
|
|
|
|
|
|
|
assert await verify_overlay_token(stream_token) is False
|
|
assert await verify_overlay_token(stream_token) is False
|
|
|
assert await verify_overlay_token(camwall_token) is False
|
|
assert await verify_overlay_token(camwall_token) is False
|
|
@@ -273,7 +248,7 @@ class TestOverlayTokenReachesTheVideo:
|
|
|
"""Symmetric guard: the new scope must not widen the Cam Wall either."""
|
|
"""Symmetric guard: the new scope must not widen the Cam Wall either."""
|
|
|
from backend.app.core.auth import verify_camwall_token
|
|
from backend.app.core.auth import verify_camwall_token
|
|
|
|
|
|
|
|
- jwt = await _setup_admin(async_client, suffix="_gate_camwall")
|
|
|
|
|
- overlay_token = await _mint(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
+ jwt = await setup_admin(async_client, suffix="_gate_camwall")
|
|
|
|
|
+ overlay_token = await mint_token(async_client, jwt, scope="overlay")
|
|
|
|
|
|
|
|
assert await verify_camwall_token(overlay_token) is False
|
|
assert await verify_camwall_token(overlay_token) is False
|