fix: return structured 503 on camera read failure in /api/observe

Wrap camera.read_frame() call in try/except to catch RuntimeError and
return a 503 Service Unavailable response with structured error JSON
(error + detail fields) instead of letting the exception propagate
as an unhandled 500. This gives kiosk operators a friendly, actionable
error message when the camera becomes unavailable.

Added FailingCamera test fixture and regression test that verifies
the 503 response with expected error details.
This commit is contained in:
pyr0ball 2026-07-13 10:58:30 -07:00
parent ae73ab2329
commit 7d0bebf16d
2 changed files with 36 additions and 1 deletions

View file

@ -1,5 +1,6 @@
import cv2
from fastapi import APIRouter, Request, Response
from fastapi.responses import JSONResponse
router = APIRouter()
@ -14,7 +15,10 @@ def get_session_state(request: Request):
def observe(request: Request):
session = request.app.state.session
camera = request.app.state.camera
frame = camera.read_frame()
try:
frame = camera.read_frame()
except RuntimeError as exc:
return JSONResponse(status_code=503, content={"error": "Camera unavailable", "detail": str(exc)})
request.app.state.latest_frame = frame
fired = session.observe_frame(frame)
return {"fired": fired, "state": session.state.value, "book_id": session.current_book_id}

View file

@ -27,6 +27,14 @@ class FakeCamera:
return self.frame
class FailingCamera:
def __init__(self, error_message: str = "Camera not initialized") -> None:
self.error_message = error_message
def read_frame(self) -> np.ndarray:
raise RuntimeError(self.error_message)
def make_client(tmp_path):
catalogue = Catalogue(tmp_path / "catalogue.csv")
strategy = ManualButtonStrategy()
@ -81,3 +89,26 @@ def test_preview_returns_jpeg_after_observe(tmp_path):
response = client.get("/api/preview.jpg")
assert response.status_code == 200
assert response.headers["content-type"] == "image/jpeg"
def test_observe_returns_503_on_camera_failure(tmp_path):
catalogue = Catalogue(tmp_path / "catalogue.csv")
strategy = ManualButtonStrategy()
reference = make_reference()
markers = [MarkerRegion(name="small_boundary", bucket="small", x=0, y=0, width=10, height=10)]
session = IntakeSession(
catalogue=catalogue,
strategy=strategy,
reference_frame=reference,
markers=markers,
captures_dir=tmp_path / "captures",
)
camera = FailingCamera("Camera connection lost")
app = create_app(session, camera)
client = TestClient(app)
response = client.post("/api/observe")
assert response.status_code == 503
body = response.json()
assert body["error"] == "Camera unavailable"
assert "Camera connection lost" in body["detail"]