pyr0ball
7d0bebf16d
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.
114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
import numpy as np
|
|
from fastapi.testclient import TestClient
|
|
|
|
from bookmark.capture_strategies.manual_button import ManualButtonStrategy
|
|
from bookmark.catalogue import Catalogue
|
|
from bookmark.main import create_app
|
|
from bookmark.session import IntakeSession
|
|
from bookmark.sizing import MarkerRegion
|
|
|
|
|
|
def make_reference(size=100, color=200):
|
|
return np.full((size, size, 3), color, dtype=np.uint8)
|
|
|
|
|
|
def make_book_frame(size=100, ref_color=200, book_color=50, box=(20, 20, 40, 40)):
|
|
frame = np.full((size, size, 3), ref_color, dtype=np.uint8)
|
|
x, y, w, h = box
|
|
frame[y:y + h, x:x + w] = book_color
|
|
return frame
|
|
|
|
|
|
class FakeCamera:
|
|
def __init__(self, frame: np.ndarray) -> None:
|
|
self.frame = frame
|
|
|
|
def read_frame(self) -> np.ndarray:
|
|
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()
|
|
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 = FakeCamera(reference)
|
|
app = create_app(session, camera)
|
|
return TestClient(app), session, camera
|
|
|
|
|
|
def test_session_state_starts_waiting_front(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
response = client.get("/api/session-state")
|
|
assert response.status_code == 200
|
|
assert response.json()["state"] == "waiting_front"
|
|
|
|
|
|
def test_observe_without_trigger_does_not_fire(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
response = client.post("/api/observe")
|
|
assert response.status_code == 200
|
|
assert response.json()["fired"] is False
|
|
|
|
|
|
def test_manual_trigger_then_observe_captures_front(tmp_path):
|
|
client, _, camera = make_client(tmp_path)
|
|
camera.frame = make_book_frame()
|
|
client.post("/api/manual-trigger")
|
|
response = client.post("/api/observe")
|
|
body = response.json()
|
|
assert body["fired"] is True
|
|
assert body["state"] == "waiting_back"
|
|
assert body["book_id"] is not None
|
|
|
|
|
|
def test_preview_returns_404_before_any_observe(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
response = client.get("/api/preview.jpg")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_preview_returns_jpeg_after_observe(tmp_path):
|
|
client, _, _ = make_client(tmp_path)
|
|
client.post("/api/observe")
|
|
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"]
|