From 2e80447264d55b0e51e3fa74188476a37ab55d8e Mon Sep 17 00:00:00 2001 From: Olivia Casadei Date: Tue, 29 Sep 2026 13:09:02 +0200 Subject: [PATCH] test: the API on a fake 100 ms camera (parallel captures, unchanged contract) tests/fake_realsense.py: a pipeline whose wait_for_frames renders a textured 640x480 frameset after 100 ms (no RealSense on the development PC). tests/test_parallel_capture.py: - 3 parallel /api/capture take < 2x one alone (medians of 5): the criterion declared for optimize-streams R01. On the previous code: 391 vs 128 ms. - /api/capture keeps its fields; parallel requests share one frame and its bytes, frame_id still differs per response; one thread calls the pipeline. - /api/rgb, /api/depth, /api/depth/color decode; /ws streams; close() stops the acquisition thread. Skipped when pyrealsense2 is not importable, like the existing driver tests. Co-Authored-By: Claude Opus 5.5 --- tests/fake_realsense.py | 90 ++++++++++++++++++++++++++++ tests/test_parallel_capture.py | 106 +++++++++++++++++++++++++++++++++ 2 files changed, 196 insertions(+) create mode 100644 tests/fake_realsense.py create mode 100644 tests/test_parallel_capture.py diff --git a/tests/fake_realsense.py b/tests/fake_realsense.py new file mode 100644 index 0000000..83044b9 --- /dev/null +++ b/tests/fake_realsense.py @@ -0,0 +1,90 @@ +"""A fake pyrealsense2 pipeline for tests: renders one 640x480 frameset per call after a fixed latency. + +There is no RealSense on the development PC. The driver only uses a few +methods of the SDK objects, and these fakes provide exactly those. +""" + +import threading +import time + +import numpy as np + +FAKE_LATENCY_S = 0.1 # one frame every 100 ms, as the request's test asks +WIDTH, HEIGHT = 640, 480 + + +class FakeVideoFrame: + def __init__(self, data: np.ndarray): + self._data = data + + def get_data(self): + return self._data + + def get_height(self) -> int: + return self._data.shape[0] + + def get_width(self) -> int: + return self._data.shape[1] + + def __bool__(self) -> bool: + return True + + +class FakeFrameset: + def __init__(self, index: int): + rng = np.random.default_rng(index) + # Textured images, so the JPEG/PNG encoders do realistic work. + self._color = FakeVideoFrame(rng.integers(0, 255, (HEIGHT, WIDTH, 3), dtype=np.uint8)) + self._depth = FakeVideoFrame(rng.integers(300, 3000, (HEIGHT, WIDTH), dtype=np.uint16)) + + def get_color_frame(self): + return self._color + + def get_depth_frame(self): + return self._depth + + +class FakePipeline: + """wait_for_frames blocks FAKE_LATENCY_S and returns a new frameset; counts the calls.""" + + def __init__(self, latency_s: float = FAKE_LATENCY_S): + self.latency_s = latency_s + self.calls = 0 + self.max_concurrent = 0 + self._active = 0 + self._lock = threading.Lock() + self.stopped = False + + def wait_for_frames(self, timeout_ms: int = 5000): + with self._lock: + self._active += 1 + self.max_concurrent = max(self.max_concurrent, self._active) + self.calls += 1 + index = self.calls + try: + time.sleep(self.latency_s) + return FakeFrameset(index) + finally: + with self._lock: + self._active -= 1 + + def stop(self) -> None: + self.stopped = True + + +def open_fake_camera(camera, pipeline: FakePipeline) -> None: + """Put a RealSenseD435 in the 'open' state on a fake pipeline (USB 3 path, no align). + + Uses the driver's own start hook when it has one, so the same helper serves + the driver before and after the acquisition thread was introduced. + """ + import pyrealsense2 as rs + camera._pipeline = pipeline + camera._align = None + camera._color_format = rs.format.bgr8 + camera._depth_scale = 0.001 + camera._initialized = True + camera._open = True + start = getattr(camera, "_start_acquisition", None) + if start is not None: + start() diff --git a/tests/test_parallel_capture.py b/tests/test_parallel_capture.py new file mode 100644 index 0000000..cb9bca6 --- /dev/null +++ b/tests/test_parallel_capture.py @@ -0,0 +1,106 @@ +"""The API on a fake 100 ms camera: parallel requests are not serialized, the contract is unchanged.""" + +import asyncio +import statistics +import time + +import cv2 +import numpy as np +import pytest + +pytest.importorskip("pyrealsense2") + +from httpx import ASGITransport, AsyncClient # noqa: E402 +from starlette.testclient import TestClient # noqa: E402 + +from src.backend.api import routes, websocket # noqa: E402 +from src.backend.main import app # noqa: E402 +from src.backend.services.camera import RealSenseD435 # noqa: E402 +from tests.fake_realsense import FakePipeline, open_fake_camera # noqa: E402 + +# Fields of /api/capture before optimize-streams R01 (the multi-camera service parses them). +CAPTURE_FIELDS = { + "frame_id", "timestamp", "rgb_shape", "depth_shape", "depth_min_mm", "depth_max_mm", + "depth_valid_pixels", "rgb_jpeg_b64", "depth_png_b64", "depth_color_b64", +} + + +@pytest.fixture +def fake_camera(): + cam = RealSenseD435() + pipe = FakePipeline() + open_fake_camera(cam, pipe) + routes.set_camera(cam) # the lifespan (not run in tests) injects the camera into both + websocket.set_camera(cam) + yield cam, pipe + cam.close() + routes.set_camera(None) + websocket.set_camera(None) + + +async def _wall_time(client, n: int) -> tuple[float, list]: + t0 = time.perf_counter() + responses = await asyncio.gather(*(client.get("/api/capture") for _ in range(n))) + return time.perf_counter() - t0, [r.json() for r in responses] + + +def test_three_parallel_captures_take_less_than_twice_one(fake_camera): + """The criterion declared for R01: t(3 parallel) < 2 x t(1 alone), medians of 5.""" + async def run(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as client: + await _wall_time(client, 1) + one = [(await _wall_time(client, 1))[0] for _ in range(5)] + three = [(await _wall_time(client, 3))[0] for _ in range(5)] + return statistics.median(one), statistics.median(three) + + one, three = asyncio.run(run()) + assert three < 2 * one, f"1 alone {one*1000:.0f} ms, 3 parallel {three*1000:.0f} ms" + + +def test_capture_fields_unchanged_and_one_frame_shared(fake_camera): + _, pipe = fake_camera + + async def run(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as client: + return (await _wall_time(client, 3))[1] + + bodies = asyncio.run(run()) + assert all(b["success"] for b in bodies) + data = [b["data"] for b in bodies] + assert all(set(d) == CAPTURE_FIELDS for d in data) + assert len({d["frame_id"] for d in data}) == 3 # frame_id still counts responses + assert len({d["timestamp"] for d in data}) == 1 # ... all three served by one frame + assert len({d["rgb_jpeg_b64"] for d in data}) == 1 # ... with the same bytes + assert pipe.max_concurrent == 1 # only one thread calls the pipeline + + +def test_rgb_and_depth_endpoints_decode(fake_camera): + client = TestClient(app) + rgb = cv2.imdecode(np.frombuffer(client.get("/api/rgb").content, np.uint8), cv2.IMREAD_COLOR) + depth = cv2.imdecode(np.frombuffer(client.get("/api/depth").content, np.uint8), cv2.IMREAD_UNCHANGED) + color = cv2.imdecode(np.frombuffer(client.get("/api/depth/color").content, np.uint8), cv2.IMREAD_COLOR) + assert rgb.shape == (480, 640, 3) + assert depth.shape == (480, 640) and depth.dtype == np.uint16 + assert color.shape == (480, 640, 3) + + +def test_websocket_streams_frames(fake_camera): + with TestClient(app).websocket_connect("/ws") as ws: + ws.send_text("start") + first, second = ws.receive_json(), ws.receive_json() + ws.send_text("close") + assert set(first) == CAPTURE_FIELDS + assert second["timestamp"] > first["timestamp"] + + +def test_close_stops_the_acquisition_thread(): + cam = RealSenseD435() + pipe = FakePipeline() + open_fake_camera(cam, pipe) + assert cam.capture() is not None + acquisition = cam._acquisition + cam.close() + assert not cam.is_open() + assert pipe.stopped + assert not acquisition._thread.is_alive() + assert cam.capture() is None