mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
a2c7fd4542
The 0.2.3b4 #1003 "fix" POSTed JPEG bytes as multipart form data,
but Obico's /p/ endpoint is declared methods=['GET'] upstream and
reads ?img=URL from the query string. Every POST was 405'd by
Flask's router before any handler ran, which is why the Obico
container logs were silent while Bambuddy kept reporting
"ML API call failed for printer N:" with a blank suffix —
raise_for_status() on the 405 produced an exception whose str()
rendered empty.
Restored the pre-#1003 nonce-URL approach (commit 3e434458):
capture locally with a 20s timeout we control, stash the JPEG
under a single-use 32-byte nonce, hand Obico a
GET /api/v1/obico/cached-frame/{nonce} URL that resolves in
<50ms so its hardcoded 5s read timeout never races RTSP.
Also guards against future silent exceptions: the error format
now falls back to type(exc).__name__ when str(exc) is empty.
Detection also early-returns with an explicit error if
external_url is unset instead of handing Obico a URL it can't
resolve.
The #1003 reverse-proxy scenario (Authelia/Authentik/CF Access
in front of Bambuddy) is addressed by documenting that the
/api/v1/obico/cached-frame/ path must be whitelisted from
external auth at the proxy layer — it is already public on
Bambuddy's side.
Backend: services/obico_detection.py, api/routes/obico.py,
main.py (PUBLIC_API_PATTERNS).
Frontend: FailureDetectionSettings banner + client.ts type +
all 7 locales restored.
Tests: 15 unit + 5 integration tests pass.
72 lines
3.0 KiB
Python
72 lines
3.0 KiB
Python
"""Integration tests for Obico API endpoints (#172 follow-up).
|
|
|
|
Verifies the /obico/cached-frame/{nonce} endpoint used by Obico's ML API to fetch
|
|
pre-captured JPEG frames. This endpoint lets the detection loop sidestep Obico's
|
|
hardcoded 5s read timeout by pre-populating a cache before issuing the ML call.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from backend.app.services.obico_detection import _frame_cache, stash_frame
|
|
|
|
FAKE_JPEG = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xd9"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_cache():
|
|
_frame_cache.clear()
|
|
yield
|
|
_frame_cache.clear()
|
|
|
|
|
|
class TestObicoCachedFrame:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_valid_nonce_returns_jpeg(self, async_client: AsyncClient):
|
|
"""A stashed nonce returns the stored JPEG bytes with image/jpeg."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
response = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "image/jpeg"
|
|
assert response.content == FAKE_JPEG
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unknown_nonce_is_404(self, async_client: AsyncClient):
|
|
"""An unguessable URL must not leak that the endpoint exists — return 404."""
|
|
response = await async_client.get("/api/v1/obico/cached-frame/definitely-not-a-real-nonce")
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_nonce_is_single_use(self, async_client: AsyncClient):
|
|
"""A second fetch with the same nonce returns 404 — prevents replay."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
first = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
assert first.status_code == 200
|
|
second = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
assert second.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_endpoint_is_public(self, async_client: AsyncClient):
|
|
"""Obico's ML API can't send auth headers, so the nonce IS the credential.
|
|
The path must be in PUBLIC_API_PATTERNS (no auth wall)."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
# Intentionally omit any auth headers even if the fixture would normally inject them
|
|
response = await async_client.get(
|
|
f"/api/v1/obico/cached-frame/{nonce}",
|
|
headers={}, # no Authorization header
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_response_is_not_cached(self, async_client: AsyncClient):
|
|
"""Browsers/proxies must not hold onto the image after Obico consumes it."""
|
|
nonce = await stash_frame(FAKE_JPEG)
|
|
response = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
|
|
assert response.status_code == 200
|
|
assert "no-store" in response.headers.get("cache-control", "")
|